ruststream 0.7.0

// The harness macros generate the group module, its items and the paths between them, and a
// benchmark function takes its setup value by value because the harness owns the drop; the
// crate's lints are written for the library surface, not for generated benchmark scaffolding.
#![allow(
    missing_docs,
    unused_qualifications,
    unreachable_pub,
    clippy::must_use_candidate,
    clippy::needless_pass_by_value
)]
//! Wall time per message over the in-process transport, for the framework's own code alone.
//!
//! Informational, never a gate. It shows what an instruction count cannot - caches, branch
//! prediction, the scheduler - and it is noisy enough on a busy machine that a percent of
//! difference means nothing. The numbers that gate a change are the ones the other benchmarks
//! here produce.

mod common;

use std::hint::black_box;

use common::{Latch, MESSAGES, Order, Ready, Started};
use divan::Bencher;
use divan::counter::ItemsCount;
use ruststream::memory::prelude::*;
use serde::Serialize;

fn main() {
    divan::main();
}

#[derive(Debug, Serialize, Outgoing)]
#[outgoing(name = "confirmations")]
struct Confirmation {
    id: u64,
}

#[subscriber("orders")]
async fn consume(order: &Order, ctx: &mut Context<'_, (), Latch>) -> HandlerOutcome {
    black_box((order.id, order.quantity));
    ctx.state().arrived();
    HandlerOutcome::ack()
}

#[subscriber("orders", workers(4))]
async fn pooled(order: &Order, ctx: &mut Context<'_, (), Latch>) -> HandlerOutcome {
    black_box((order.id, order.quantity));
    ctx.state().arrived();
    HandlerOutcome::ack()
}

#[subscriber("orders", threads(4))]
async fn threaded(order: &Order, ctx: &mut Context<'_, (), Latch>) -> HandlerOutcome {
    black_box((order.id, order.quantity));
    ctx.state().arrived();
    HandlerOutcome::ack()
}

#[subscriber("orders", threads(1))]
async fn threaded_one(order: &Order, ctx: &mut Context<'_, (), Latch>) -> HandlerOutcome {
    black_box((order.id, order.quantity));
    ctx.state().arrived();
    HandlerOutcome::ack()
}

#[subscriber("orders", reply)]
async fn confirm(order: &Order, ctx: &mut Context<'_, (), Latch>) -> Confirmation {
    ctx.state().arrived();
    Confirmation {
        id: black_box(order.id),
    }
}

// The wall-clock harness times the closure it is handed, so the service is started and the queue
// filled here, outside the timing, rather than inside it as the instruction counts do it.
fn consuming() -> Ready {
    common::pending(MESSAGES, 0, |b| {
        b.include(consume);
    })
    .ready()
}

fn replying() -> Ready {
    common::pending(MESSAGES, 0, |b| {
        b.include(confirm);
    })
    .ready()
}

// The same consume on a runtime with worker threads, publishing inside the timing: what a
// delivery costs when the thread that produced it and the thread that handles it are different
// ones. The instruction counts cannot see this - an atomic read-modify-write is one instruction
// to callgrind whether the cache line is its own core's or another's.
fn consuming_across_threads() -> Started {
    common::pending_on(common::worker_runtime(), MESSAGES, 0, |b| {
        b.include(consume);
    })
    .started()
}

// The pool of four on a runtime with worker threads: the workers are tasks of that runtime, and
// every delivery is handed from the loop to one of them, wherever each happens to run.
fn pooled_across_threads() -> Started {
    common::pending_on(common::worker_runtime(), MESSAGES, 0, |b| {
        b.include(pooled);
    })
    .started()
}

// Four dedicated threads beside a runtime with worker threads: the loop runs on the runtime and
// hands every delivery to a thread's ring.
fn threaded_across_threads() -> Started {
    common::pending_on(common::worker_runtime(), MESSAGES, 0, |b| {
        b.include(threaded);
    })
    .started()
}

// One dedicated thread: with a handler this light the thread drains its ring faster than the loop
// fills it, so what this shows is the cost of a thread that keeps running out of work.
fn threaded_one_across_threads() -> Started {
    common::pending_on(common::worker_runtime(), MESSAGES, 0, |b| {
        b.include(threaded_one);
    })
    .started()
}

#[divan::bench]
fn consume_json(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(consuming)
        .bench_local_values(|ready| ready.runtime.block_on(ready.latch.drained()));
}

#[divan::bench]
fn reply(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(replying)
        .bench_local_values(|ready| ready.runtime.block_on(ready.latch.drained()));
}

#[divan::bench]
fn consume_json_across_threads(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(consuming_across_threads)
        .bench_local_values(|started| started.publish_and_drain());
}

#[divan::bench]
fn pool_across_threads(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(pooled_across_threads)
        .bench_local_values(|started| started.publish_and_drain());
}

#[divan::bench]
fn threads_across_threads(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(threaded_across_threads)
        .bench_local_values(|started| started.publish_and_drain());
}

#[divan::bench]
fn one_thread_across_threads(bencher: Bencher) {
    bencher
        .counter(ItemsCount::new(MESSAGES))
        .with_inputs(threaded_one_across_threads)
        .bench_local_values(|started| started.publish_and_drain());
}