[worker] Change background appender channel to be memory-bound instead of record count bound

[MohamedBassem]

Context:

Before this PR, the background appender's channel was a bounded channel of size 50. This has two caveats:

1. Large commands (e.g. large schema upserts) can inflate the size of this channel in the worst case to 1.5GB (per partition). A coordinated event (e.g. schema change) can end accumulating 10s of GBs of unaccounted memory across all partitions causing the node to OOM.
2. On the other hands, if the append latency is high, and the records are tiny (as shown later in the benchmarks), this 50 records arbitrary limits hinders the batching efficiency quite a bit.

Solution:

The main problem is that the number of records is just hard to configure correctly across different workloads. We want reasonable batching efficiency with bounded memory growth. As such, this PR switches the background appender channel to be memory-bound instead of record-count bound. The memory bound is configurable and will default to 64MB (i.e. allowing the background appender to do up to two full batches per run, though we might want to tune how large we want a single batch to be).

Implementation wise:

1. We're switching the underlying channel to be unbounded channel with a MemoryPool on top.
2. Because we don't know how much to reserve from the memory pool beforehand, we're going to allow overcomitting the memory pool by at most once enqueueXX call.
   a. To simplify reasoning about the overcomitting, I'm dropping the Clone support for the LogSender. Otherwise, we might overcomitt by more than one enqueue calls across concurrent senders. We don't currently need it to be Clone, so this was a no-op.
3. Once the memory pool is fully exhusted, the background appender is going to start rejecting future appends. As such, the LeaderState stops polling the effects stream once the pool is exhusted, and waits for it to become available again. Because leader state has an exclusive reference over the self proposer (and its underlying sender), it can be sure that if it sees that the pool has capacity, it'll be able to send without getting any reservation.
4. As a nice side effect for all the self proposer APIs becoming sync, we no longer need to worry about cancellation safety, so the events are handled inline inside LeaderState::run.

Benchmarks (Steady State):

Started a local one node cluster comparing base against this commit. The workfload is a 30s of Counter::get on a virtual object coming from 1000 concurrent connections.

# Base

❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080                                                                                                                                                                                                                                 ✘ 130 main
thread 1784573163 started
Running 30s test @ http://localhost:8080
  1 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    56.97ms   89.58ms   1.36s    98.26%
    Req/Sec    21.23k     2.93k   29.85k    76.59%
  Latency Distribution
     50%   45.84ms
     75%   53.83ms
     90%   62.58ms
     99%  581.58ms
  631995 requests in 30.04s, 168.16MB read
Requests/sec:  21036.54
Transfer/sec:      5.60MB
thread 1784573163 made 632995 requests and got 631995 responses

# This commit

❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080                                                      main
thread 1784573695 started
Running 30s test @ http://localhost:8080
  1 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    55.14ms   88.14ms   1.17s    98.28%
    Req/Sec    22.02k     2.79k   31.03k    76.92%
  Latency Distribution
     50%   44.09ms
     75%   51.81ms
     90%   60.31ms
     99%  574.92ms
  655652 requests in 30.07s, 174.45MB read
Requests/sec:  21805.57
Transfer/sec:      5.80MB
thread 1784573695 made 656652 requests and got 655652 responses

So there's no degradation perf wise for this whole branch.

Benchmarks (higher append latency):

Where this becomes interesting is when append latency is higher because the 50recs bounded channel gets filled faster, and starts pushing back specially with tiny records, and due to the lack of pipelining, the throughput of the system collapses. So doing the same benchmarks but with a 200ms sleep in the background appender:

# Base

❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080                                                      main
thread 1784574220 started
Running 30s test @ http://localhost:8080
  1 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency     1.04s   308.07ms   2.00s    68.70%
    Req/Sec     0.87k   592.52     2.87k    65.53%
  Latency Distribution
     50%  916.07ms
     75%    1.14s
     90%    1.52s
     99%    1.97s
  25328 requests in 30.08s, 6.74MB read
  Socket errors: connect 0, read 0, write 0, timeout 1437
Requests/sec:    842.15
Transfer/sec:    229.45KB
thread 1784574220 made 26329 requests and got 25328 responses

vs:

# This commit

❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080                                                      main
thread 1784573956 started
Running 30s test @ http://localhost:8080
  1 threads and 1000 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   590.43ms  136.85ms   1.85s    81.02%
    Req/Sec     1.72k   679.51     3.67k    71.75%
  Latency Distribution
     50%  605.69ms
     75%  611.69ms
     90%  617.10ms
     99%    1.44s
  50304 requests in 30.00s, 13.38MB read
Requests/sec:   1676.56
Transfer/sec:    456.80KB
thread 1784573956 made 51305 requests and got 50304 responses

This commit was able to have twice the throughput of the base branch under high latency.

---

Stack created with Sapling. Best reviewed with ReviewStack.

• -> [worker] Change background appender channel to be memory-bound instead of record count bound #5076
• [MemoryPool] Add support for overdrafts in MemoryPool #5057

[github-actions]

Test Results

  8 files  ±0    8 suites  ±0   5m 3s ⏱️ -2s
 61 tests ±0   61 ✅ ±0  0 💤 ±0  0 ❌ ±0 
268 runs  ±0  268 ✅ ±0  0 💤 ±0  0 ❌ ±0 

Results for commit 4df8927. ± Comparison against base commit 63c65e8.

♻️ This comment has been updated with latest results.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 7d37d7f613

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[tillrohrmann]

Thanks a lot for making the background appender channel memory bound @MohamedBassem. The changes look really great.

The one question I had was whether stopping to batch leader events will have a noticeable effect on the overall tail latencies as processing journal events from the invoker needs now more partition processor loop iterations, for example. The benchmarks you've showed, didn't indicate this.

[tillrohrmann]

Thanks for addressing my comments @MohamedBassem. LGTM. +1 for merging :-)