unifex.cpp

// write the data to one end of the pipe
auto writer_seq = 
    sequence(
        lazy([&]{
                printf("writes starting!\n");
            }),
        transform_done(
            repeat_effect(
                typed_via(
                    defer(
                        [&](){
                            return discard(
                                async_write_some(wPipeRef, databuffer));
                        }),
                    scheduler)),
            []{return just();}),
        lazy([&]{
                printf("writes stopped!\n");
            }));

// read the data 1 byte at a time from the other end
auto reader_seq = sequence(
    // read for some time before starting measurement
    // this is done to reduce startup effects
    pipe_bench(WARMUP_DURATION, stopWarmup), // warmup
    // reset measurements to exclude warmup
    lazy([&]{
            // restart reps and keep offset in data
            offset = reps%sizeof(data);
            reps = 0;
            printf("warmup completed!\n");
            // exclude the warmup time
            start = std::chrono::high_resolution_clock::now();
        }),
    // do more reads and measure how many reads occur
    pipe_bench(BENCHMARK_DURATION, stopRead),
    // report results
    lazy([&]{
            end = std::chrono::high_resolution_clock::now();
            printf("benchmark completed!\n");
            auto ms = std::chrono::duration_cast<std::chrono::milliseconds>(
                end - start)
                .count();
            auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(
                end - start)
                .count();
            double reads = 1000000000.0 * reps / ns;
            fmt::print("completed in {} ms, {} ns, {} ops\n", ms, ns, reps);
            fmt::print("stats - {} reads, {} ns-per-op, {} ops-per-ms\n", reads, ns/reps, reps/ms);
            stopWrite.request_stop();
        }));

sync_wait(
    with_query_value(
        when_all(writer_seq, reader_seq),
        get_stop_token,
        stopWrite.get_token()));