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

[frankmcsherry]

Thanks for the PR! It prompted some investigation about the code path, and the conclusion was that this change introduces a consequential regression, but there is a similar change. #816 has it as one of the commits, but it is to notify on each data transmission, but not at the end of the feed when done() is called.

The reasoning is that in a concurrent system, it's important to signal when there is work to be done, for reasons of performance not only liveness. Signaling only at done() would make the system live, but other workers would be awoken later than they could have productively started work. In this PR, the first wake alerts them to the first message, which is good in ping-pong situations, but regresses in the case that there is a sequence of consequential messages: the remote workers are woken for the first of the messages, but then sleep until the chain closes. If the work piling up is meaningful, the worker could productively run in this time, overlapping computation and getting "pipeline parallelism".

I had Claude do a mock up of this, and with one worker sending another eight messages, varying the time required to retire each of them, the time taken looks like so:

  ┌─────────────────────────────────────┬───────────────────┬──────────────────┐
  │                rule                 │ process = 1200 µs │ process = 400 µs │
  ├─────────────────────────────────────┼───────────────────┼──────────────────┤
  │ per message (#816)                  │ 13.3              │ 12.5             │
  ├─────────────────────────────────────┼───────────────────┼──────────────────┤
  │ first message, then trailing (#815) │ 20.5              │ 14.9             │
  ├─────────────────────────────────────┼───────────────────┼──────────────────┤
  │ done() only                         │ 21.7              │ 15.3             │
  └─────────────────────────────────────┴───────────────────┴──────────────────┘

#816 should also clean up a bunch of noise in the measurements, as busy-waiting for a few microseconds is an important clean-up: otherwise the various notifications stand in for busy waiting, and removing notifications slows things down only because the workers are more often found asleep, but not because they are awake at the right moments to do work.