Конвейеры на PUSH/PULL

Это руководство строит конвейер из руководства ZeroMQ: вентилятор отправляет задания, воркеры конкурируют за них, а приёмник собирает результаты.

В примере два сервиса. Воркер берёт задания у вентилятора и по каждому отправляет результат приёмнику. Приёмник собирает результаты пакетами. Вентилятор - любой узел, который отправляет задания, например узел на Python из руководства Начало работы.

Кто привязывается к адресу

Устойчивая сторона соединения привязывается к адресу, а сторона, которая появляется и исчезает, подключается. В конвейере устойчивы вентилятор и приёмник, поэтому привязываются они. Воркеры подключаются к обоим, и каждый новый процесс воркера - ещё одно подключение.

Сторону задают ZmqEndpoint::bind и ZmqEndpoint::connect. Сторона входит в тип брокера: брокер на connect - это ZmqQueue<Connect>. От неё зависит, что брокер может публиковать и куда может уйти копия для повтора, поэтому ошибка обнаруживается во время компиляции или при старте, а не в работе.

Воркер

Макросы

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize)]
struct Job {
    id: u64,
    items: u32,
}

/// What the sink collects: one result per job.
#[derive(Debug, PartialEq, Serialize, Outgoing)]
#[outgoing(name = "results")]
struct Done {
    id: u64,
    items: u32,
}

#[subscriber("jobs", reply)]
async fn work(job: &Job) -> Done {
    tracing::info!(job.id, job.items, "job worked");
    Done {
        id: job.id,
        items: job.items,
    }
}

Вручную

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Job {
    id: u64,
    items: u32,
}

/// What the sink collects: one result per job.
#[derive(Debug, PartialEq, Serialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Done {
    id: u64,
    items: u32,
}

// `#[outgoing(name = "results")]` by hand: the type names the subscription it is pushed to.
impl OutgoingDestination for Done {
    type Form = FixedName;
    const DESTINATION: &'static str = "results";
}

impl MessageHeaders for Done {
    type Contract = NoHeaders;
}

struct Work;

impl Handle<Job, Done> for Work {
    fn handle(
        &self,
        job: &Job,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Done, HandlerOutcome>> {
        tracing::info!(job.id, job.items, "job worked");
        ready(Ok(Done {
            id: job.id,
            items: job.items,
        }))
    }
}

work принимает задание и возвращает Done. Done сам называет свой адрес назначения, results, поэтому подписчик объявлен с аргументом reply без имени.

Макросы

src/main.rs

#[ruststream::app]
fn app() -> impl App {
    // The sink listens; every worker dials it and only pushes there.
    let sink = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5558")).bindable();
    // The retry copies come back through a queue this worker binds for itself.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5559")).bindable();
    let (results, retried_results) = (sink.bind(Publish), sink.bind(Publish));
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled(
            "ventilator",
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5557")),
            |b| {
                b.include(work)
                    .out_reply(results)
                    .out_retry(copies)
                    .to("jobs");
            },
        )
        .with_broker_labeled("retries", retries, |b| {
            b.include(work).out_reply(retried_results);
        })
        .with_broker_labeled("sink", sink, |_b| {})
}

Вручную

src/main.rs

fn app() -> impl App {
    // The sink listens; every worker dials it and only pushes there.
    let sink = ZmqQueue::new(ZmqEndpoint::connect("tcp://sink:5558")).bindable();
    // The retry copies come back through a queue this worker binds for itself.
    let retries = ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5559")).bindable();
    let (results, retried_results) = (sink.bind(Publish), sink.bind(Publish));
    let copies = retries.bind(Publish);
    RustStream::new(AppInfo::new("worker", "0.1.0"))
        .with_broker_labeled(
            "ventilator",
            ZmqQueue::new(ZmqEndpoint::connect("tcp://ventilator:5557")),
            |b| {
                b.include(subscriber("jobs", Work).reply().build())
                    .out_reply(results)
                    .out_retry(copies)
                    .to("jobs");
            },
        )
        .with_broker_labeled("retries", retries, |b| {
            b.include(subscriber("jobs", Work).reply().build())
                .out_reply(retried_results);
        })
        .with_broker_labeled("sink", sink, |_b| {})
}

fn main() -> ExitCode {
    run_main(app)
}

Брокер ventilator подключается к вентилятору, и его подписка читает задания. Воркеры, подключённые к одному вентилятору, конкурируют: PUSH отдаёт каждое задание одному воркеру по очереди.

Результаты уходят через второй брокер, sink, который подключается к приёмнику. bindable() превращает брокер в такой, к которому можно привязать политики публикации, а sink.bind(Publish) - политика, которая публикует через него. .out_reply(results) отправляет туда каждый Done. Ответить через брокер ventilator воркер не может: вентилятор только отправляет, поэтому после подключения подписки брокер отвергает любую публикацию. Эти отказы описывает страница Ответы.

Подписка, которая подключается, тоже называет, куда уходят её копии для повтора, даже если её обработчик никогда не просит повтора. Здесь они уходят в retries - очередь, которую воркер привязывает для себя, и её читает тот же обработчик. У копии имя jobs - имя подписки, которая принимает её обратно. Повторы целиком разбирает руководство Повторы и очереди dead-letter.

Приёмник

Макросы

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Debug, PartialEq, Deserialize)]
struct Done {
    id: u64,
    items: u32,
}

