This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 21fc4d849 feat(shard): split the inbox into consensus and client-reply
lanes (#3915)
21fc4d849 is described below
commit 21fc4d849178d85381d8d78716a1e58b8ee6e566
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Aug 20 13:54:35 2026 +0200
feat(shard): split the inbox into consensus and client-reply lanes (#3915)
---
core/configs/src/server_config/sharding.rs | 46 ++++++----
core/server/config.toml | 20 +++--
core/server/src/bootstrap.rs | 21 ++++-
core/server/src/dispatch.rs | 132 +++++++++++++++++++++++++++-
core/server/src/partition_reconciler.rs | 48 ++++++----
core/server/src/server_error.rs | 2 +
core/shard/src/builder.rs | 9 +-
core/shard/src/coordinator.rs | 12 +--
core/shard/src/lib.rs | 135 ++++++++++++++++++++---------
core/shard/src/router.rs | 42 ++++++++-
core/simulator/src/lib.rs | 29 ++++++-
core/simulator/src/replica.rs | 2 +
12 files changed, 401 insertions(+), 97 deletions(-)
diff --git a/core/configs/src/server_config/sharding.rs
b/core/configs/src/server_config/sharding.rs
index b690d45fb..9f4cead92 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -83,22 +83,18 @@ pub struct ShardingConfig {
/// `false` drops both the CPU and memory-node bindings (and logs a
/// warning, since NUMA placement without pinning is meaningless).
pub pin_cores: bool,
- /// Per-shard inter-shard inbox channel capacity. Bounded by design.
- /// Drops on full inbox of consensus frames are recovered by VSR
- /// retransmit. Drops of cross-shard client Reply frames are terminal:
- /// the client never receives the reply (no in-protocol retransmit).
- /// Both frame classes share this one channel, so a consensus burst
- /// can starve client-reply forwards: size against the worst-case sum
- /// of consensus working set + peak client-reply fan-out per shard
- /// occurring together.
- ///
- // TODO(hubcio): split into two priority lanes - one bounded queue for
- // consensus frames (drops recovered by VSR retransmit) and one for
- // client `Reply` frames (drops terminal, must be sized for worst-case
- // fan-out). Current single-channel design is the minimum-viable
- // wiring so `frame_drops_total{variant,reason}` surfaces under load
- // and yields real numbers to size the split against.
+ /// Per-shard inter-shard inbox channel capacity (the main lane:
+ /// consensus frames, connection setup, reconcile wakes). Bounded by
+ /// design; drops of consensus frames on a full inbox are recovered by
+ /// VSR retransmit. Size against the consensus working set per shard.
pub inbox_capacity: usize,
+ /// Capacity of the reply lane: the separate bounded channel carrying
+ /// cross-shard client `Reply` forwards, whose drops are terminal (no
+ /// in-protocol retransmit; the client never receives the reply). Split
+ /// from [`Self::inbox_capacity`] so a consensus burst cannot evict
+ /// reply forwards and each lane is sized for its own worst case: this
+ /// one against peak client-reply fan-out per shard.
+ pub reply_inbox_capacity: usize,
/// Wall-clock budget for a single shard's bus drain on shutdown.
/// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog
/// and the parallel-join survivor path. Sized larger than typical
@@ -145,6 +141,7 @@ impl Default for ShardingConfig {
cpu_allocation: CpuAllocation::default(),
pin_cores: SERVER_CONFIG.system.sharding.pin_cores,
inbox_capacity: SERVER_CONFIG.system.sharding.inbox_capacity as
usize,
+ reply_inbox_capacity:
SERVER_CONFIG.system.sharding.reply_inbox_capacity as usize,
shutdown_drain_timeout: SERVER_CONFIG
.system
.sharding
@@ -191,6 +188,22 @@ impl Validatable<ConfigurationError> for ShardingConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.reply_inbox_capacity == 0 {
+ eprintln!(
+ "Invalid sharding configuration: reply_inbox_capacity must be
> 0 (crossfire \
+ silently rounds 0 to 1, masking config errors)"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if self.reply_inbox_capacity > INBOX_CAPACITY_MAX {
+ eprintln!(
+ "Invalid sharding configuration: reply_inbox_capacity {}
exceeds the {} cap \
+ (each shard preallocates a channel of this size; oversizing
here OOMs the \
+ process at boot)",
+ self.reply_inbox_capacity, INBOX_CAPACITY_MAX
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
let drain = self.shutdown_drain_timeout.get_duration();
if drain.is_zero() {
@@ -385,6 +398,7 @@ mod tests {
let sharding = &config.system.sharding;
assert!(sharding.pin_cores);
assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.reply_inbox_capacity, 1024);
assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap());
@@ -403,6 +417,7 @@ mod tests {
assert!(!sharding.pin_cores);
assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.reply_inbox_capacity, 1024);
assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap());
@@ -418,6 +433,7 @@ mod tests {
assert!(sharding.pin_cores);
assert_eq!(sharding.inbox_capacity, 1024);
+ assert_eq!(sharding.reply_inbox_capacity, 1024);
assert_eq!(sharding.shutdown_drain_timeout, "10 s".parse().unwrap());
assert_eq!(sharding.shutdown_poll_interval, "50 ms".parse().unwrap());
assert_eq!(sharding.shutdown_join_timeout, "30 s".parse().unwrap());
diff --git a/core/server/config.toml b/core/server/config.toml
index d5046c509..d98098799 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -747,15 +747,21 @@ cpu_allocation = "numa:auto"
# process onto the same low-numbered cores.
pin_cores = true
-# Per-shard inter-shard inbox capacity. Bounded by design: consensus-frame
-# drops recover via VSR retransmit, but cross-shard client-reply drops are
-# terminal. Size for the worst-case sum of both: the consensus working set
-# (~ the prepare queue depth of the planes the shard hosts - [metadata] on
-# shard 0, [partition] elsewhere - times replica_count times directions) plus
-# peak client-reply fan-out per shard. Both depths are tunable, so raising
-# either raises the capacity needed here.
+# Per-shard inter-shard inbox capacity (the main lane: consensus frames,
+# connection setup, reconcile wakes). Bounded by design; consensus-frame
+# drops recover via VSR retransmit. Size for the consensus working set:
+# ~ the prepare queue depth of the planes the shard hosts ([metadata] on
+# shard 0, [partition] elsewhere) times replica_count times directions.
+# Both depths are tunable, so raising either raises the capacity needed here.
inbox_capacity = 1024
+# Capacity of the reply lane: a separate bounded channel for cross-shard
+# client-reply forwards, whose drops are terminal (no in-protocol
+# retransmit - the client never receives the reply). Split from
+# inbox_capacity so a consensus burst cannot evict reply forwards; size for
+# peak client-reply fan-out per shard.
+reply_inbox_capacity = 1024
+
# Wall-clock budget for a single shard's bus drain on shutdown. Drives
# the per-shard watchdog and the parallel-join survivor path; sized
# larger than typical TCP RTT times in-flight write-batch so writers
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index f2c5a9d11..3d983f9e6 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -685,6 +685,13 @@ fn validate_sharding_runtime_knobs(
max: INBOX_CAPACITY_MAX,
});
}
+ let reply_inbox_capacity = sharding.reply_inbox_capacity;
+ if reply_inbox_capacity == 0 || reply_inbox_capacity > INBOX_CAPACITY_MAX {
+ return Err(ServerError::InvalidReplyInboxCapacity {
+ value: reply_inbox_capacity,
+ max: INBOX_CAPACITY_MAX,
+ });
+ }
let drain_timeout = sharding.shutdown_drain_timeout.get_duration();
if drain_timeout.is_zero() || drain_timeout > SHUTDOWN_DRAIN_TIMEOUT_MAX {
return Err(ServerError::InvalidShutdownDrainTimeout {
@@ -756,9 +763,11 @@ pub fn bootstrap(
// busy-loop every shutdown watchdog on a zero poll cadence, or wedge
// process exit on an unbounded drain budget.
let inbox_capacity = config.system.sharding.inbox_capacity;
+ let reply_inbox_capacity = config.system.sharding.reply_inbox_capacity;
validate_sharding_runtime_knobs(&config.system.sharding)?;
- let (senders, mut inboxes) = shard_mesh_channels(total_shards,
inbox_capacity);
+ let (senders, mut inboxes, mut reply_inboxes) =
+ shard_mesh_channels(total_shards, inbox_capacity,
reply_inbox_capacity);
let shutdown_flag = Arc::new(AtomicBool::new(false));
let config = Arc::new(config);
// One owner table per server process, Arc-cloned into every shard's bus so
@@ -803,6 +812,9 @@ pub fn bootstrap(
let inbox = inboxes[idx]
.take()
.expect("shard_mesh_channels populates every inbox slot exactly
once");
+ let reply_inbox = reply_inboxes[idx]
+ .take()
+ .expect("shard_mesh_channels populates every reply-inbox slot
exactly once");
let senders_for_shard = senders.clone();
let config_for_shard = Arc::clone(&config);
let shutdown_flag_for_shard = Arc::clone(&shutdown_flag);
@@ -838,6 +850,7 @@ pub fn bootstrap(
assignment,
senders_for_shard,
inbox,
+ reply_inbox,
config_for_shard,
shutdown_flag_for_shard,
metadata_handoff_for_shard,
@@ -902,6 +915,7 @@ fn run_shard_thread(
assignment: ShardInfo,
senders: Vec<TaggedSender>,
inbox: ShardReceiver<ShardFrame>,
+ reply_inbox: ShardReceiver<ShardFrame>,
config: Arc<ServerConfig>,
shutdown_flag: Arc<AtomicBool>,
metadata_handoff: MetadataHandoff,
@@ -944,6 +958,7 @@ fn run_shard_thread(
replica_id,
senders,
inbox,
+ reply_inbox,
&config,
shutdown_flag,
metadata_handoff,
@@ -972,6 +987,7 @@ async fn shard_main(
replica_id: Option<u8>,
senders: Vec<TaggedSender>,
inbox: ShardReceiver<ShardFrame>,
+ reply_inbox: ShardReceiver<ShardFrame>,
config: &ServerConfig,
shutdown_flag: Arc<AtomicBool>,
metadata_handoff: MetadataHandoff,
@@ -1170,6 +1186,7 @@ async fn shard_main(
Rc::clone(&bus),
senders,
inbox,
+ reply_inbox,
shard_metrics,
Arc::clone(&metadata_view),
))
@@ -1769,6 +1786,7 @@ async fn build_shard_for_thread(
bus: Rc<IggyMessageBus>,
senders: Vec<TaggedSender>,
inbox: ShardReceiver<ShardFrame>,
+ reply_inbox: ShardReceiver<ShardFrame>,
metrics: ShardMetrics,
metadata_view: Arc<AtomicU64>,
) -> Result<(Rc<ServerShard>, Rc<RefCell<SessionManager>>), ServerError> {
@@ -2034,6 +2052,7 @@ async fn build_shard_for_thread(
partitions,
senders,
inbox,
+ reply_inbox,
shards_table,
PartitionConsensusConfig::new(
topology.cluster_id,
diff --git a/core/server/src/dispatch.rs b/core/server/src/dispatch.rs
index 429c46fd7..5dab8e2fd 100644
--- a/core/server/src/dispatch.rs
+++ b/core/server/src/dispatch.rs
@@ -4454,6 +4454,128 @@ mod tests {
);
}
+ /// A test shard wired to its own lanes (the held sender feeds them),
+ /// for the reply-lane pump tests below.
+ fn reply_lane_test_shard(name: &str) -> (SpyBus, shard::TaggedSender,
Rc<TestShard>) {
+ let bus = SpyBus::default();
+ let metadata = IggyMetadata::new(None, None, None, None,
TestMux::default(), None);
+ let partitions = IggyPartitions::new(
+ ShardId::new(0),
+ PartitionsConfig {
+ messages_required_to_save: 1,
+ size_of_messages_required_to_save:
iggy_common::IggyByteSize::from(1024_u64),
+ enforce_fsync: false,
+ validate_checksum: true,
+ segment_size: iggy_common::IggyByteSize::from(1_048_576_u64),
+ preallocate_segments: false,
+ encryptor: None,
+ path_layout: PartitionPathLayout::default(),
+ },
+ );
+ let (sender, inbox_rx, reply_inbox_rx) = shard_channel(0, 16, 16);
+ let lane_sender = sender.clone();
+ let shard = TestShard::new(
+ ShardIdentity::new(0, name.to_string()),
+ bus.clone(),
+ Rc::new(|_, _| {}),
+ Rc::new(|_, _| {}),
+ Rc::new(|_| {}),
+ Rc::new(|_| {}),
+ Rc::new(|_, _, _| {}),
+ metadata,
+ partitions,
+ vec![sender],
+ inbox_rx,
+ reply_inbox_rx,
+ PapayaShardsTable::new(),
+ PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1),
bus.clone()),
+ None,
+ ShardMetrics::for_shard(),
+ )
+ .expect("single-sender ring is canonically ordered");
+ (bus, lane_sender, Rc::new(shard))
+ }
+
+ fn reply_lane_forward(client_id: u128) -> ShardFrame {
+ ShardFrame::lifecycle(LifecycleFrame::ForwardClientSend {
+ client_id,
+ msg:
server_common::iobuf::Owned::<MESSAGE_ALIGN>::zeroed(64).into(),
+ })
+ }
+
+ /// A frame on the reply lane must reach the client through the RUNNING
+ /// pump's reply arm: the lane split moved `ForwardClientSend` off the
+ /// main inbox, so a pump that forgot to service the new lane would
+ /// strand every cross-shard reply while the send sites happily report
+ /// success.
+ #[compio::test]
+ async fn pump_live_arm_delivers_reply_lane_forwards() {
+ const TRANSPORT: u128 = 92;
+ let (bus, lane_sender, shard) =
reply_lane_test_shard("reply-lane-live-arm-test");
+
+ let (stop_tx, stop_rx) = shard::channel::<()>(1);
+ let pump_shard = Rc::clone(&shard);
+ let pump = compio::runtime::spawn(async move {
+ pump_shard.run_message_pump(stop_rx).await;
+ });
+
+ lane_sender
+ .reply_sender()
+ .try_send(reply_lane_forward(TRANSPORT))
+ .expect("reply lane has capacity");
+
+ // The pump is idle on the main lane, so its bottom reply arm must
+ // serve the frame without any main-lane traffic or shutdown drain.
+ let mut delivered = false;
+ for _ in 0..500 {
+ if !bus.client_replies.borrow().is_empty() {
+ delivered = true;
+ break;
+ }
+ compio::time::sleep(std::time::Duration::from_millis(1)).await;
+ }
+ stop_tx.try_send(()).expect("stop channel has capacity");
+ let _ = pump.await;
+
+ assert!(
+ delivered,
+ "the live reply arm must deliver a forward while the pump runs"
+ );
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies[0].0, TRANSPORT, "forward must reach its client");
+ }
+
+ /// The shutdown path must ALSO deliver reply-lane frames: a forward
+ /// already accepted by the lane when the stop signal wins the biased
+ /// select would otherwise be silently destroyed at teardown.
+ #[compio::test]
+ async fn pump_shutdown_drain_delivers_reply_lane_forwards() {
+ const TRANSPORT: u128 = 93;
+ let (bus, lane_sender, shard) =
reply_lane_test_shard("reply-lane-drain-test");
+
+ lane_sender
+ .reply_sender()
+ .try_send(reply_lane_forward(TRANSPORT))
+ .expect("reply lane has capacity");
+
+ // Pre-armed stop: the pump exits through the biased stop arm and the
+ // post-loop drain must still deliver the reply-lane frame.
+ let (stop_tx, stop_rx) = shard::channel::<()>(1);
+ stop_tx.try_send(()).expect("stop channel has capacity");
+ shard.run_message_pump(stop_rx).await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(
+ replies.len(),
+ 1,
+ "the pump's reply-lane drain must deliver the forwarded reply"
+ );
+ assert_eq!(
+ replies[0].0, TRANSPORT,
+ "the forward must reach the client it was addressed to"
+ );
+ }
+
/// A send parked for a namespace that is torn down before materialising
/// (create -> delete before the reconciler's `InsertOwned`) is discarded
/// on `ConfirmRemove`. The discard must stage the same retriable
@@ -4482,9 +4604,10 @@ mod tests {
},
);
// Real sender ring so the staged deny is observable: the test holds
- // the receiving end of this shard's own channel.
- let (sender, pump_rx) = shard_channel(0, 16);
- let (_inbox_tx, inbox_rx) = shard_channel(0, 1);
+ // the receiving ends of this shard's own lanes. The deny is a client
+ // Reply forward, so it lands on the REPLY lane.
+ let (sender, _pump_rx, reply_rx) = shard_channel(0, 16, 16);
+ let (_inbox_tx, inbox_rx, reply_inbox_rx) = shard_channel(0, 1, 1);
let shard = TestShard::new(
ShardIdentity::new(0, "discarded-parked-send-test".to_string()),
bus.clone(),
@@ -4497,6 +4620,7 @@ mod tests {
partitions,
vec![sender],
inbox_rx,
+ reply_inbox_rx,
PapayaShardsTable::new(),
PartitionConsensusConfig::new(1, ReplicaTopology::new(0, 1),
bus.clone()),
None,
@@ -4517,7 +4641,7 @@ mod tests {
shard.apply_reconcile_ops();
let mut denies = Vec::new();
- while let Ok(frame) = pump_rx.try_recv() {
+ while let Ok(frame) = reply_rx.try_recv() {
if let ShardFrame::Lifecycle(LifecycleFrame::ForwardClientSend {
client_id, msg }) =
frame
{
diff --git a/core/server/src/partition_reconciler.rs
b/core/server/src/partition_reconciler.rs
index fbf99df6b..4df3e5402 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -1544,19 +1544,30 @@ mod tests {
/// Mesh covers `0..=shard_id` since consumers index `senders[shard_id]`.
/// Peer receivers are dropped, so a misroute fails loudly instead of
landing
/// in this shard's inbox and reading as success.
+ /// Both receiving ends of a test shard's own sender-ring slot: parked
+ /// frames re-dispatch onto the main lane, staged client answers onto the
+ /// reply lane.
+ struct TestLanes {
+ main: shard::Receiver<shard::ShardFrame>,
+ reply: shard::Receiver<shard::ShardFrame>,
+ }
+
fn build_test_shard_with_inbox(
shard_id: u16,
config: &ServerConfig,
mux: TestMux,
capacity: usize,
- ) -> (Rc<TestShard>, shard::Receiver<shard::ShardFrame>) {
+ ) -> (Rc<TestShard>, TestLanes) {
let mut senders = Vec::with_capacity(usize::from(shard_id) + 1);
let mut own_rx = None;
for peer in 0..=shard_id {
- let (tx, rx) = shard::shard_channel(peer, capacity);
+ let (tx, rx, reply_rx) = shard::shard_channel(peer, capacity,
capacity);
senders.push(tx);
if peer == shard_id {
- own_rx = Some(rx);
+ own_rx = Some(TestLanes {
+ main: rx,
+ reply: reply_rx,
+ });
}
}
let mut shard = Rc::into_inner(build_test_shard(shard_id, config, mux))
@@ -1568,27 +1579,32 @@ mod tests {
)
}
- /// Drain a test shard's inbox into `(re-dispatched frames, staged client
+ /// Drain a test shard's lanes into `(re-dispatched frames, staged client
/// sends)`: served parked frames vs answers headed for a client.
- fn drain_inbox(rx: &shard::Receiver<shard::ShardFrame>) -> (usize, usize) {
+ fn drain_inbox(lanes: &TestLanes) -> (usize, usize) {
let mut served = 0;
+ while let Ok(frame) = lanes.main.try_recv() {
+ if matches!(frame, shard::ShardFrame::Consensus { .. }) {
+ served += 1;
+ }
+ }
let mut answered = 0;
- while let Ok(frame) = rx.try_recv() {
- match frame {
- shard::ShardFrame::Consensus { .. } => served += 1,
-
shard::ShardFrame::Lifecycle(shard::LifecycleFrame::ForwardClientSend {
- ..
- }) => answered += 1,
- _ => {}
+ while let Ok(frame) = lanes.reply.try_recv() {
+ if matches!(
+ frame,
+
shard::ShardFrame::Lifecycle(shard::LifecycleFrame::ForwardClientSend { .. })
+ ) {
+ answered += 1;
}
}
(served, answered)
}
- /// Count the `ForwardClientSend` frames sitting in a test shard's inbox:
the
- /// staged transient denies, which is what actually reaches a client.
- fn drain_staged_client_sends(rx: &shard::Receiver<shard::ShardFrame>) ->
usize {
- drain_inbox(rx).1
+ /// Count the `ForwardClientSend` frames sitting in a test shard's reply
+ /// lane: the staged transient denies, which is what actually reaches a
+ /// client.
+ fn drain_staged_client_sends(lanes: &TestLanes) -> usize {
+ drain_inbox(lanes).1
}
fn make_ctx(
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index 06c8ec250..34c561632 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -87,6 +87,8 @@ pub enum ServerError {
},
#[error("system.sharding.inbox_capacity must be in 1..={max}; got
{value}")]
InvalidInboxCapacity { value: usize, max: usize },
+ #[error("system.sharding.reply_inbox_capacity must be in 1..={max}; got
{value}")]
+ InvalidReplyInboxCapacity { value: usize, max: usize },
#[error("system.sharding.shutdown_drain_timeout must be in (0, {max:?}];
got {value:?}")]
InvalidShutdownDrainTimeout {
value: std::time::Duration,
diff --git a/core/shard/src/builder.rs b/core/shard/src/builder.rs
index f171784d9..07363f451 100644
--- a/core/shard/src/builder.rs
+++ b/core/shard/src/builder.rs
@@ -73,6 +73,7 @@ where
partitions: IggyPartitions<B, SB>,
senders: Vec<TaggedSender>,
inbox: Receiver<ShardFrame>,
+ reply_inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coord_config: CoordinatorConfig,
@@ -103,6 +104,7 @@ where
partitions: IggyPartitions<B, SB>,
senders: Vec<TaggedSender>,
inbox: Receiver<ShardFrame>,
+ reply_inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coord_config: CoordinatorConfig,
@@ -120,6 +122,7 @@ where
partitions,
senders,
inbox,
+ reply_inbox,
shards_table,
partition_consensus,
coord_config,
@@ -190,7 +193,10 @@ where
client_id,
msg,
});
- sender.try_send(frame).map_err(|err| {
+ // Reply lane: a forwarded client reply is terminal on
+ // drop, so it must not compete with consensus frames
+ // for main-lane capacity.
+ sender.reply_sender().try_send(frame).map_err(|err| {
metrics_for_client.record_frame_drop(
frame_drop_variant::FORWARD_CLIENT_SEND,
classify_try_send_err(&err),
@@ -227,6 +233,7 @@ where
self.partitions,
self.senders,
self.inbox,
+ self.reply_inbox,
self.shards_table,
self.partition_consensus,
coordinator,
diff --git a/core/shard/src/coordinator.rs b/core/shard/src/coordinator.rs
index f85daba69..72de53324 100644
--- a/core/shard/src/coordinator.rs
+++ b/core/shard/src/coordinator.rs
@@ -417,7 +417,7 @@ mod tests {
fn build_senders(total: u16) -> Rc<Vec<TaggedSender>> {
let mut senders = Vec::with_capacity(total as usize);
for shard_id in 0..total {
- let (tx, _rx) = crate::shard_channel(shard_id, 16);
+ let (tx, _rx, _reply_rx) = crate::shard_channel(shard_id, 16, 16);
senders.push(tx);
}
Rc::new(senders)
@@ -429,7 +429,7 @@ mod tests {
let mut senders = Vec::with_capacity(total as usize);
let mut receivers = Vec::with_capacity(total as usize);
for shard_id in 0..total {
- let (tx, rx) = crate::shard_channel(shard_id, 16);
+ let (tx, rx, _reply_rx) = crate::shard_channel(shard_id, 16, 16);
senders.push(tx);
receivers.push(rx);
}
@@ -440,10 +440,10 @@ mod tests {
fn ctor_rejects_permuted_sender_vec() {
// Build senders in correct order, then swap two entries so the
// indexed position no longer matches the tagged shard id.
- let (tx0, _rx0) = crate::shard_channel(0, 16);
- let (tx1, _rx1) = crate::shard_channel(1, 16);
- let (tx2, _rx2) = crate::shard_channel(2, 16);
- let (tx3, _rx3) = crate::shard_channel(3, 16);
+ let (tx0, _rx0, _reply_rx0) = crate::shard_channel(0, 16, 16);
+ let (tx1, _rx1, _reply_rx1) = crate::shard_channel(1, 16, 16);
+ let (tx2, _rx2, _reply_rx2) = crate::shard_channel(2, 16, 16);
+ let (tx3, _rx3, _reply_rx3) = crate::shard_channel(3, 16, 16);
let permuted = Rc::new(vec![tx0, tx2, tx1, tx3]);
let result = ShardZeroCoordinator::new(
permuted,
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index eda20d4dd..0725798a4 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -400,44 +400,60 @@ where
}
}
-/// Create a bounded inter-shard channel whose sender is tagged with the
-/// owning shard.
+/// Create the bounded inter-shard channel pair (main lane + reply lane)
+/// whose sender is tagged with the owning shard.
///
/// Bootstrap uses this to build the per-shard sender `Vec` such that
-/// `vec[i]` necessarily reaches shard `i`.
+/// `vec[i]` necessarily reaches shard `i`. The second receiver is the reply
+/// lane: cross-shard client `Reply` forwards, whose drops are terminal, ride
+/// a channel of their own so a consensus burst filling the main lane cannot
+/// evict them (see `[system.sharding] reply_inbox_capacity`).
#[must_use]
-pub fn shard_channel(owner_shard: u16, capacity: usize) -> (TaggedSender,
Receiver<ShardFrame>) {
+pub fn shard_channel(
+ owner_shard: u16,
+ capacity: usize,
+ reply_capacity: usize,
+) -> (TaggedSender, Receiver<ShardFrame>, Receiver<ShardFrame>) {
let (tx, rx) = channel::<ShardFrame>(capacity);
- (TaggedSender::new(owner_shard, tx), rx)
+ let (reply_tx, reply_rx) = channel::<ShardFrame>(reply_capacity);
+ (TaggedSender::new(owner_shard, tx, reply_tx), rx, reply_rx)
}
-/// Build canonical-ordered `(senders, inboxes)` pair for an N-shard mesh.
+/// Build canonical-ordered `(senders, inboxes, reply_inboxes)` for an
+/// N-shard mesh.
///
-/// Each `inboxes[i]` drains exclusively on the runtime owning shard `i`. The
-/// returned `senders` Vec satisfies `senders[i].shard_id() == i` by
-/// construction; clone it into every shard before spawning so all shards
-/// share the same mesh.
+/// Each `inboxes[i]` / `reply_inboxes[i]` drains exclusively on the runtime
+/// owning shard `i`. The returned `senders` Vec satisfies
+/// `senders[i].shard_id() == i` by construction; clone it into every shard
+/// before spawning so all shards share the same mesh.
///
/// Receivers are wrapped in `Option` because [`Receiver`] (crossfire
-/// `AsyncRx`) is non-cloneable on purpose; bootstrap takes the slot for
+/// `AsyncRx`) is non-cloneable on purpose; bootstrap takes the slots for
/// shard `i` exactly once when spawning the owning thread.
#[must_use]
-pub fn shard_mesh_channels(
- total_shards: u16,
- capacity: usize,
-) -> (Vec<TaggedSender>, Vec<Option<Receiver<ShardFrame>>>) {
+pub fn shard_mesh_channels(total_shards: u16, capacity: usize, reply_capacity:
usize) -> ShardMesh {
let mut senders = Vec::with_capacity(total_shards as usize);
let mut inboxes = Vec::with_capacity(total_shards as usize);
+ let mut reply_inboxes = Vec::with_capacity(total_shards as usize);
for shard_id in 0..total_shards {
- let (tx, rx) = shard_channel(shard_id, capacity);
+ let (tx, rx, reply_rx) = shard_channel(shard_id, capacity,
reply_capacity);
senders.push(tx);
inboxes.push(Some(rx));
+ reply_inboxes.push(Some(reply_rx));
}
- (senders, inboxes)
+ (senders, inboxes, reply_inboxes)
}
-/// A [`Sender`] annotated with the id of the shard whose paired receiver it
-/// feeds.
+/// The canonical N-shard mesh: lane senders plus the per-shard receivers
+/// (`inboxes[i]` / `reply_inboxes[i]` drain on the runtime owning shard `i`).
+pub type ShardMesh = (
+ Vec<TaggedSender>,
+ Vec<Option<Receiver<ShardFrame>>>,
+ Vec<Option<Receiver<ShardFrame>>>,
+);
+
+/// The pair of lane [`Sender`]s annotated with the id of the shard whose
+/// paired receivers they feed.
///
/// Inter-shard routing indexes `senders[i]` with `i == target_shard`. The
/// plain `Sender` form has no way to verify that invariant at runtime, so a
@@ -446,25 +462,41 @@ pub fn shard_mesh_channels(
/// [`TaggedSender::new`]) at the channel-creation site; the coordinator and
/// [`IggyShard`] ctors then validate `senders[i].shard_id() == i`,
/// returning [`ShardCtorError`] if violated.
+///
+/// `Deref` targets the main lane; [`Self::reply_sender`] exposes the reply
+/// lane (cross-shard client `Reply` forwards, terminal on drop).
pub struct TaggedSender {
shard_id: u16,
inner: Sender<ShardFrame>,
+ reply: Sender<ShardFrame>,
}
impl TaggedSender {
- /// Wrap an already-constructed sender with the id of the shard whose
- /// paired receiver drains it. Prefer [`shard_channel`] unless an
- /// existing sender is being re-tagged (e.g., tests that build senders
- /// manually and know the ordering is correct).
+ /// Wrap already-constructed lane senders with the id of the shard whose
+ /// paired receivers drain them. Prefer [`shard_channel`] unless existing
+ /// senders are being re-tagged (e.g., tests that build senders manually
+ /// and know the ordering is correct).
#[must_use]
- pub const fn new(shard_id: u16, inner: Sender<ShardFrame>) -> Self {
- Self { shard_id, inner }
+ pub const fn new(shard_id: u16, inner: Sender<ShardFrame>, reply:
Sender<ShardFrame>) -> Self {
+ Self {
+ shard_id,
+ inner,
+ reply,
+ }
}
#[must_use]
pub const fn shard_id(&self) -> u16 {
self.shard_id
}
+
+ /// The reply lane's sender. Client `Reply` forwards go here so the main
+ /// lane's consensus traffic cannot evict them; everything else stays on
+ /// the main lane via `Deref`.
+ #[must_use]
+ pub const fn reply_sender(&self) -> &Sender<ShardFrame> {
+ &self.reply
+ }
}
impl Clone for TaggedSender {
@@ -472,6 +504,7 @@ impl Clone for TaggedSender {
Self {
shard_id: self.shard_id,
inner: self.inner.clone(),
+ reply: self.reply.clone(),
}
}
}
@@ -552,17 +585,19 @@ fn forward_nonce_seed<B: MessageBus>(consensus:
Option<&VsrConsensus<B>>) -> u64
/// Lifecycle frame variants.
///
/// Connection setup and cross-shard forwards: every frame the inter-shard
-/// channel carries that is NOT a consensus protocol message lives here.
+/// channels carry that is NOT a consensus protocol message lives here.
/// Splitting these out from [`ShardFrame::Consensus`] keeps the consensus
-/// dispatch path hot and cache-tight while leaving lifecycle traffic on
-/// the same single channel (preserving relative ordering between consensus
-/// and lifecycle frames at near-zero cost).
+/// dispatch path hot and cache-tight.
///
-/// Trade-off: consensus and lifecycle traffic compete for one bounded
-/// inbox. A consensus burst or retransmit storm can fill it exactly when
-/// a terminal-drop [`LifecycleFrame::ForwardClientSend`] needs the space;
-/// `inbox_capacity` is a single knob and cannot isolate the two frame
-/// classes.
+/// Lane placement: every variant rides the main inbox EXCEPT
+/// [`LifecycleFrame::ForwardClientSend`], which rides the dedicated reply
+/// lane (`reply_inbox_capacity`) because its drops are terminal while every
+/// main-lane variant's loss is recovered by some retry (VSR retransmit,
+/// reconnect sweep, periodic tick). The two lanes are independent queues:
+/// there is NO relative ordering between a consensus frame and a client
+/// reply forward, which is safe because a reply forward never
+/// order-couples with consensus traffic (it requires a served request,
+/// and per-client reply order is preserved within the reply lane itself).
#[non_exhaustive]
pub enum LifecycleFrame {
/// Shard 0 distributes an inbound replica TCP connection fd to the
@@ -1296,6 +1331,12 @@ where
/// messages here via the corresponding sender.
inbox: Receiver<ShardFrame>,
+ /// Receiver end of this shard's reply lane: cross-shard client `Reply`
+ /// forwards, split off the main inbox because their drops are terminal
+ /// (no in-protocol retransmit) while a consensus burst can legitimately
+ /// fill the main lane. Fed via [`TaggedSender::reply_sender`].
+ reply_inbox: Receiver<ShardFrame>,
+
/// Partition namespace -> owning shard lookup.
shards_table: T,
@@ -1444,6 +1485,15 @@ where
self.inbox.len()
}
+ /// [`Self::inbox_len`] for the reply lane, so the simulator's lost-wakeup
+ /// tripwire covers both queues: a frame stranded in either lane at
+ /// quiescence is a missed wake.
+ #[cfg(any(test, feature = "simulator"))]
+ #[must_use]
+ pub fn reply_inbox_len(&self) -> usize {
+ self.reply_inbox.len()
+ }
+
/// Create a new shard with channel links and a shards table.
///
/// * `bus` - shard-local bus handle (kept alongside the buses owned
@@ -1454,6 +1504,8 @@ where
/// construction time so every sender carries the id of the shard
/// whose receiver drains it.
/// * `inbox` - the receiver that this shard drains in its message pump.
+ /// * `reply_inbox` - the reply lane's receiver, drained by the same
+ /// pump (client `Reply` forwards only; see
[`TaggedSender::reply_sender`]).
/// * `shards_table` - namespace -> shard routing table.
/// * `coordinator` - `Some` on shard 0 (supplied by the builder when
/// `is_shard_zero`), `None` everywhere else. Immutable post-ctor:
@@ -1483,6 +1535,7 @@ where
partitions: IggyPartitions<B, SB>,
senders: Vec<TaggedSender>,
inbox: Receiver<ShardFrame>,
+ reply_inbox: Receiver<ShardFrame>,
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
coordinator: Option<Rc<crate::coordinator::ShardZeroCoordinator>>,
@@ -1509,6 +1562,7 @@ where
senders,
shard_count,
inbox,
+ reply_inbox,
shards_table,
partition_consensus,
coordinator,
@@ -1808,12 +1862,14 @@ where
shards_table: T,
partition_consensus: PartitionConsensusConfig<B>,
) -> Self {
- // Placeholder inbox: the simulator delivers frames straight to
+ // Placeholder lanes: the simulator delivers frames straight to
// `on_message` (see the `shard_count` note below), so nothing ever
- // sends here and capacity 1 exists only to satisfy the field. The
- // real inbox is bounded on purpose (`inbox_capacity` is the shard's
- // backpressure), so no unbounded variant is wanted here either.
+ // sends here and capacity 1 exists only to satisfy the fields. The
+ // real lanes are bounded on purpose (`inbox_capacity` /
+ // `reply_inbox_capacity` are the shard's backpressure), so no
+ // unbounded variant is wanted here either.
let (_tx, inbox) = channel(1);
+ let (_reply_tx, reply_inbox) = channel(1);
let nonce_seed = forward_nonce_seed(metadata.consensus.as_ref());
let plane = MuxPlane::new(variadic!(metadata, partitions));
let ShardIdentity { id, name } = identity;
@@ -1837,6 +1893,7 @@ where
// through, while preserving single-shard routing semantics.
shard_count: 1,
inbox,
+ reply_inbox,
shards_table,
partition_consensus,
metrics: crate::metrics::ShardMetrics::for_shard(),
@@ -3285,7 +3342,7 @@ where
// Count only what was actually handed to the pump: crediting before
the
// send reports an answer to a client that never received one, which is
// the opposite of what this counter is read for.
- if let Err(error) = sender.try_send(frame) {
+ if let Err(error) = sender.reply_sender().try_send(frame) {
self.metrics.record_frame_drop(
crate::metrics::frame_drop_variant::PARTITION,
crate::coordinator::classify_try_send_err(&error),
diff --git a/core/shard/src/router.rs b/core/shard/src/router.rs
index 0432ecc22..4340761a0 100644
--- a/core/shard/src/router.rs
+++ b/core/shard/src/router.rs
@@ -261,9 +261,10 @@ where
_ = stop.recv().fuse() => break,
() = consensus_tick.as_mut() => {
// Sharing the pump task is what keeps `tick_partitions`
- // borrow-safe, but it bounds the tick's worst-case delay
to
- // one frame body's longest `.await` (replication append +
- // commit_journal fsync/rotate + reply).
+ // borrow-safe, but it bounds the tick's worst-case delay
+ // to one main frame body's longest `.await` (replication
+ // append + commit_journal fsync/rotate + reply) plus the
+ // one reply-lane bus send drained per main frame.
// TODO(hubcio): if a load test shows tick starvation,
// make `tick_partitions` borrow-free so the tick can be
// decoupled from the pump again without reintroducing the
@@ -303,6 +304,32 @@ where
// Tail drain catches reconcile ops whose
marker was dropped.
self.apply_reconcile_ops();
}
+ // Guaranteed reply-lane service: `select_biased!`
+ // polls the main lane first, so a saturated main
+ // lane would otherwise starve the reply arm below
+ // indefinitely. Taking at most ONE reply per main
+ // frame makes the worst case a deterministic 1:1
+ // interleave - replies keep flowing under a
+ // consensus storm, and consensus keeps flowing
+ // under a reply flood.
+ if let Ok(reply) = self.reply_inbox.try_recv()
+ && self.accept_frame_for_self(&reply)
+ {
+ self.process_frame(reply).await;
+ }
+ }
+ Err(_) => break,
+ }
+ }
+ frame = self.reply_inbox.recv().fuse() => {
+ // Reached only while the main lane is quiet (biased
order):
+ // serve forwarded client replies without waiting for the
+ // next main frame or tick.
+ match frame {
+ Ok(frame) => {
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ }
}
Err(_) => break,
}
@@ -310,7 +337,9 @@ where
}
}
- // Drain remaining frames so in-flight requests get a response.
+ // Drain remaining frames so in-flight requests get a response, and
+ // the reply lane so already-forwarded replies still reach their
+ // clients before the bus tears down.
while let Ok(frame) = self.inbox.try_recv() {
if self.accept_frame_for_self(&frame) {
self.process_frame(frame).await;
@@ -319,6 +348,11 @@ where
self.apply_reconcile_ops();
}
}
+ while let Ok(frame) = self.reply_inbox.try_recv() {
+ if self.accept_frame_for_self(&frame) {
+ self.process_frame(frame).await;
+ }
+ }
// Final flush: committed messages still resident in the in-memory
// journal must reach segment storage before the process exits, or a
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 17f24a620..1662ea6a1 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -270,8 +270,11 @@ impl Simulator {
// One crossfire mesh per replica; every shard gets a clone of
// the canonical senders vec and exclusively takes its inbox.
- let (senders, mut inboxes) =
- shard::shard_mesh_channels(shards_per_replica,
SIM_INBOX_CAPACITY);
+ let (senders, mut inboxes, mut reply_inboxes) =
shard::shard_mesh_channels(
+ shards_per_replica,
+ SIM_INBOX_CAPACITY,
+ SIM_INBOX_CAPACITY,
+ );
let mut shards =
Vec::with_capacity(usize::from(shards_per_replica));
let mut stop_txs =
Vec::with_capacity(usize::from(shards_per_replica));
@@ -286,6 +289,9 @@ impl Simulator {
let inbox = inboxes[usize::from(shard_idx)]
.take()
.expect("mesh yields exactly one inbox per shard");
+ let reply_inbox = reply_inboxes[usize::from(shard_idx)]
+ .take()
+ .expect("mesh yields exactly one reply inbox per shard");
// Only shard 0 owns metadata consensus, so only it carries the
// superblock. Peer shards persist nothing.
let shard_superblock = if shard_idx == 0 {
@@ -302,6 +308,7 @@ impl Simulator {
rc,
senders.clone(),
inbox,
+ reply_inbox,
consensus_clock.clone(),
shell,
metadata_bundle.clone(),
@@ -609,6 +616,16 @@ impl Simulator {
self.seed,
self.executor.schedule_hash(),
);
+ let pending_replies = shard.reply_inbox_len();
+ assert_eq!(
+ pending_replies,
+ 0,
+ "lost wakeup: replica {replica_id} shard {} reply lane
holds \
+ {pending_replies} frame(s) at quiescence (seed {:#x},
schedule hash {:#x})",
+ shard.id,
+ self.seed,
+ self.executor.schedule_hash(),
+ );
}
}
}
@@ -755,8 +772,8 @@ impl Simulator {
let consensus_clock =
ConsensusClock::new(Rc::new(SimClock::new(self.executor.timer())));
let outbox = Rc::clone(&self.outboxes[idx]);
- let (senders, mut inboxes) =
- shard::shard_mesh_channels(shards_per_replica, SIM_INBOX_CAPACITY);
+ let (senders, mut inboxes, mut reply_inboxes) =
+ shard::shard_mesh_channels(shards_per_replica, SIM_INBOX_CAPACITY,
SIM_INBOX_CAPACITY);
let mut shards = Vec::with_capacity(usize::from(shards_per_replica));
let mut stop_txs = Vec::with_capacity(usize::from(shards_per_replica));
@@ -766,6 +783,9 @@ impl Simulator {
let inbox = inboxes[usize::from(shard_idx)]
.take()
.expect("mesh yields exactly one inbox per shard");
+ let reply_inbox = reply_inboxes[usize::from(shard_idx)]
+ .take()
+ .expect("mesh yields exactly one reply inbox per shard");
let shard_superblock = if shard_idx == 0 {
Some(superblock.clone())
} else {
@@ -780,6 +800,7 @@ impl Simulator {
self.replica_count,
senders.clone(),
inbox,
+ reply_inbox,
consensus_clock.clone(),
self.shell,
metadata_bundle.clone(),
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 362f3a144..cbcffc671 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -118,6 +118,7 @@ pub fn new_shard(
replica_count: u8,
senders: Vec<shard::TaggedSender>,
inbox: shard::Receiver<shard::ShardFrame>,
+ reply_inbox: shard::Receiver<shard::ShardFrame>,
clock: ConsensusClock,
shell: bool,
reader_bundle: Option<SimMetadataBundle>,
@@ -339,6 +340,7 @@ pub fn new_shard(
partitions,
senders,
inbox,
+ reply_inbox,
PapayaShardsTable::new(),
shard::PartitionConsensusConfig::with_clock(
CLUSTER_ID,
