[carlhoerberg]

Why

Client connections are mostly idle, yet each one holds its own read and write buffer (socket_buffer_size, 16 KiB each by default). Sharing buffers across connections cuts memory use with many connections. Revives #1549 on top of current main (its BoolChannel part already landed in #1603).

Builds on #2363 (merged), which closes client sockets under the write lock. Without it, close could flush concurrently with a writer, and with pooled buffers one of them could hand the buffer back to the pool while the other still writes from it.

What

Buffer pool (src/stdlib/io_buffered.cr)

• IO::BufferPool, one per buffer size (IO::BufferPool.for(size)). Each thread has its own cache per pool: a LIFO stack, so acquire and release take no locks and reuse the most recently used buffer. Each cache holds up to 4 MiB of idle buffers (256 at the default 16 KiB). Releases beyond that go to the GC and are counted as dropped. When socket_buffer_size changes on a config reload, new connections use a pool for the new size, and the old pool keeps serving the connections opened before the reload.
• IO::Buffered#buffer_pool= is opt-in, and set only for client sockets. IOs without a pool use the stdlib implementation unchanged (previous_def). With a pool attached, buffer_size= refuses any size other than the pool's.
  • Write buffers are taken from the pool on first write and returned after each flush. A flush detaches the buffer before writing it, so a concurrent flush or close can't write or release it a second time.
  • Read buffers are returned once emptied: at end of stream, and before any read that may wait (including after read_byte/peek/skip, and direct reads into large slices).
  • Buffers are also returned when a read or write fails, including a read that fails because another fiber closed the IO, and when the write buffer is discarded on an abandoned connection.
  • Plain sockets read with Socket#read_nonblock into a pooled buffer. When no data is available, the buffer goes back to the pool and the socket waits with Socket#wait_readable, so idle sockets hold no read buffer. Unlike a blocking read, the wait doesn't serialize readers, so this assumes one reading fiber per socket, as in LavinMQ's client read loops.
  • close flushes pending data through the pool, so the write buffer is returned, and then detaches the IO from the pool.
• Socket#read_nonblock(slice) : Int32? and Socket#wait_readable (src/stdlib/socket_read_nonblock.cr) add the non-blocking read the stdlib lacks, shaped like a stdlib API. read_nonblock does recv(MSG_DONTWAIT) under the socket's fd read lock. wait_readable holds an fd reference, uses the socket's event loop, and times out with "Read timed out", like a read.
• The overrides and the socket patch rely on stdlib internals, so compiling with a Crystal version other than 1.21.x emits a warning. CI builds with --error-on-warnings, so a new Crystal minor release fails CI until they're re-verified and the version is bumped.

