Repository navigation
Replicate rings to other hosts over the network - #3
Conversation
Add ringfire::replication: ReplicaServer streams a ring's slot payloads to any number of mirrors over TCP; Mirror keeps a ring with the same geometry and the same sequence numbers in the local /dev/shm, so readers attach to it exactly as on the source host. Own binary protocol (16-byte frame header: HELLO, GEOMETRY, DATA, GAP, HEARTBEAT), batching, resume from the last local sequence, restart detection, busy-poll mode. CLI: ringfire serve / ringfire mirror. Add FLAG_SPARSE for rings whose sequence numbers may have holes; RingConsumer skips to the next message present instead of waiting on a hole (only on sparse rings, checked every 64 empty polls). Add Unsupported and Protocol errors, integration tests, a loopback latency example and a two-host ping-pong example.
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: cdd4d22b55
ℹ️ 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".
| let (stream, peer) = self.listener.accept()?; | ||
| let ring = SourceRing::open(&self.ring_path)?; | ||
| let (batch, spin) = (self.batch, self.spin); | ||
| thread::Builder::new() | ||
| .name(format!("ringfire-serve-{}", peer)) | ||
| .spawn(move || { | ||
| let _ = serve_client(ring, stream, batch, spin); |
There was a problem hiding this comment.
When the producer restarts and recreates the ring at the same path, create_backing_file replaces the inode, but each accepted client retains this one SourceRing mapping indefinitely. serve_client therefore keeps sending heartbeats from the orphaned mapping and never observes records written to the new ring, so already-connected mirrors silently freeze and never reconnect. Track path/inode replacement or terminate and reopen the client session when the source ring changes.
Useful? React with 👍 / 👎.
| wanted if wanted > write_seq + 1 => (write_seq + 1, GEOMETRY_RESET), | ||
| wanted => (wanted.max(oldest), 0), |
There was a problem hiding this comment.
A resumed mirror is marked as reset only when its requested sequence is ahead of the source. If a source restarts and publishes at least as far as the mirror's requested sequence before reconnection—for example, the mirror has 1–50 and the restarted source reaches 100—this branch treats the new epoch as a continuation and appends its 51–100 records to the old epoch. That silently combines unrelated streams despite the documented restart detection; the handshake needs a persistent source generation/instance identifier rather than inferring restarts solely from sequence ordering.
Useful? React with 👍 / 👎.
| let (ring, resumed) = match adopted { | ||
| Some(ring) if !reset && ring.geometry.same_layout(&geometry) => (ring, true), | ||
| other => { |
There was a problem hiding this comment.
With MirrorStart::Resume, any unlocked, structurally valid ordinary ring whose capacity/schema/offsets match the source passes this guard, because same_layout ignores its flags. The mirror then writes into that existing inode without changing its header to FLAG_SPARSE | FLAG_POLICY_LATEST_WINS; after a clamped resume or later gap, consumers of that ring can wait forever for missing sequences because they do not know the ring is sparse. Require the adopted header to have the expected mirror flags, otherwise recreate it.
Useful? React with 👍 / 👎.
ReplicaServer::multicast sends each DATA frame once as a datagram to a group; the per-mirror TCP connection carries the handshake (new MULTICAST frame) and answers NAK requests from the source ring, with GAP for unretained ranges. Mirrors hold back datagrams that overtake a hole so rings are still written in order, repeat an unanswered NAK after 20 ms, notice a lost last datagram through a 1 ms multicast heartbeat and drop datagrams from an earlier source incarnation via a session byte. CLI: serve --multicast GROUP:PORT --iface ADDR --mtu --ttl, mirror --iface ADDR. Tests cover live order, history over TCP, injected datagram loss and tail loss; examples take --multicast for loopback and two-host runs.
… sent sequences Frame linger (adaptive by default, --linger-us to fix it) fills a frame before sending it: without batching every record above ~20k/s cost a datagram and a system call and round-trip latency went from 50 us to over a millisecond. Mirror::step now processes at most 32 datagrams per call so a single-threaded caller keeps control. Multicast heartbeats carry the sender's last sent sequence instead of the ring's write_seq, which could announce a record before its datagram and make every mirror NAK it. New tests: a lapped source in multicast mode stays in order, datagrams from another session are ignored. New open-loop two-host stress example; README gains the sustained-rate table.
Fault injection sends every n-th datagram after its successor; the mirror must hold the overtaking one back and write the ring strictly by sequence.
…ze ones GEOMETRY carries the payload arena; DATA records carry the slot payload followed by the blob bytes its descriptor names. The source copies a blob and re-checks the arena's reservation counter, so a payload overwritten under it is reported as lost and becomes a GAP. The mirror writes blobs into its own arena and rewrites the descriptor's location, keeping length, flags and sequence. Blobs beyond the MTU ride IP fragmentation; beyond the host's datagram limit (EMSGSIZE) the sender skips them and mirrors fetch them over TCP. BlobConsumer learns FLAG_SPARSE hole skipping like RingConsumer. Tests: TCP mirroring, multicast with payloads up to 70 KB, arena lapping without corruption, hole skipping after a late join.
Adaptive lingering waited up to 50 us whenever the previous frame was recent, which delayed steady streams near 20k/s and every record after the first of a burst. Frames now leave at most once per 50 us unless full, and a lone record that arrives later than that goes out immediately: quiet and bursty streams keep the 55 us floor, loaded streams keep the datagram rate near 20k/s. The stress example gains --burst; the arena-lapping blob test floods the arena before the mirror connects so the lost-payload path is exercised every run.
The master stamps each record on push; consumers on the master, on same-host mirrors and on each remote mirror stamp the read. Remote slaves estimate the clock offset PTP-style (minimum-round-trip probe) and report in the master's clock. Six remote and two local slaves measured at 1k and 20k msg/s over multicast and TCP; README gains the stage table.
For routes without multicast (between sites, into clouds). A mirror announces in HELLO that it can receive UDP unicast; the source answers with its UDP port and a token in the MULTICAST frame (group 0.0.0.0) and the mirror punches to that port, so the source sends to the punched address and delivery works from behind NAT. --dup N sends every datagram N times; mirrors drop the copies by sequence. Tests cover in-order delivery with duplicates and a mirror served again as a source (a site hub). Measured Tokyo -> Los Angeles at 1000 msg/s: TCP p99 one-way 99.6 ms against a 50.4 ms median, UDP unicast p99 51.7 ms against 51.7 ms, no NAK needed; the stages example gains echo round trips measured on the master's clock and a time-bounded clock calibration.
Linux answers a close with unread receive data by a reset, so a mirror that had just sent a NAK saw ConnectionReset instead of end-of-file and reported a failure when the source shut down. Resets, aborts and broken pipes now end a mirror cleanly like EOF. The server logs a mirror it drops for any other reason. Multicast tests probe for multicast delivery first and skip on hosts without it (GitHub's macOS runners).
…nts; release 0.5.0 README gains a mirrors section with LAN, WAN and site-hub topologies, a transport matrix, CLI and Rust recipes, consolidated numbers and a tuning checklist; docs/replication.md tells the design, the protocol, what the stress test found and every measurement. Version 0.5.0, changelog dated.
1 participant
What
ringfire::replication: one source ring, byte-identical copies on other hosts.ReplicaServer(source host) streams a ring's slot payloads to any number of mirrors, one thread and one ring reader per mirror.Mirror(each other host) keeps a ring with the same geometry and the same sequence numbers in the local/dev/shm. Readers attach to it withRingConsumerexactly as on the source host and never touch the network.HELLO,GEOMETRY,DATA(up to 65,535 records per frame),GAP,HEARTBEAT,NAK,MULTICAST. Resume from the last local sequence after a reconnect (the source ring is the retransmission buffer), restart detection, busy-poll mode.TCP_NODELAY, one stream per mirror. Or UDP multicast: eachDATAframe goes out once as a datagram and every mirror receives it; the TCP connection then only carries the handshake andNAKretransmission from the source ring. Mirrors hold back datagrams that overtake a hole, so rings are still written strictly in order; a 1 ms multicast heartbeat exposes a lost last datagram; a session byte drops datagrams from an earlier source incarnation.ringfire serve <ring> --bind <addr> [--multicast GROUP:PORT --iface ADDR --mtu N --ttl N] [--spin]andringfire mirror <source> <ring> [--from latest|oldest|resume|N] [--iface ADDR] [--spin].FLAG_SPARSE(0x0400) marks rings whose sequence numbers may have holes;RingConsumerthen skips to the next message present instead of waiting on a hole. Checked every 64 empty polls and only on sparse rings, so existing rings are unchanged.UnsupportedandProtocol; rings with a payload arena are rejected for now.Measured
64-byte slots, one message every 100 µs, everything busy-polling, Linux 6.8, kernel network stack:
About 27 µs one way on the LAN either way: that is the two kernel network stacks. Multicast buys the tail and a cost that stays flat as mirrors are added.
Tests
tests/replication_tests.rs(TCP): exact byte-for-byte copy, oldest/latest start, hole skipping by readers, resume without duplicates, restart of the source, constant lapping without torn records.tests/replication_multicast_tests.rs: live order, history fetched over TCP then live by multicast, every third datagram dropped (fault injection) and recovered in order, tail loss recovered by heartbeat. Full suite and clippy-D warningspass on Linux and macOS.Next
Kernel bypass for the multicast path (
AF_XDP/ DPDK / Onload), blob (arena) rings,FLAG_SPARSEin the C header and Python readers.