hubcio commented on code in PR #4506:
URL: https://github.com/apache/iggy/pull/4506#discussion_r4261449475
##########
core/server/src/dispatch/mod.rs:
##########
@@ -557,25 +688,48 @@ async fn handle_client_request<B, MJ, S, SB>(
// (`submit_register_local_or_forward`). Every other non-replicated
// code MUST go through Register first, which binds the acting user
// the per-op authz gates resolve.
- handle_non_replicated_request(
- shard,
- sessions,
- server_config,
- transport_client_id,
- request,
- ConnectionContext {
- bound,
- user_id,
- address: client_address,
- metadata_watermark,
- },
- )
- .await;
+ let context = ConnectionContext {
+ bound,
+ user_id,
+ address: client_address,
+ metadata_watermark,
+ };
+ // `ClientDrain` says which frames stay on the drain.
+ if matches!(non_replicated_code(header), PING_CODE |
BIND_SESSION_CODE) {
+ handle_non_replicated_request(
+ shard,
+ sessions,
+ server_config,
+ transport_client_id,
+ request,
+ context,
+ )
+ .await;
+ return;
+ }
+ let in_flight = drain.admit(Admission::Cap).await;
+ let task_shard = Rc::clone(shard);
+ let task_sessions = Rc::clone(sessions);
+ let task_config = Arc::clone(server_config);
+ shard.bus.spawn(async move {
Review Comment:
The consumer-progress ordering bug is fixed in 5a2e85e54, with a disk-tier
regression that fails when the guard is removed. Inline read promotion and the
throughput regression are deferred to multi-leader PR6. Any fast path must
preserve metadata barriers, per-consumer reply ordering and PING progress;
rerun the standalone and concurrent-produce benchmarks before accepting it.
##########
core/server/src/dispatch/mod.rs:
##########
@@ -240,13 +249,130 @@ impl Drop for ActiveDrainSlot {
}
}
+/// The state that the drain task of one connection shares with the requests
+/// it spawned.
+///
+/// The drain takes the frames of the connection in arrival order and runs the
+/// prologue of each. It finishes a frame that later frames depend on before it
+/// takes the next one: a login, a logout, a `BIND_SESSION`, the rewrite of a
+/// metadata write, or a partition admission. A `PING` finishes there too,
+/// because it never waits. Every other read, every poll and every wait for a
+/// metadata commit runs in a task of its own, so a held frame does not stall
+/// the frames behind it.
+///
+/// `capacity` caps those tasks at `message_bus.client_queue_capacity`, the
+/// outbound queue size of a connection, so their replies never overflow that
+/// queue. A metadata write, a segment delete, a login and a logout also take
+/// the metadata lane, which holds one of them at a time. The metadata client
+/// table keeps one pending write per client and refuses a lower request id
+/// after a higher one committed, so a second write in flight would fail with
+/// `TransientNotAccepted` or `RequestTooOld`. The lane also runs the rewrite
of
+/// a write after the previous write replied, because the rewrite reads state
+/// that the previous write can change.
+///
+/// At the cap or the lane the drain waits, and the frames behind it queue up
+/// to [`MAX_QUEUED_CLIENT_REQUESTS`]. While tasks run, the drain outlives its
+/// queue, so the cap and the lane still count them when the next frame
+/// arrives.
+pub(in crate::dispatch) struct ClientDrain {
+ capacity: usize,
+ in_flight: Cell<usize>,
+ /// Whether a frame holds the metadata lane.
+ metadata_write: Cell<bool>,
+ /// The drain task, while it waits for a frame or for room to admit one.
+ parked: Cell<Option<Waker>>,
+}
+
+/// What a frame waits for before [`ClientDrain::admit`] counts it.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Admission {
+ /// Room under the in-flight cap.
+ Cap,
+ /// Room under the cap and a free metadata lane.
+ MetadataLane,
+}
+
+impl ClientDrain {
+ pub(in crate::dispatch) const fn new(server_config: &ServerConfig) -> Self
{
+ Self {
+ capacity: server_config.message_bus.client_queue_capacity,
Review Comment:
Deferred to multi-leader PR6. PR4 retains the approved client_queue_capacity
bound. A separate read cap needs multi-connection fairness and throughput
measurements, plus saturation and cancellation tests, before choosing its value.
##########
core/server/src/http/handlers.rs:
##########
@@ -1314,12 +1313,28 @@ pub(in crate::http) async fn poll_messages(
.streams()
.poll_metadata(namespace, None, HTTP_READ_CLIENT_ID)
.ok_or(ReadError::Rejected(IggyError::TransientNotAccepted))?;
- let context =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
- if let Some(requested) = query.strategy.context {
- // TCP parity: a group poll already got the re-sync sentinel at resolve
- // above, so this poll has no group and names group 0, as TCP does.
- fence_poll_context(requested, context, 0,
partition_id).map_err(ReadError::Rejected)?;
+ let local =
metadata.context(state.shard.plane.metadata().applied_frontier().get());
+ let requested = request
+ .route()
+ .map(|route| route.context)
+ .or(query.strategy.context)
+ .unwrap_or(local);
+ let effect = if query.auto_commit {
+ PartitionEffect::Write
+ } else {
+ PartitionEffect::Read
+ };
+ if let Some(response) = SendWrapper::new(route_partition(
Review Comment:
Deferred to multi-leader PR6. PR4 now gates the actual HTTP read on the
partition primary. Local-first dispatch changes the approved forwarding flow
and must prove non-admission before rerouting, preserve the original deadline,
and keep session-gate timeout distinct from a routing refusal; compare HTTP
produce/read benchmarks with those guarantees intact.
##########
core/server/src/http/forward.rs:
##########
@@ -241,19 +361,108 @@ pub(in crate::http) async fn forward_to_primary(
SendWrapper::new(forward_or_pass(state, request, next)).await
}
-/// Route-layer fallback for acknowledged partition writes over HTTP.
-///
-/// HTTP has no persistent leader-aware connection to retarget. Execute on the
-/// contacted node first, then retry across the configured HTTP nodes within
-/// the deadline after a typed `TransientNotAccepted` denial. That
-/// denial proves the write never entered a partition pipeline. Every ambiguous
-/// outcome is returned without replay.
-pub(in crate::http) async fn forward_partition_write(
+/// Route layer of the partition routes: buffer the request once and attach it
+/// as a [`PartitionRequest`], so its handler can relay it to the partition
+/// primary with [`route_partition`]. No attempt runs here.
+pub(in crate::http) async fn forward_partition_request(
State(state): State<HttpState>,
request: Request,
next: Next,
) -> Response {
- SendWrapper::new(forward_partition_or_pass(state, request, next)).await
+ SendWrapper::new(attach_partition_request(state, request, next)).await
+}
+
+/// Decide where a partition request runs once its partition and context are
+/// fixed. `None` serves it on this node. `Some` is the final answer: the reply
+/// of the partition primary, or a refusal that proves nothing was admitted.
+///
+/// Until the deadline, every hop waits for a view with a primary, because a
+/// new partition starts on each replica at its own pace. A hop target never
+/// relays: it refuses with `NotPrimary` when another replica is the primary,
+/// so a stale view never chains hops. The contacted node starts another turn
+/// only after such a proven non-admission, a failed connect, or a read that
+/// failed after its send.
+pub(in crate::http) async fn route_partition(
+ state: &HttpInner,
+ request: &PartitionRequest,
+ namespace: IggyNamespace,
+ context: PartitionContext,
+ effect: PartitionEffect,
+) -> Option<Response> {
+ if request.hop == PartitionHop::Unrouted {
+ return None;
+ }
+ let Ok(partition_id) = u32::try_from(namespace.partition_id()) else {
+ return
Some(CustomError::from(IggyError::InvalidCommand).into_response());
+ };
+ let mut guard = None;
+ loop {
+ if let Some(primary) = partition_primary(state, namespace).await {
Review Comment:
Deferred to multi-leader PR6: carry the refusing replica's view/primary hint
and test lagging ingress views, stale hints and repeated refusals under one
bounded deadline.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -690,7 +688,7 @@ pub const fn fence_poll_context(
}
if requested.owner_generation != local.owner_generation {
return Err(IggyError::ConsumerGroupPartitionNotOwned(
- group_id,
+ client_id as u32,
Review Comment:
Deferred to multi-leader PR6 diagnostics. This needs one consistent error
identity contract that distinguishes the full client ID from the group ID, with
tests for both reporting paths.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1674,28 +1559,25 @@ impl TcpClient {
Option<iggy_binary_protocol::consensus::RequestHeader>,
Result<Bytes, IggyError>,
) {
- let stream = self.stream.clone();
- let consensus_session = self.consensus_session.clone();
- let metadata_watermark =
Arc::clone(&self.poll_router.metadata_watermark);
+ let link = self.link.clone();
+ let session_owner = Arc::clone(&self.session_owner);
let mut outcome_so_far = std::mem::take(retry_outcome);
- // SAFETY: we run code holding the `stream` lock in a task so we can't
be cancelled while holding the lock.
+ // A task finishes the exchange when its caller goes away. An unbound
+ // request that left before its reply would close the link.
let joined = tokio::spawn(async move {
- let mut stream_guard = stream.lock().await;
- let Some(stream) = stream_guard.as_mut() else {
+ let mut link_guard = link.lock().await;
Review Comment:
Deferred to multi-leader PR5 with Rust driver activation. Move exchange
ownership there while preserving request identity, mutation lanes, cancellation
and bounded in-flight work. PR6 will compare transport throughput and
task/context-switch overhead after activation.
##########
core/sdk/src/tcp/tcp_client.rs:
##########
@@ -1707,115 +1589,65 @@ impl TcpClient {
Err(error) => return (preencoded.header, Err(error),
outcome_so_far),
}
};
- let mut frame_complete = false;
- let outcome = async {
- loop {
- let header_bytes = bytemuck::bytes_of(&request_header);
- frame_complete = false;
- stream.write(header_bytes).await?;
- if !payload.is_empty() {
- stream.write(&payload).await?;
+ let mut link_failed = false;
+ let mut replay = false;
+ let outcome = loop {
+ let frame = Frame {
+ header: request_header,
+ body: payload.clone(),
+ };
+ trace!("Sent a TCP request with code: {code}, waiting for a
response...");
+ // The deadline guards against server-side reply loss (e.g. a
+ // stalled replication quorum that never commits the op), which
+ // would hold the link lock forever and wedge every later
+ // request on this client.
+ let reply = match link.send(frame, read_deadline, &mut
false).await {
+ Ok(reply) => reply,
+ Err(error) => {
+ link_failed = true;
+ break Err(outcome_so_far.observe_link_error(error,
replay));
}
- stream.flush().await?;
- trace!("Sent a TCP request with code: {code}, waiting for
a response...");
-
- let mut response_header = [0u8;
iggy_binary_protocol::HEADER_SIZE];
- // `stream.read` delegates to `read_exact`; on success it
- // always returns the requested length, so no short-read
- // guard is needed here.
- //
- // Deadline guards against server-side reply loss (e.g. a
- // stalled replication quorum that never commits the op):
- // the connection is lockstep, so an unanswered read would
- // hold the stream lock forever and wedge every later
- // request on this client. On expiry drop the stream --
- // a late reply would desync framing for the next request.
- //
- // One deadline spans BOTH the header and body reads: a
- // reply that delivers a header then stalls must not get a
- // fresh full timeout for the body.
- let header_read =
- tokio::time::timeout_at(read_deadline,
stream.read(&mut response_header))
- .await;
- let Ok(header_read) = header_read else {
- error!(
- "Timed out after {RESPONSE_READ_TIMEOUT:?} waiting
for VSR response header for TCP request with code: {code}",
- );
- return Err(IggyError::Disconnected);
- };
- header_read.map_err(|error| {
- error!(
- "Failed to read VSR response header for TCP
request with code: {code}: {error}",
- );
- IggyError::Disconnected
- })?;
-
- let response_size =
crate::vsr::response_size(&response_header)?;
- let body_size = response_size -
iggy_binary_protocol::HEADER_SIZE;
- let body = if body_size > 0 {
- let mut body = BytesMut::with_capacity(body_size);
- let body_read = tokio::time::timeout_at(
- read_deadline,
- stream.read_buf(&mut body, body_size),
- )
- .await;
- let Ok(body_read) = body_read else {
- error!(
- "Timed out after {RESPONSE_READ_TIMEOUT:?}
waiting for VSR response body for TCP request with code: {code}",
- );
- return Err(IggyError::Disconnected);
- };
- body_read.map_err(|error| {
- error!(
- "Failed to read VSR response body for TCP
request with code: {code}: {error}",
- );
- IggyError::Disconnected
- })?;
- body.freeze()
- } else {
- Bytes::new()
- };
-
- frame_complete = true;
- crate::vsr::observe_metadata_reply(&metadata_watermark,
&response_header);
- match crate::vsr::decode_response_split(&response_header,
body)
- .map_err(|error| outcome_so_far.observe(error))
+ };
+ replay = true;
+ session_owner.observe_metadata_reply(&reply.header);
+ match crate::vsr::decode_response_split(&reply.header,
reply.body)
+ .map_err(|error| outcome_so_far.observe(error))
+ {
+ Err(error) if !retry_transient => break Err(error),
+ // `TransientNotCommitted`: the op's outcome is unknown
+ // (e.g. a view change canceled it in flight) -- ONLY a
+ // same-session replay of the same request id is safe
+ // (the client-table serves it from cache if it did
+ // commit). Replay on this connection for the whole
+ // request budget; never hand it to the failover path,
+ // which re-issues under a fresh session and could
+ // double-apply a committed write.
+ Err(IggyError::TransientNotCommitted)
+ if code !=
iggy_binary_protocol::codes::POLL_MESSAGES_ON_PRIMARY_CODE
+ && tokio::time::Instant::now() < read_deadline =>
{
- Err(error) if !retry_transient => return Err(error),
- // `TransientNotCommitted`: the op's outcome is unknown
- // (e.g. a view change canceled it in flight) -- ONLY a
- // same-session replay of the same request id is safe
- // (the client-table serves it from cache if it did
- // commit). Replay on this connection for the whole
- // request budget; never hand it to the failover path,
- // which re-issues under a fresh session and could
- // double-apply a committed write.
- Err(IggyError::TransientNotCommitted)
- if code !=
iggy_binary_protocol::codes::POLL_MESSAGES_ON_PRIMARY_CODE
- && tokio::time::Instant::now() < read_deadline
=>
- {
- let remaining = read_deadline
-
.saturating_duration_since(tokio::time::Instant::now());
-
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
- }
- // `TransientNotAccepted`: the server never admitted
the
- // request, so it is re-issuable anywhere. Replay here
- // briefly, then hand it back to the caller for a
- // leader recheck / failover.
- Err(IggyError::TransientNotAccepted)
- if tokio::time::Instant::now() <
transient_deadline =>
- {
- let remaining = transient_deadline
-
.saturating_duration_since(tokio::time::Instant::now());
-
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
- }
- other => return other,
+ let remaining =
+
read_deadline.saturating_duration_since(tokio::time::Instant::now());
+
tokio::time::sleep(NOT_READY_RETRY_INTERVAL.min(remaining)).await;
Review Comment:
Deferred to multi-leader PR5 with Rust driver activation. Release the link
guard during retry waits while retaining the exact request identity and
mutation-lane ownership; verify concurrent PING, replay and cancellation across
TCP, WebSocket and QUIC.
##########
core/sdk/src/links/mux.rs:
##########
@@ -0,0 +1,815 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! One TCP or WebSocket connection that carries many requests at once.
+//!
+//! A writer task and a reader task own the two halves of the connection, and
+//! each reply goes to the request with its id. So a caller that goes away
+//! never leaves a frame half written or half read on the connection.
+
+use crate::links::{Frame, Reply};
+use crate::vsr::{decode_eviction, peek_command, reply_request_id,
response_size};
+use bytes::{BufMut, BytesMut};
+use futures_util::stream::{SplitSink, SplitStream};
+use futures_util::{SinkExt, StreamExt};
+use iggy_binary_protocol::consensus::{Command, HEADER_SIZE};
+use iggy_common::IggyError;
+use std::collections::HashMap;
+use std::collections::hash_map::Entry;
+use std::future::Future;
+use std::io;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex as StdMutex, MutexGuard, PoisonError};
+use std::time::Duration;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader,
BufWriter};
+use tokio::sync::{Mutex, Semaphore, mpsc, oneshot, watch};
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, timeout, timeout_at};
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::Message;
+use tokio_tungstenite::tungstenite::protocol::CloseFrame;
+use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode;
+use tracing::{debug, error, trace};
+
+/// Frames queued for the writer task before a sender waits for room.
+const LINK_WRITE_QUEUE: usize = 32;
+/// Requests that wait for a reply on one link at once. It matches
+/// `message_bus.client_queue_capacity`, the most requests the server runs at
+/// once for one connection.
+const MAX_PENDING_PER_LINK: usize = 256;
+/// A peer that stops reading cannot hold up a shutdown for longer than this.
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
+const WEBSOCKET_CLOSE_REASON: &str = "Client requested shutdown";
+
+type Waiter = oneshot::Sender<Result<Reply, IggyError>>;
+
+/// A connection that carries many requests at once. Dropping the link stops
+/// its tasks, which closes the connection.
+#[derive(Debug)]
+pub(crate) struct MuxLink {
+ shared: Arc<MuxShared>,
+ queue: mpsc::Sender<Frame>,
+ tasks: LinkTasks,
+}
+
+#[derive(Debug)]
+struct MuxShared {
+ peer: SocketAddr,
+ pending: StdMutex<Pending>,
+ permits: Semaphore,
+ /// Before bind, request ids repeat, so one such exchange runs at a time.
+ unbound: Mutex<()>,
+ /// Turns true once, when the link closes.
+ closed: watch::Sender<bool>,
+}
+
+#[derive(Debug)]
+struct Pending {
+ /// Cleared when the link closes. A closed link never opens again.
+ open: bool,
+ waiters: HashMap<u64, Waiter>,
+}
+
+#[derive(Debug)]
+struct LinkTasks {
+ writer: JoinHandle<()>,
+ reader: JoinHandle<()>,
+}
+
+impl Drop for LinkTasks {
+ fn drop(&mut self) {
+ self.writer.abort();
+ self.reader.abort();
+ }
+}
+
+/// A request that waits for its reply. Dropping it ends the wait.
+struct PendingEntry<'a> {
+ shared: &'a MuxShared,
+ request: u64,
+ /// Set for a repeated id that reached the queue: its late reply would
+ /// answer the next request with the same id.
+ close_if_abandoned: bool,
+}
+
+impl MuxLink {
+ /// A link over the two halves of a TCP connection, plain or TLS.
+ pub(crate) fn tcp<R, W>(reader: R, writer: W, peer: SocketAddr) -> Self
+ where
+ R: AsyncRead + Unpin + Send + 'static,
+ W: AsyncWrite + Unpin + Send + 'static,
+ {
+ Self::start(
+ peer,
+ |frames, shared| write_tcp(BufWriter::new(writer), frames, shared),
+ |shared| read_tcp(BufReader::new(reader), shared),
+ )
+ }
+
+ /// A link over a WebSocket connection, plain or TLS.
+ pub(crate) fn websocket<S>(stream: WebSocketStream<S>, peer: SocketAddr)
-> Self
+ where
+ S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
+ {
+ let (sink, stream) = stream.split();
+ Self::start(
+ peer,
+ |frames, shared| write_websocket(sink, frames, shared),
+ |shared| read_websocket(stream, shared),
+ )
+ }
+
+ /// Sends `frame` and waits for its reply until `deadline`. `sent` turns
+ /// true when the frame reaches the queue, so a caller that goes away
+ /// before the reply still knows that the frame left.
+ ///
+ /// `NotConnected` means the frame never reached the queue, so the request
+ /// can go again. `Disconnected` or an eviction reason means it did, and no
+ /// reply came.
+ pub(crate) async fn send(
+ &self,
+ frame: Frame,
+ deadline: Instant,
+ sent: &mut bool,
+ ) -> Result<Reply, IggyError> {
+ let request = frame.header.request;
+ let unbound = frame.header.session == 0;
+ let _unbound_turn = if unbound {
+ let turn = timeout_at(deadline, self.shared.unbound.lock()).await;
+ Some(turn.map_err(|_| IggyError::NotConnected)?)
+ } else {
+ None
+ };
+ let _permit = timeout_at(deadline, self.shared.permits.acquire())
+ .await
+ .map_err(|_| IggyError::NotConnected)?
+ .map_err(|_| IggyError::NotConnected)?;
+ let (waiter, reply) = oneshot::channel();
+ let mut entry = self.shared.wait_for(request, waiter)?;
+ if !matches!(
+ timeout_at(deadline, self.queue.send(frame)).await,
+ Ok(Ok(()))
+ ) {
+ return Err(IggyError::NotConnected);
+ }
+ *sent = true;
+ entry.close_if_abandoned = unbound;
+ match timeout_at(deadline, reply).await {
+ Ok(Ok(result)) => result,
+ Ok(Err(_)) => Err(IggyError::Disconnected),
+ Err(_) => {
+ error!(
+ "No reply to request {request} from {} before its
deadline.",
+ self.shared.peer
+ );
+ Err(IggyError::Disconnected)
+ }
+ }
+ }
+
+ /// Fails every waiting request. The writer task then ends the connection
+ /// the way its protocol expects.
+ pub(crate) fn close(&self) {
+ self.shared.close(&IggyError::Disconnected);
+ }
+
+ /// Returns when the link closes, for any reason.
+ pub(crate) async fn closed(&self) {
+ let mut closed = self.shared.closed.subscribe();
+ // The sender lives as long as the link.
+ let _ = closed.wait_for(|closed| *closed).await;
+ }
+
+ /// Ends the connection the way its protocol expects: a FIN for TCP, a
+ /// Close message for WebSocket. The link counts as closed first, so the
+ /// reader does not report the end of the connection as an error.
+ pub(crate) async fn shutdown(self) {
+ let Self {
+ shared,
+ queue,
+ mut tasks,
+ } = self;
+ shared.close(&IggyError::Disconnected);
+ drop(queue);
+ let _ = timeout(SHUTDOWN_TIMEOUT, &mut tasks.writer).await;
+ }
+
+ fn start<Writer, Reader>(
+ peer: SocketAddr,
+ writer: impl FnOnce(mpsc::Receiver<Frame>, Arc<MuxShared>) -> Writer,
+ reader: impl FnOnce(Arc<MuxShared>) -> Reader,
+ ) -> Self
+ where
+ Writer: Future<Output = ()> + Send + 'static,
+ Reader: Future<Output = ()> + Send + 'static,
+ {
+ let shared = Arc::new(MuxShared {
+ peer,
+ pending: StdMutex::new(Pending {
+ open: true,
+ waiters: HashMap::new(),
+ }),
+ permits: Semaphore::new(MAX_PENDING_PER_LINK),
+ unbound: Mutex::new(()),
+ closed: watch::Sender::new(false),
+ });
+ let (queue, frames) = mpsc::channel(LINK_WRITE_QUEUE);
+ let tasks = LinkTasks {
+ writer: tokio::spawn(writer(frames, Arc::clone(&shared))),
+ reader: tokio::spawn(reader(Arc::clone(&shared))),
+ };
+ Self {
+ shared,
+ queue,
+ tasks,
+ }
+ }
+}
+
+impl MuxShared {
+ fn wait_for(&self, request: u64, waiter: Waiter) ->
Result<PendingEntry<'_>, IggyError> {
+ let mut pending = self.lock_pending();
+ if !pending.open {
+ return Err(IggyError::NotConnected);
+ }
+ match pending.waiters.entry(request) {
+ // Two requests with one id could not tell their replies apart.
+ Entry::Occupied(_) => {
+ error!(
+ "Request {request} already waits for a reply from {}.",
+ self.peer
+ );
+ Err(IggyError::NotConnected)
+ }
+ Entry::Vacant(entry) => {
+ entry.insert(waiter);
+ Ok(PendingEntry {
+ shared: self,
+ request,
+ close_if_abandoned: false,
+ })
+ }
+ }
+ }
+
+ /// Cuts every whole frame off the front of `buffer` and dispatches it.
+ fn deliver_buffered(&self, buffer: &mut BytesMut) -> Result<(), IggyError>
{
+ while let Some(&header) = buffer.first_chunk::<HEADER_SIZE>() {
+ let Ok(size) = response_size(&header) else {
+ error!(
+ "Received a frame shorter than its header from {}.",
+ self.peer
+ );
+ return Err(IggyError::Disconnected);
+ };
+ if buffer.len() < size {
+ break;
+ }
+ let body = buffer.split_to(size).freeze().slice(HEADER_SIZE..);
+ self.dispatch(Reply { header, body })?;
+ }
+ Ok(())
+ }
+
+ /// Gives a reply to the request that waits for it. An error is the reason
+ /// to close the link: an eviction ends the session, so it ends every
+ /// request.
+ fn dispatch(&self, reply: Reply) -> Result<(), IggyError> {
+ match peek_command(&reply.header) {
+ Command::Reply => {
+ let request = reply_request_id(&reply.header);
+ let waiter = self.lock_pending().waiters.remove(&request);
+ match waiter {
+ Some(waiter) => {
+ let _ = waiter.send(Ok(reply));
+ }
+ None => trace!(
+ "Dropped a reply to request {request} from {}, which
nothing waits for.",
+ self.peer
+ ),
+ }
+ Ok(())
+ }
+ Command::Eviction => Err(decode_eviction(&reply.header)),
+ _ => {
+ error!(
+ "Received a frame that is neither a reply nor an eviction
from {}.",
+ self.peer
+ );
+ Err(IggyError::Disconnected)
+ }
+ }
+ }
+
+ /// Fails every waiting request with `reason`. A request that comes later
+ /// fails with `NotConnected`.
+ fn close(&self, reason: &IggyError) {
+ let waiters = {
+ let mut pending = self.lock_pending();
+ pending.open = false;
+ std::mem::take(&mut pending.waiters)
+ };
+ self.permits.close();
+ self.closed.send_replace(true);
+ for waiter in waiters.into_values() {
+ let _ = waiter.send(Err(reason.clone()));
+ }
+ }
+
+ /// The next frame to write, or `None` once the link closed or every
+ /// sender went away.
+ async fn next_frame(&self, frames: &mut mpsc::Receiver<Frame>) ->
Option<Frame> {
+ let mut closed = self.closed.subscribe();
+ tokio::select! {
+ biased;
+ _ = closed.wait_for(|closed| *closed) => None,
+ frame = frames.recv() => frame,
+ }
+ }
+
+ fn is_open(&self) -> bool {
+ self.lock_pending().open
+ }
+
+ fn lock_pending(&self) -> MutexGuard<'_, Pending> {
+ self.pending.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
+impl Drop for PendingEntry<'_> {
+ fn drop(&mut self) {
+ let abandoned = self
+ .shared
+ .lock_pending()
+ .waiters
+ .remove(&self.request)
+ .is_some();
+ if abandoned && self.close_if_abandoned {
+ debug!(
+ "Closing the link to {}: request {} left before its reply.",
+ self.shared.peer, self.request
+ );
+ self.shared.close(&IggyError::Disconnected);
+ }
+ }
+}
+
+async fn write_tcp<W: AsyncWrite + Unpin>(
+ mut writer: W,
+ mut frames: mpsc::Receiver<Frame>,
+ shared: Arc<MuxShared>,
+) {
+ while let Some(frame) = shared.next_frame(&mut frames).await {
+ let written = async {
+ writer.write_all(bytemuck::bytes_of(&frame.header)).await?;
Review Comment:
Deferred to multi-leader PR6. Measure gather writes against the current
buffering path across the 8 KiB boundary, including nodelay=false and delayed
ACKs, and preserve partial-write and frame-boundary handling.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