TLS: the TCP socket under each TLS socket now uses the pool for its read buffer (it used to hold the stdlib's default 32 KiB for the whole connection), so idle TLS connections don't hold it. The TLS socket's own read buffer is still held while idle: turning it off saved another 87 MiB at 10k idle AMQPS connections, but cost about 14% TLS throughput, because each small read becomes an SSL_read. A follow-up could avoid holding it by waiting for readability before SSL_read.

Metrics: /metrics exposes socket_buffer_pool_available, _allocated_total, _reused_total, _released_total and _dropped_total, labeled by buffer_size (documented in docs/monitoring.md). A steadily increasing dropped_total means more connections are busy at once than the cache holds.

Debug output: SIGUSR1 prints stack usage per fiber, or "running" for fibers whose saved stack pointer is stale, plus the pool stats.

Differences from #1549:

• Add buffer pool for socket buffers to reduce memory for idle connections #1549 waited for readability before reading. With the edge-triggered event loop, a burst larger than the buffer could leave unread data that never signals readable again, so the connection hung. A spec reproduces this.
• Pooling is opt-in per socket, with per-thread caches, instead of a global pool applied to any buffered IO whose size matched.

Measurements

Release builds, 4-core container, server and lavinmqperf on the same host. LavinMQ's default execution context runs client fibers on one thread, so the per-thread cache acts as a global one.

Idle connections (lavinmqperf connection-count, 10,000 connections with one channel each):

	main	this PR
AMQP: server memory in use (RSS)	562 MiB	387 MiB (−31%)
AMQP: GC heap	570 MiB	185 MiB (−68%)
AMQPS: server memory in use (RSS)	976 MiB	805 MiB (−18%)
AMQPS: GC heap	927 MiB	381 MiB (−59%)

2 pooled buffers served all 10k plain connections.

Throughput (lavinmqperf throughput, 15 s runs, average consume rate in messages per second):

Scenario	main	this PR
50 / 50 across 50 queues, 16 B (10 / 11 runs)	685k	678k (−1%, within noise)
1 publisher / 1 consumer, 16 B (3 runs)	699k	674k
50 / 50 across 50 queues, 4 KiB (3 runs)	82.3k	84.2k
TLS 1 / 1, 16 B (3 runs)	306k	302k
TLS 50 / 50, 16 B (3 runs)	292k	285k
TLS 50 / 50, 4 KiB (3 runs)	62.0k	61.5k

Run-to-run noise was ±5%. The previous shared-mutex pool was about 3% below main on 50 / 50 at 16 B.

Cache size (15 s runs, 2 per cap, buffers dropped from the cache per run):

Scenario	64 buffers	256 buffers (4 MiB)	1024 buffers
50 / 50 across 50 queues, 16 B	0, 0	0, 0	0, 0
10 publishers / 500 consumers, 16 B	205, 204	0, 0	0, 0
200 / 200 across 200 queues, 4 KiB	158, 195	0, 0	0, 0

At most about 350 buffers were in use at once. Throughput didn't differ measurably between caps.

Testing

• spec/stdlib/io_buffered_spec.cr covers:
  • per-thread caches (LIFO reuse, the 4 MiB cap and dropped, releases on another thread, pools for another size still reusing buffers);
  • reading a burst larger than the buffer;
  • no read buffer held while idle, at end of stream (also for non-socket IOs, the path TLS sockets take), or during a direct read after read_byte;
  • read and write buffers reused;
  • a close during a blocked flush neither writing nor releasing its buffer a second time (the spec fills the socket's send buffer first, so the flush blocks for certain);
  • buffers returned when a read or flush fails, when a read fails after another fiber closed the IO, when the write buffer is discarded, and when an IO is closed with buffered data;
  • buffer_size= refused while a pool is attached;
  • Socket#read_nonblock and #wait_readable, including its timeout message;
  • non-pooled IOs unchanged.
• spec/server_spec.cr: an idle TLS connection doesn't hold its TCP read buffer.
• spec/api/prometheus_spec.cr: the pool metrics are exposed for the configured buffer size.
• Each regression spec fails without its fix. make test: 2584 examples, 0 failures. make lint is clean, and the release build passes with --error-on-warnings.

[chatgpt-codex-connector]

Codex Review Summary

This comment shows the latest Codex review activity on this pull request.

Review	Status	Commit	Review trigger
📝 Code Review	✅ Completed 2026-10-08T16:01:17.532321Z	3ef201c	Manual request

ℹ️ 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" or "@codex security review".

Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings.

[claude]

No issues found.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: bd38f47cfd

ℹ️ 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".

[carlhoerberg]

Re the review above, both points are fixed in 9604be2 (pushing next):

1. Release during a concurrent close: reproduced and fixed. A writer fiber blocks in flush (peer not reading), another fiber calls close, and the buffer in use goes back to the pool while both still write from it. IO::Buffered#close now sets @buffer_pool = nil before flushing, so any flush or read still running leaves its buffer to the GC, as before pooling. A new spec in spec/stdlib/io_buffered_spec.cr fails without the fix and passes with it. Frame sends already go through @write_lock, and close was the only path where two flushes could run at once, so this also covers the double-release case.
2. TLS: the CHANGELOG now says only non-TLS connections stop holding a read buffer while idle. TLS connections still return write buffers after each flush. Releasing the read buffer while waiting would need a TLS-aware readiness check (OpenSSL's buffered data plus the underlying socket), so I've left that out of this PR.

---

Generated by Claude Code

[carlhoerberg]

CI on bd38f47 had two Spec failures:

• Spec macOS: IO::Buffered with buffer pool reads data larger than the buffer that arrived in one burst timed out. This was a bug in my spec: macOS's UNIX socket send buffer is smaller than the 10 kB it wrote before spawning the reader, so the write blocked forever. Fixed in f59ceff (pushing next), which writes from a separate fiber.
• Spec arm64: retain_store subscribing to topic with retained message does not crash failed with Cast from MQTT::Protocol::SubAck to MQTT::Protocol::Publish. I don't think this PR causes it. MQTT::Broker#subscribe queues the retained message into the session before Client#recieve_subscribe sends the SubAck, so the session's delivery fiber can send the Publish first. The spec's subscribe helper then consumes the Publish as if it were the SubAck. This PR doesn't change which fiber gets the socket first, and there is no existing fix for that ordering. The new commit triggers a fresh run, and I'll check whether it shows up again.

---

Generated by Claude Code

[carlhoerberg]

amqp-quickcheck property tests was cancelled on 3277587 after GitHub's 6-hour job limit. Every test passed within a minute except tests::publish_with_combined_properties, which never finished ("has been running for over 60 seconds", then nothing). I don't think this PR causes it. The same job hangs the same way on an unrelated branch without these changes (grant-suback-failure, run 37622703107, stuck in "Run property tests" since 12:44). The suite runs with no per-test timeout. The previous head of this PR, 5b208b3 (pooling plus the TLS change), passed the property tests. No fix exists yet; it belongs in lavinmq-quickcheck, as a per-test timeout, and in whatever makes publish_with_combined_properties wait forever. Re-running the job once.

---

Generated by Claude Code

[carlhoerberg]

Two updates, pushing next as f6b9603:

1. Review finding on new_thread_cache: confirmed. 76d4045 moved the stats registration (@caches_lock) to after the cache is installed. A fiber suspended on that lock could resume on another thread and use the first thread's cache without a lock. Fixed in 4b42927: thread_cache loops until it finds the current thread's cache in the thread local without yielding, so acquire/release always use their own thread's cache. I didn't add a spec, since it needs a fiber to move threads in the middle of the call.
2. Spec macOS on 76d4045 died with "Process terminated because of an invalid memory access" during the federation specs. This PR doesn't cause it. CI runs the merge with main, and main's spec/upstream_spec.cr (from Replace observer pattern with direct callbacks for federation and MQTT #2371) has a gated proxy that spawns its copy fibers inside the accept loop. When the proxy closes, accept? sets the shared client variable to nil, and a copy fiber that finishes after that closes a nil socket. Shovel and federation rewrite #2352 hit the same crash on macOS and fixed it in 50bd4c7. I merged main into this branch and ported that fix. When Shovel and federation rewrite #2352 lands, the change will already be in main and merge cleanly.

After merging main, make test passes (2581 examples, 0 failures), make lint is clean, and the release build passes with --error-on-warnings.

---

Generated by Claude Code

[carlhoerberg]

CI on f6b9603 is green, Spec macOS included, except amqp-quickcheck property tests. That job has been stuck in "Run property tests" since 06:40. I don't think this PR causes it:

• The queue-lifecycle branch, which has none of these changes (run 38100843139), has been stuck at the same step since 01:11.
• Quickcheck passed on this PR's previous commit, 76d4045.
• The workflow itself notes it as "Known flaky under parallel load".

The property tests have no per-test timeout, so a hang lasts until GitHub's 6-hour job limit. No fix exists yet; it would belong in lavinmq-quickcheck. I'll re-run the job once after it's cancelled.

---

Generated by Claude Code

[carlhoerberg]

Spec arm64 failed on 55ad42e in LavinMQ::Federation::Upstream shutdown closes federation links without logging unexpected-close errors. It captured connection closed unexpectedly: End of file reached from amqp.client.connection.

This failure isn't caused by this PR. It depends on spec order: it happens whenever "keeps the link of an exchange redeclared under the same name" (spec/upstream_spec.cr, from #2371) runs right before it, as this run's random order did (seed 58088).

That earlier spec closes its link while the link is still connecting, so Link#stop has no connection to close yet. The connection then completes against a server that is shutting down and gets force-closed, and its client logs the EOF during the next spec's Log.capture. Running just those two examples in that order (--location for each) fails 30 out of 30 times, on main (0b9adae) as well as on this branch. Of the specs in upstream_spec.cr, it's the only one that breaks the shutdown spec this way.

No fix exists yet. Here's a proposed patch for that spec. With it, the pair passed 30 out of 30 on main, and the whole file passed in 5 random orders:

         upstream.links.should eq [link]
+        # Closing the link closes its connection only once it has one. A link
+        # still connecting would connect to a server already shutting down,
+        # and its client would log the forced close during a later spec.
+        wait_for { link.state.running? }
       ensure
         upstream.try &.close

Every other check passed on this head. I'm re-running the failed job once.

---

Generated by Claude Code