perf(broker): ordered batched fan-out worker for non-persistent publish - #112
Merged
Conversation
XTLine
force-pushed
the
perf/non-persistent-fanout-worker
branch
from
August 21, 2026 08:51
67a95ff to
c55198e
Compare
XTLine
force-pushed
the
perf/topic-write-queue-backpressure
branch
from
August 21, 2026 08:51
2efad6d to
585e904
Compare
XTLine
force-pushed
the
perf/non-persistent-fanout-worker
branch
from
August 21, 2026 10:10
c55198e to
9e77f9e
Compare
Default options keep every SST open (max_open_files=-1) with an unbounded block cache, so accumulated data (hundreds of GB across perf runs) let anon memory grow past the 4G cgroup wall and OOM-kill the broker during high-load fanout scenarios. - max_open_files 512: resident table readers bounded - block cache LRU 256MB, index/filter blocks included - bounded memtable (64MB x2) and background jobs (4)
- Pass `cleanup=True` on restart in broker lib to clear /tmp data. - Ensure logs and timeseries are saved before storage is wiped. - Reset broker storage in a `finally` block in the stress test.
Reworks metric extraction to use the median of steady-state interval windows rather than relying solely on aggregated stats. Also adds detailed logging and exit-code explanations for E2E process failures to improve debuggability.
Adds an external broker backend to target standalone instances without lifecycle management. Also introduces local resource constraints via systemd-run and taskset, and adds a new 10GB solo-consumer backlog drain scenario.
Sample cgroup memory stats (anon/file) when broker is managed by systemd. Reset sampler between scenarios so metrics cover only the scenario window.
…ge-trigger - Unacked budget gate (max_unacked_messages=50000, aligned with Pulsar maxUnackedMessagesPerConsumer): dispatch pauses when a consumer's delivered-but-unacked set exceeds the budget, cutting the fuel of the redelivery/Flow feedback loop. - send-failure branches switch continue->break: a failed send no longer re-pops the same entry within the same pass, eliminating the read-fail-restore spin (150k entrylog reads/s of allocations). - Merge-trigger dispatch (Pulsar shouldRescheduleRead pattern): triggers arriving while a loop runs only set a flag; the pass checks it after releasing the guard and runs one more pass. No dropped triggers, no fixed time constant, empty passes never read storage. - Add Consumer::pending_acks_len accessor + unit test (shared_dispatch_pauses_when_unacked_budget_exceeded).
glibc malloc keeps freed memory in per-thread arenas and only returns the top chunk to the OS, so high-allocation-churn workloads (300k msg/s x ~5 allocs/message) pin RSS at the peak footprint. Under a 4G cgroup limit the broker OOM'd with 4.16GB anon while every data structure was empty (verified via depth instrumentation). jemalloc uses size-class buckets plus dirty-page decay: idle pages are munmapped back to the OS within ~10s, so RSS tracks live data. Same workload drops anon from 4.16GB to ~88MB and the full 10-scenario suite passes under 4G. - #[global_allocator] Jemalloc in main.rs - unprefixed_malloc_on_supported_platforms: RocksDB and other C libs allocate through jemalloc too - tikv-jemalloc-sys profiling feature enabled for MALLOC_CONF-based heap profiling (diagnostic runs only)
…e; move seek helpers into storage core
The per-Send hot path took Mutex<Storage> to clone the write-queue appender and held the topic write lock for the rate-limit check. Under 8-subscription fanout the storage mutex is saturated by dispatch reads and acks, so enqueueing producers queued behind it (P stuck at ~124k). - cache ConcurrentAppender once per connection at setup; Send path enqueues without locking storage - TopicPublishRateLimiter becomes interior-mutable (std Mutex behind &self) so validate_publish_rate runs under the topic read lock - producer publish_message rate check downgraded write->read lock
Shared-subscription acks past the mark-delete frontier were stored as one BTreeSet entry per position; 8-subscription fanout acking ~1M msg/s grew the set to gigabytes and OOM'd the broker. Store out-of-order acks as merged inclusive ranges (O(ranges) memory), mirroring Pulsar's individualDeletedMessages. - new RangeSet<K>: insert/contains/take_covering with adjacency merging - mark-delete advancement skips whole ranges in one step - persistence now encodes real ranges; decode coalesces legacy points
Layered diagnostic counters writing to /data/*_diag.txt to locate where the 16-consumer ack pipeline stalls (2100万 acks expected, storage sees <10k): - dispatch_diag: consumer send_message entry (per 100k) - ack_cmd_diag: handle_ack entry with ids/known/ack_type (per 100) - msg_acked_diag: message_acked entry with sub_type/is_persistent (per 100) - ack_drop_diag: dropped-ack points in message_acked (per 100k) - ack_enter_diag: storage ack_message/_shared entry (per 10k) - pending_acks_diag: PendingAcksMap len growth (per 100k insert) - cursor_diag: persist_state mark_delete/ranges CSV (per 100k) All probes are TEMPORARY and must be removed after root-cause.
ServerCnx::new gained a rocksdb-storage persistent_appender param in the enqueue-lock-removal change; build_test_connection was not updated, so the test profile failed to compile with rocksdb-storage.
…asts for clippy 1.98
…lock The backpressure change moved Flow handling into a spawned task, which split permit accounting in time: consumer-local permits were applied at frame processing while the dispatcher aggregate landed only when the task ran. Two races surfaced under CI timing: - key_shared: remove_consumer_with_recovery subtracts the consumer balance from the aggregate. With a refill Flow half-applied, the close zeroed the aggregate, and the flow task then dropped its own increment (contains_key guard), so the post-close redelivery dispatch exited on max_batch <= 0 and the surviving consumer never received the message. - non-persistent shared: send_messages dropped entries on a stale zero aggregate while the consumer actually held permits (the ack refill Flow from another connection had not been applied yet). Apply both permit layers synchronously under the subscription read lock (mutually exclusive with consumer removal writes), keep the dispatch trigger in its own task to preserve the event-loop starvation fix, and let per-consumer reservations instead of the aggregate fast path decide non-persistent drops. Verified locally: both previously-flaky tests now pass 30/30 runs (key_shared unacked redelivery preserves key owner, non-persist shared flow full consumer stops receiving); workspace tests and clippy 1.98 clean.
Rust keeps the OS default (Nagle on) for accepted sockets. The Pulsar protocol is request-response with small replies (Send -> SendReceipt, Ping -> Pong); with Nagle, each reply waits for the previous segment's ACK, adding ~1.4ms steady RTT. Netty defaults tcpNoDelay=true, so the Java broker never had this problem. Measured (8-sub non-persistent, 4 cores, same client): produce-only 417k -> 1000k msg/s (client cap), feed 397k -> 988k msg/s, p50 2.31ms -> 0.76ms.
The non-persistent SEND path ran fully inline in the connection task: topic.write() exclusive lock -> prepare -> dispatch_sequential().await (8 subscriptions, one lock + one dispatcher call per message) -> receipt, pipeline depth 1. Mirror the persistent write-queue shape instead: - per-topic bounded mpsc (4096) drained by one ordered worker; the connection task snapshots subscriptions under a short read lock, enqueues, and returns to reading frames (accept-ack receipt, same contract as the drop gate's fake receipts) - prepare_non_persistent_publish takes &self (the write lock was phantom: both callees only need &self) - worker drains up to 128 jobs per wakeup and groups entries per subscription: one lock + one vectorized send_non_persistent_entries per subscription per batch, amortizing wakeup and lock costs; per-subscription entry order still follows enqueue order (FIFO) - publish_message routes through the same queue with a oneshot completion reply so in-process callers and tests stay deterministic; topics built without a runtime keep the inline fallback - new test: fire-and-forget enqueue preserves per-subscription order (no-subscription publishes drain; 300-job FIFO; barrier semantics) Measured together with TCP_NODELAY fix (8-sub non-persistent, 4 cores): fanout feed 701k -> 988k msg/s, aggregate delivery 841k -> 1.18M msg/s, p99 119.5ms -> 17.5ms.
Clients send thousands of Flow commands per second under load; an INFO line each flooded the journal/PTY (10.5MB per 80s stress run) and the synchronous logger write stalled runtime workers under backpressure, observed as second-scale producer stalls. Ack handling already logs at debug.
XTLine
force-pushed
the
perf/topic-write-queue-backpressure
branch
from
August 21, 2026 10:40
80b5f17 to
3acae83
Compare
XTLine
force-pushed
the
perf/non-persistent-fanout-worker
branch
from
August 21, 2026 10:40
9e77f9e to
21238db
Compare
complete_persistent_send flushed a SendReceipt per completion, so with TCP_NODELAY every persistent message produced one small TCP segment (~200k pkt/s at max produce rate). That regressed bulk persistent produce ~29% (10GiB fill 204k -> 145k msg/s) when TCP_NODELAY landed. Feed each receipt into the write buffer instead and flush once per drained batch: the connection loop now drains up to 128 completions from conn_append_rx per wakeup, completes them all, and issues a single flush (mirroring write_message_batch on the consumer side). The bounded-drain path in handle_send and the test helper flush accordingly. Recovered from the 08-19 perf session (checkpoint b4c0b517 on the perf server); measured there at 10GiB fill 145k -> 277.9k msg/s (+36% over the pre-NODELAY baseline) with TCP_NODELAY retained. Verified locally: tests/persist 29/29, cargo test 128, clippy 1.98 clean.
XTLine
changed the base branch from
perf/topic-write-queue-backpressure
to
main
August 24, 2026 02:22
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Move non-persistent publish fan-out off the connection task into a per-topic ordered batched worker (mirroring the persistent write-queue shape), enable TCP_NODELAY on accepted sockets, and demote the per-Flow INFO log to debug. Seventh PR of the perf/topic-write-queue stack.
What Changed
prepare_non_persistent_publishtakes&self— the exclusive write lock was phantom (both callees only need&self)send_non_persistent_entriesper subscription per batch, amortizing wakeup and lock costs; per-subscription order still follows enqueue order (FIFO)publish_messageroutes through the same queue with a oneshot completion reply so in-process callers and tests stay deterministic; topics built without a runtime keep the inline fallbacktcpNoDelay=true, so the Java broker never had thisu32casts in key_shared dispatchWhy
The non-persistent SEND path ran fully inline in the connection task — one exclusive topic lock + one dispatcher call per message across 8 subscriptions, pipeline depth 1. Measured together with TCP_NODELAY (8-sub non-persistent, 4 cores): fanout feed 701k → 988k msg/s, aggregate delivery 841k → 1.18M msg/s, p99 119.5ms → 17.5ms. TCP_NODELAY alone: produce-only 417k → 1000k msg/s (client cap), p50 2.31ms → 0.76ms.
Verification
Developed and verified on the local perf stress harness during authoring (measurements above); the new unit test covers per-subscription ordering through the queue. No fresh local run was performed for this PR; CI re-runs the workspace suite here.
Notes
perf/topic-write-queue-backpressure.