Coalesce communication channel notifications

[kboniadi]

Problem & Solution Overview

Timely's communication counters currently enqueue a channel activation for every pushed message and for the final done() call. The worker later sorts and deduplicates those channel IDs, so a batch of N messages creates N + 1 notifications even though most collapse into one useful scheduling decision. This adds MPSC traffic, thread wake attempts, event-queue growth, and deduplication work under bursty or backpressured workloads.

Timely previously had done-only coalescing, but it was disabled because some producers did not reliably close batches and because waiting until done() delayed receiver pipelining. This change uses bounded coalescing instead: notify immediately for the first message, suppress intermediate notifications, and emit one trailing notification for a multi-message batch. Input flush() and send_batch() now establish explicit batch boundaries so same-epoch input remains responsive after a worker becomes idle.

Summary

• Coalesce each pusher batch from N + 1 notifications to one notification for a single-message batch or at most two for a multi-message batch.
• Preserve low latency by waking the receiver immediately on the first message.
• Preserve concurrent correctness with a trailing wake when later messages may arrive after the receiver drains the first wake.
• Make InputHandle::flush() and send_batch() close their batches; send_batch() closes exactly one combined buffered/direct batch.
• Document that Push implementations may coalesce follow-up notifications until done().
• Add intra-thread, inter-thread, concurrent-tail, and same-epoch input regression tests.

Details

• The first notification keeps producer/consumer pipelining intact. A done-only prototype roughly doubled latency in the large-batch pingpong scenario.
• A first-only design is not sufficient for inter-thread communication: the receiver can drain the first message and sleep before the rest of the batch arrives. The trailing notification closes that race.
• Empty done() calls produce no notification, and single-message batches do not need a trailing wake.
• Built-in output sessions already call done() when dropped. The input changes supply the missing boundaries identified in issue Communication: coalesce counters #243.
• Low-level callers that indefinitely ignore the documented done() boundary can still defer later messages after the first wake. Eliminating that requirement would need a more invasive receiver-acknowledged atomic state spanning allocator event delivery.
• The input change is intentionally coupled to counter coalescing. Under the new counters, closing send_batch() restores one notification per direct batch--the same useful activation behavior that per-message notification previously provided.

Performance Results

The example syntax is exchange <batch> <rounds> -w <workers>:

• batch is the number of records inserted at each logical timestamp/epoch.
• rounds is the number of timestamps/epochs executed.
• -w is the number of Timely worker threads.

For example, exchange 1 200000 -w 8 runs 200,000 one-record epochs across eight workers. This intentionally emphasizes per-batch notification and scheduling overhead rather than record-processing throughput.

Measurements used release builds on local macOS and alternated clean master and patched executions to reduce run-order bias.

Notification hot-path microbenchmark

This synthetic benchmark wraps a no-op data pusher with the inter-thread counter and drains its notification channel on another thread. It measures only sender-side notification-channel sends and wake attempts; it does not execute Timely operators, progress tracking, exchange routing, serialization, or record processing.

Metric	Clean master	Patched	Change
Notifications per 64-message batch	65	2	-96.9%
Notifications over 100,000 batches	6,500,000	200,000	-96.9%
Sender-side elapsed time	approximately 43-60 ms	approximately 5-6 ms	approximately 8-10x faster

This result establishes the direct mechanism-level saving. It is not an expected whole-dataflow speedup.

End-to-end: controlled worker scaling

The batch size and epoch count are fixed at exchange 1 200000; only the worker count changes. Each row reports the median of ten runs per revision.

Workers	Clean master median	Patched median	Change
1	0.2398 s	0.2369 s	-1.2%
2	0.6334 s	0.5685 s	-10.3%
4	1.2820 s	1.0093 s	-21.3%
8	3.7745 s	2.7881 s	-26.1%

The near-neutral one-worker result is expected: it uses intra-thread delivery and avoids the MPSC sends and cross-thread wakeups targeted by the largest part of this change. The benefit grows with worker count as those costs become a larger share of each one-record epoch.

End-to-end: larger record batch

This row reports the median of eight runs per revision and demonstrates that the change remains beneficial when useful record-processing work dominates notification overhead.

Workload	Clean master median	Patched median	Change
exchange 100000 5000 -w 2	1.8094 s	1.7709 s	-2.1%

Steady-state pingpong remained neutral. The intended benefit is lower scheduling overhead and bounded notification backlog for frequent batch boundaries, bursty senders, and lagging receivers--not a claim that every Timely workload becomes faster.

Result scope summary

Benchmark	Scope measured	Result
Notification microbenchmark	Counter event sends and wake attempts only	approximately 8-10x faster
exchange 1 200000, 1-8 workers	Complete notification-heavy Timely dataflow	1.2% to 26.1% faster
exchange 100000 5000 -w 2	Complete record-processing-heavy Timely dataflow	2.1% faster
Steady-state pingpong	Complete iterative latency workload	Neutral

Testing Done

This change added tests and was verified as follows:

Testing Done

• Local code review completed
• Full Timely workspace test suite passes
• Intra-thread and inter-thread notification batching covered
• Concurrent tail-arrival race covered with bounded timeouts
• Same-epoch send_batch() and explicit flush() reactivation covered

Validation completed:

• cargo test --workspace --quiet: 265 passed, 48 ignored, 0 failed
• Clippy passes for timely and timely_communication with warnings denied after allowing only lint classes already present on clean master under the current Rust toolchain
• cargo test -p timely_communication --test counters
• cargo test -p timely --test coalesced_input
• git diff --check
• Controlled alternating clean-master/patched release sweep for 1, 2, 4, and 8 workers
• Larger-batch alternating clean-master/patched exchange benchmark