redclaw/clawsync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Complete mmap transport: functional MmapSender/MmapReceiver + SyncPeer::Mmap

Browse Source 
MmapSender and MmapReceiver were structural stubs with only open() methods.
Now each holds its own MmapMut view of the channel file and implements full
ring-buffer send/recv using AtomicU64 Acquire/Release stores for correct
cross-process visibility.

Extracted shared ring_send/ring_recv logic used by all three types
(MmapChannel, MmapSender, MmapReceiver). Replaced plain write_u64/read_u64
header accessors with atomic_store/atomic_load in the hot path; plain
helpers retained for one-time initialisation in create().

SyncPeer gains a Mmap { send_ch, receiver } variant; PipeWriteHalf and
PipeReadHalf gain Mmap arms. Async SyncPeer::send/recv/into_pipe_halves
wrap the blocking spin-wait calls with tokio::task::block_in_place.

Added 5 new mmap tests: sender_receiver_open_same_file, roundtrip,
multiple_messages, ring_wraps, cross_thread (two-thread SPSC).

Co-Authored-By: Claude Sonnet 4.6 <[email protected]>
This commit is contained in:
osobh
2026-04-04 20:44:38 -05:00
co-authored by Claude Sonnet 4.6
parent 1c107fe58a
commit f1bdf921be
3 changed files with 315 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/clawsync-transport/src/lib.rs
+2 -2
View File
@@ -3,7 +3,7 @@
//! - [`protocol`]: `SyncMessage` wire protocol + length-prefixed framing
//! - [`tcp`]: Async TCP transport (tokio)
//! - [`quic`]: QUIC transport (quinn 0.11, TLS 1.3)
//! - [`mmap`]: Memory-mapped local transport (Phase 4 stub)
//! - [`mmap`]: Memory-mapped local transport — zero-copy same-node IPC
//! - [`error`] — [`TransportError`]
#![deny(unsafe_code)]
@@ -16,7 +16,7 @@ pub mod quic;
pub mod tcp;
pub use error::TransportError;
pub use mmap::{DEFAULT_CAPACITY, MmapChannel};
pub use mmap::{DEFAULT_CAPACITY, MmapChannel, MmapReceiver, MmapSender};
pub use peer::{PipeReadHalf, PipeWriteHalf, SyncPeer};
pub use protocol::SyncMessage;
pub use quic::{QuicConfig, QuicConnection, QuicServer, quic_connect};
crates/clawsync-transport/src/mmap.rs
+287 -134
View File
@@ -21,10 +21,28 @@
//! Messages are framed identically to the TCP transport (4-byte LE length
//! prefix) so the same `SyncMessage` serialization works unchanged.
//!
//! ## Bidirectional IPC
//!
//! Use **two** channel files for full-duplex communication between processes:
//! one file per direction. Create both with `MmapChannel::create`, then
//! split each into a `MmapSender` + `MmapReceiver` pair (or open them
//! directly). Feed the two halves into `SyncPeer::Mmap { sender, receiver }`
//! for a transport-agnostic sync session.
//!
//! ## Limitations
//! - Single producer / single consumer (no concurrent senders).
//! - Single producer / single consumer per file (no concurrent senders).
//! - Capacity is fixed at creation time.
//! - No cross-machine support (file must be on a shared filesystem).
//!
//! ## Atomicity
//!
//! `write_head` (offset 8) and `read_head` (offset 16) are updated with
//! `Acquire`/`Release` atomic stores so that concurrent readers in a
//! separate process see consistent pointer values without kernel involvement.
//! The ring-buffer body bytes are written before `write_head` is advanced,
//! ensuring the consumer never reads a partial frame.
use std::sync::atomic::{AtomicU64, Ordering};
use memmap2::MmapMut;
use std::fs::OpenOptions;
@@ -57,16 +75,20 @@ pub struct MmapChannel {
capacity: usize,
}
/// Send half of an `MmapChannel`.
/// Send half of a shared-memory channel.
///
/// Opened independently from the channel file; shares the same OS page
/// mapping. For bidirectional IPC use one `MmapSender` + one `MmapReceiver`
/// backed by **separate** files (one per direction).
pub struct MmapSender {
_path: PathBuf,
_capacity: usize,
mmap: MmapMut,
capacity: usize,
}
/// Receive half of an `MmapChannel`.
/// Receive half of a shared-memory channel.
pub struct MmapReceiver {
_path: PathBuf,
_capacity: usize,
mmap: MmapMut,
capacity: usize,
}
impl MmapChannel {
@@ -85,11 +107,11 @@ impl MmapChannel {
let mut mmap = unsafe { MmapMut::map_mut(&file).map_err(TransportError::Io)? };
// Write header
// Write header (no concurrent readers yet — plain stores are fine here)
mmap[..8].copy_from_slice(MAGIC);
write_u64(&mut mmap, 8, 0); // write_head
write_u64(&mut mmap, 16, 0); // read_head
write_u64(&mut mmap, 24, capacity as u64); // capacity
write_u64_plain(&mut mmap, 8, 0); // write_head
write_u64_plain(&mut mmap, 16, 0); // read_head
write_u64_plain(&mut mmap, 24, capacity as u64); // capacity
mmap.flush().map_err(TransportError::Io)?;
Ok(Self {
@@ -101,19 +123,7 @@ impl MmapChannel {
/// Open an existing channel file.
pub fn open(path: &Path) -> Result<Self, TransportError> {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(TransportError::Io)?;
let mmap = unsafe { MmapMut::map_mut(&file).map_err(TransportError::Io)? };
if &mmap[..8] != MAGIC {
return Err(TransportError::Protocol(
"invalid mmap channel magic".into(),
));
}
let capacity = read_u64(&mmap, 24) as usize;
let (mmap, capacity) = open_mmap(path)?;
Ok(Self {
path: path.to_owned(),
mmap,
@@ -125,148 +135,204 @@ impl MmapChannel {
///
/// Blocks (spin-waits) if there is not enough space yet.
pub fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
let frame = msg.to_frame().map_err(TransportError::Protocol)?;
let frame_len = frame.len();
if frame_len > self.capacity - 8 {
return Err(TransportError::Protocol(format!(
"message too large for mmap channel: {} bytes",
frame_len
)));
}
// Spin until space is available
loop {
let wh = read_u64(&self.mmap, 8) as usize;
let rh = read_u64(&self.mmap, 16) as usize;
let used = wh.wrapping_sub(rh);
let free = self.capacity.saturating_sub(used);
if free >= frame_len {
break;
}
std::hint::spin_loop();
}
let wh = read_u64(&self.mmap, 8) as usize;
let buf_off = HEADER_SIZE + (wh % self.capacity);
// Write frame (may wrap around)
self.write_ring(buf_off, &frame);
let new_wh = (wh + frame_len) as u64;
write_u64(&mut self.mmap, 8, new_wh);
self.mmap.flush_range(0, 32).map_err(TransportError::Io)?;
Ok(())
ring_send(&mut self.mmap, self.capacity, msg)
}
/// Receive the next message from the ring buffer.
///
/// Blocks (spin-waits) until a complete message is available.
pub fn recv(&mut self) -> Result<SyncMessage, TransportError> {
// Wait for at least 4 bytes (length prefix)
loop {
let wh = read_u64(&self.mmap, 8) as usize;
let rh = read_u64(&self.mmap, 16) as usize;
if wh.wrapping_sub(rh) >= 4 {
break;
}
std::hint::spin_loop();
}
let rh = read_u64(&self.mmap, 16) as usize;
let buf_off = HEADER_SIZE + (rh % self.capacity);
// Read 4-byte length
let mut len_bytes = [0u8; 4];
self.read_ring(buf_off, &mut len_bytes);
let msg_len = u32::from_le_bytes(len_bytes) as usize;
if msg_len > MAX_FRAME_SIZE {
return Err(TransportError::Protocol(format!(
"mmap frame too large: {msg_len}"
)));
}
// Wait for full message
loop {
let wh = read_u64(&self.mmap, 8) as usize;
let rh2 = read_u64(&self.mmap, 16) as usize;
if wh.wrapping_sub(rh2) >= 4 + msg_len {
break;
}
std::hint::spin_loop();
}
let body_off = HEADER_SIZE + ((rh + 4) % self.capacity);
let mut body = vec![0u8; msg_len];
self.read_ring(body_off, &mut body);
// Advance read head
let new_rh = (rh + 4 + msg_len) as u64;
write_u64(&mut self.mmap, 16, new_rh);
self.mmap.flush_range(0, 32).map_err(TransportError::Io)?;
SyncMessage::from_bytes(&body).map_err(TransportError::Protocol)
ring_recv(&mut self.mmap, self.capacity)
}
/// Path to the backing file.
pub fn path(&self) -> &Path {
&self.path
}
// ── Helpers ──────────────────────────────────────────────────────────────
fn write_ring(&mut self, start: usize, data: &[u8]) {
let cap = self.capacity;
let rel = start - HEADER_SIZE;
for (i, &b) in data.iter().enumerate() {
let off = HEADER_SIZE + (rel + i) % cap;
self.mmap[off] = b;
}
}
fn read_ring(&self, start: usize, buf: &mut [u8]) {
let cap = self.capacity;
let rel = start - HEADER_SIZE;
for (i, slot) in buf.iter_mut().enumerate() {
let off = HEADER_SIZE + (rel + i) % cap;
*slot = self.mmap[off];
}
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Convenience: split into send/receive halves (file-based, for multi-process)
// ─────────────────────────────────────────────────────────────────────────────
impl MmapSender {
/// Open an existing channel file for writing (producer side).
pub fn open(path: &Path) -> Result<Self, TransportError> {
let ch = MmapChannel::open(path)?;
Ok(Self {
_path: path.to_owned(),
_capacity: ch.capacity,
})
let (mmap, capacity) = open_mmap(path)?;
Ok(Self { mmap, capacity })
}
/// Send a `SyncMessage` into the ring buffer.
///
/// Blocks (spin-waits) if there is not enough space yet. For async
/// callers, wrap with `tokio::task::block_in_place`.
pub fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
ring_send(&mut self.mmap, self.capacity, msg)
}
}
impl MmapReceiver {
/// Open an existing channel file for reading (consumer side).
pub fn open(path: &Path) -> Result<Self, TransportError> {
let ch = MmapChannel::open(path)?;
Ok(Self {
_path: path.to_owned(),
_capacity: ch.capacity,
})
let (mmap, capacity) = open_mmap(path)?;
Ok(Self { mmap, capacity })
}
/// Receive the next `SyncMessage` from the ring buffer.
///
/// Blocks (spin-waits) until a complete message is available. For async
/// callers, wrap with `tokio::task::block_in_place`.
pub fn recv(&mut self) -> Result<SyncMessage, TransportError> {
ring_recv(&mut self.mmap, self.capacity)
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Shared ring-buffer logic
// ─────────────────────────────────────────────────────────────────────────────
/// Open a channel file and return `(mmap, capacity)`.
fn open_mmap(path: &Path) -> Result<(MmapMut, usize), TransportError> {
let file = OpenOptions::new()
.read(true)
.write(true)
.open(path)
.map_err(TransportError::Io)?;
let mmap = unsafe { MmapMut::map_mut(&file).map_err(TransportError::Io)? };
if &mmap[..8] != MAGIC {
return Err(TransportError::Protocol(
"invalid mmap channel magic".into(),
));
}
let capacity = read_u64_plain(&mmap, 24) as usize;
Ok((mmap, capacity))
}
fn ring_send(mmap: &mut MmapMut, capacity: usize, msg: &SyncMessage) -> Result<(), TransportError> {
let frame = msg.to_frame().map_err(TransportError::Protocol)?;
let frame_len = frame.len();
if frame_len > capacity - 8 {
return Err(TransportError::Protocol(format!(
"message too large for mmap channel: {} bytes",
frame_len
)));
}
// Spin until space is available
loop {
let wh = atomic_load(mmap, 8) as usize;
let rh = atomic_load(mmap, 16) as usize;
let used = wh.wrapping_sub(rh);
let free = capacity.saturating_sub(used);
if free >= frame_len {
break;
}
std::hint::spin_loop();
}
let wh = atomic_load(mmap, 8) as usize;
let buf_off = HEADER_SIZE + (wh % capacity);
// Write frame bytes into ring buffer, then advance write_head atomically.
// The Release store ensures the body bytes are visible before the pointer.
write_ring(mmap, buf_off, capacity, &frame);
atomic_store(mmap, 8, (wh + frame_len) as u64);
Ok(())
}
fn ring_recv(mmap: &mut MmapMut, capacity: usize) -> Result<SyncMessage, TransportError> {
// Wait for at least 4 bytes (length prefix)
loop {
let wh = atomic_load(mmap, 8) as usize;
let rh = atomic_load(mmap, 16) as usize;
if wh.wrapping_sub(rh) >= 4 {
break;
}
std::hint::spin_loop();
}
let rh = atomic_load(mmap, 16) as usize;
let buf_off = HEADER_SIZE + (rh % capacity);
// Read 4-byte length
let mut len_bytes = [0u8; 4];
read_ring(mmap, buf_off, capacity, &mut len_bytes);
let msg_len = u32::from_le_bytes(len_bytes) as usize;
if msg_len > MAX_FRAME_SIZE {
return Err(TransportError::Protocol(format!(
"mmap frame too large: {msg_len}"
)));
}
// Wait for full message
loop {
let wh = atomic_load(mmap, 8) as usize;
let rh2 = atomic_load(mmap, 16) as usize;
if wh.wrapping_sub(rh2) >= 4 + msg_len {
break;
}
std::hint::spin_loop();
}
let body_off = HEADER_SIZE + ((rh + 4) % capacity);
let mut body = vec![0u8; msg_len];
read_ring(mmap, body_off, capacity, &mut body);
// Advance read head atomically
atomic_store(mmap, 16, (rh + 4 + msg_len) as u64);
SyncMessage::from_bytes(&body).map_err(TransportError::Protocol)
}
// ─────────────────────────────────────────────────────────────────────────────
// Ring-buffer byte helpers
// ─────────────────────────────────────────────────────────────────────────────
fn write_ring(mmap: &mut MmapMut, start: usize, cap: usize, data: &[u8]) {
let rel = start - HEADER_SIZE;
for (i, &b) in data.iter().enumerate() {
let off = HEADER_SIZE + (rel + i) % cap;
mmap[off] = b;
}
}
fn read_ring(mmap: &MmapMut, start: usize, cap: usize, buf: &mut [u8]) {
let rel = start - HEADER_SIZE;
for (i, slot) in buf.iter_mut().enumerate() {
let off = HEADER_SIZE + (rel + i) % cap;
*slot = mmap[off];
}
}
// ─────────────────────────────────────────────────────────────────────────────
// Atomic header helpers
//
// `write_head` (offset 8) and `read_head` (offset 16) are accessed via
// AtomicU64 pointer casts. Both offsets are 8-byte aligned in a page-aligned
// mmap, so the cast is sound on all supported targets.
// ─────────────────────────────────────────────────────────────────────────────
fn read_u64(mmap: &[u8], off: usize) -> u64 {
/// Acquire-load a u64 from the mmap header at `off`.
fn atomic_load(mmap: &MmapMut, off: usize) -> u64 {
// SAFETY: mmap is page-aligned; off ∈ {8, 16} → 8-byte aligned; in range.
let ptr = unsafe { &*(mmap.as_ptr().add(off) as *const AtomicU64) };
ptr.load(Ordering::Acquire)
}
/// Release-store a u64 into the mmap header at `off`.
fn atomic_store(mmap: &MmapMut, off: usize, val: u64) {
// SAFETY: same as atomic_load.
let ptr = unsafe { &*(mmap.as_ptr().add(off) as *const AtomicU64) };
ptr.store(val, Ordering::Release);
}
/// Non-atomic plain read — used only during initialisation before any
/// concurrent access begins.
fn read_u64_plain(mmap: &[u8], off: usize) -> u64 {
let bytes: [u8; 8] = mmap[off..off + 8].try_into().unwrap_or([0u8; 8]);
u64::from_le_bytes(bytes)
}
fn write_u64(mmap: &mut [u8], off: usize, val: u64) {
/// Non-atomic plain write — used only during initialisation.
fn write_u64_plain(mmap: &mut [u8], off: usize, val: u64) {
mmap[off..off + 8].copy_from_slice(&val.to_le_bytes());
}
@@ -286,6 +352,8 @@ mod tests {
(f, p)
}
// ── MmapChannel (combined) ────────────────────────────────────────────────
#[test]
fn create_and_open() {
let (_f, path) = tmp_path();
@@ -376,4 +444,89 @@ mod tests {
let result = MmapChannel::open(&path);
assert!(result.is_err());
}
// ── MmapSender / MmapReceiver (split halves) ──────────────────────────────
#[test]
fn sender_receiver_open_same_file() {
let (_f, path) = tmp_path();
let _ch = MmapChannel::create(&path, DEFAULT_CAPACITY).unwrap();
let _tx = MmapSender::open(&path).unwrap();
let _rx = MmapReceiver::open(&path).unwrap();
}
#[test]
fn sender_receiver_roundtrip() {
let (_f, path) = tmp_path();
MmapChannel::create(&path, DEFAULT_CAPACITY).unwrap();
let mut tx = MmapSender::open(&path).unwrap();
let mut rx = MmapReceiver::open(&path).unwrap();
tx.send(&SyncMessage::Ack { revision: 7 }).unwrap();
let msg = rx.recv().unwrap();
assert!(matches!(msg, SyncMessage::Ack { revision: 7 }));
}
#[test]
fn sender_receiver_multiple_messages() {
let (_f, path) = tmp_path();
MmapChannel::create(&path, DEFAULT_CAPACITY).unwrap();
let mut tx = MmapSender::open(&path).unwrap();
let mut rx = MmapReceiver::open(&path).unwrap();
for i in 0u64..8 {
tx.send(&SyncMessage::Ack { revision: i }).unwrap();
}
for i in 0u64..8 {
match rx.recv().unwrap() {
SyncMessage::Ack { revision } => assert_eq!(revision, i),
other => panic!("unexpected: {other:?}"),
}
}
}
#[test]
fn sender_receiver_ring_wraps() {
let cap = 1024;
let (_f, path) = tmp_path();
MmapChannel::create(&path, cap).unwrap();
let mut tx = MmapSender::open(&path).unwrap();
let mut rx = MmapReceiver::open(&path).unwrap();
for i in 0u64..20 {
tx.send(&SyncMessage::Ack { revision: i }).unwrap();
let recv = rx.recv().unwrap();
assert!(matches!(recv, SyncMessage::Ack { revision } if revision == i));
}
}
/// Two-thread test: sender and receiver on the same channel from separate
/// threads, verifying the atomic pointer updates work correctly under
/// concurrent access.
#[test]
fn sender_receiver_cross_thread() {
let (_f, path) = tmp_path();
MmapChannel::create(&path, DEFAULT_CAPACITY).unwrap();
let path_tx = path.clone();
let tx_thread = std::thread::spawn(move || {
let mut tx = MmapSender::open(&path_tx).unwrap();
for i in 0u64..16 {
tx.send(&SyncMessage::Ack { revision: i }).unwrap();
}
});
let mut rx = MmapReceiver::open(&path).unwrap();
for i in 0u64..16 {
match rx.recv().unwrap() {
SyncMessage::Ack { revision } => assert_eq!(revision, i),
other => panic!("unexpected: {other:?}"),
}
}
tx_thread.join().unwrap();
}
}
crates/clawsync-transport/src/peer.rs
+26 -1
View File
@@ -10,10 +10,12 @@
//! - **TCP** → `OwnedReadHalf` / `OwnedWriteHalf` via `TcpConnection::into_split()`
//! - **QUIC** → two `Arc<QuicConnection>` clones (both halves share the connection
//! since QUIC `send`/`recv` are `&self`)
//! - **Mmap** → `MmapSender` / `MmapReceiver` (already independent ring buffers)
use std::sync::Arc;
use crate::error::TransportError;
use crate::mmap::{MmapReceiver, MmapSender};
use crate::protocol::SyncMessage;
use crate::quic::QuicConnection;
use crate::tcp::{TcpConnection, TcpReadHalf, TcpWriteHalf};
@@ -22,10 +24,17 @@ use crate::tcp::{TcpConnection, TcpReadHalf, TcpWriteHalf};
// SyncPeer
// ─────────────────────────────────────────────────────────────────────────────
/// A unified connection handle for a single sync session (TCP or QUIC).
/// A unified connection handle for a single sync session (TCP, QUIC, or mmap).
pub enum SyncPeer {
Tcp(TcpConnection),
Quic(Arc<QuicConnection>),
/// Local same-node transport backed by two memory-mapped ring buffers —
/// one per direction. `send_ch` is the outgoing channel; `recv_ch` is
/// the incoming channel.
Mmap {
send_ch: MmapSender,
recv_ch: MmapReceiver,
},
}
impl SyncPeer {
@@ -33,6 +42,9 @@ impl SyncPeer {
match self {
SyncPeer::Tcp(c) => c.send(msg).await,
SyncPeer::Quic(c) => c.send(msg).await,
SyncPeer::Mmap { send_ch, .. } => {
tokio::task::block_in_place(|| send_ch.send(msg))
}
}
}
@@ -40,6 +52,9 @@ impl SyncPeer {
match self {
SyncPeer::Tcp(c) => c.recv().await,
SyncPeer::Quic(c) => c.recv().await,
SyncPeer::Mmap { recv_ch, .. } => {
tokio::task::block_in_place(|| recv_ch.recv())
}
}
}
@@ -50,6 +65,7 @@ impl SyncPeer {
c.close();
Ok(())
}
SyncPeer::Mmap { .. } => Ok(()),
}
}
@@ -57,6 +73,7 @@ impl SyncPeer {
///
/// TCP: yields `OwnedReadHalf` / `OwnedWriteHalf` via `into_split()`.
/// QUIC: yields two `Arc` clones — both halves share the connection.
/// Mmap: yields the already-independent `MmapSender` / `MmapReceiver`.
pub fn into_pipe_halves(self) -> (PipeReadHalf, PipeWriteHalf) {
match self {
SyncPeer::Tcp(conn) => {
@@ -64,6 +81,9 @@ impl SyncPeer {
(PipeReadHalf::Tcp(r), PipeWriteHalf::Tcp(w))
}
SyncPeer::Quic(arc) => (PipeReadHalf::Quic(arc.clone()), PipeWriteHalf::Quic(arc)),
SyncPeer::Mmap { send_ch, recv_ch } => {
(PipeReadHalf::Mmap(recv_ch), PipeWriteHalf::Mmap(send_ch))
}
}
}
}
@@ -76,6 +96,7 @@ impl SyncPeer {
pub enum PipeWriteHalf {
Tcp(TcpWriteHalf),
Quic(Arc<QuicConnection>),
Mmap(MmapSender),
}
impl PipeWriteHalf {
@@ -83,6 +104,7 @@ impl PipeWriteHalf {
match self {
PipeWriteHalf::Tcp(h) => h.send(msg).await,
PipeWriteHalf::Quic(c) => c.send(msg).await,
PipeWriteHalf::Mmap(tx) => tokio::task::block_in_place(|| tx.send(msg)),
}
}
@@ -93,6 +115,7 @@ impl PipeWriteHalf {
c.close();
Ok(())
}
PipeWriteHalf::Mmap(_) => Ok(()),
}
}
}
@@ -101,6 +124,7 @@ impl PipeWriteHalf {
pub enum PipeReadHalf {
Tcp(TcpReadHalf),
Quic(Arc<QuicConnection>),
Mmap(MmapReceiver),
}
impl PipeReadHalf {
@@ -108,6 +132,7 @@ impl PipeReadHalf {
match self {
PipeReadHalf::Tcp(h) => h.recv().await,
PipeReadHalf::Quic(c) => c.recv().await,
PipeReadHalf::Mmap(rx) => tokio::task::block_in_place(|| rx.recv()),
}
}
}