#[subscriber("results")]
async fn collect(results: &[Done]) -> HandlerOutcome {
    let items: u32 = results.iter().map(|done| done.items).sum();
    tracing::info!(results = results.len(), items, "results collected");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("sink", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5558")),
        |b| {
            // At most 32 results per call.
            b.include(collect.batch(nonzero!(32)));
        },
    )
}

Вручную

src/main.rs

use ruststream_zeromq::queue::prelude::*;
use serde::Deserialize;

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Done {
    id: u64,
    items: u32,
}

struct Collect;

impl Handle<[Done]> for Collect {
    fn handle(
        &self,
        results: &[Done],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        let items: u32 = results.iter().map(|done| done.items).sum();
        tracing::info!(results = results.len(), items, "results collected");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("sink", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5558")),
        |b| {
            // At most 32 results per call.
            b.include(subscriber("results", Collect).batch(nonzero!(32)).build());
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

Приёмник привязывается к адресу, к которому подключаются воркеры. Привязанный сокет PULL принимает всё, что ему отправляет любой воркер, поэтому одна подписка читает результаты всех воркеров. collect принимает срез, поэтому получает пакеты. Пакет закрывается, когда в нём набралось 32 результата или прошло 20 мс после первого результата, смотря что наступит раньше. Кадры при передаче не меняются: пакет собирается на стороне приёмника.

Один вид работы на конечную точку

PUSH и PULL не смотрят на кадр имени. Привязанная очередь передаёт своей подписке каждое отправленное ей сообщение, какое бы имя в нём ни стояло. Поэтому второму виду работы, например отчётам рядом с заданиями, нужны своя конечная точка и свой брокер. Вторую подписку на привязанной конечной точке старт отвергает, и ошибка называет обе подписки.

Порядок и ключи

PUSH раздаёт сообщения подключённым воркерам по очереди и ключ не читает. Два задания одного клиента могут выполняться на двух воркерах одновременно. Обработчик, которому нужно видеть задания одного клиента по порядку, работает в одном процессе воркера. Внутри одного процесса workers в точке монтирования выполняет несколько доставок обработчика одновременно. У сообщения ZeroMQ нет ключа, поэтому пути обработки, разбитые по ключу через by_key, получают его доставки по очереди.

Проверка конвейера

В каждом файле примера есть тест, который запускает свой сервис через TestApp::start. Тест воркера отправляет задание так, как это сделал бы вентилятор, и читает результат, опубликованный в results на брокере sink. Тест приёмника отправляет сразу три результата и проверяет, что collect получил их одним пакетом. Обвязку разбирает руководство Тестирование сервиса на ZeroMQ.