struct DecryptPlaintextFallbackBatch {
fallback_tx: Option<DecryptWorkerFallbackSender>,
fallbacks: Vec<DecryptFallback>,
authenticated_session_fallback_tx: Option<DecryptWorkerFallbackSender>,
authenticated_sessions: Vec<DecryptAuthenticatedSession>,
endpoint_fallback_tx: Option<DecryptWorkerFallbackSender>,
endpoint_sink: Option<DecryptDirectSessionDeliverySink>,
endpoint_commits: Vec<DecryptDirectSessionCommit>,
endpoint_deliveries: Vec<EndpointDataDelivery>,
direct_fallback_tx: Option<DecryptWorkerFallbackSender>,
direct_commits: Vec<DecryptDirectSessionCommit>,
direct_deliveries: Vec<PendingDirectSessionDelivery>,
direct_data_fallback_tx: Option<DecryptWorkerFallbackSender>,
direct_data: Vec<DecryptDirectSessionData>,
}
impl DecryptPlaintextFallbackBatch {
fn new() -> Self {
Self {
fallback_tx: None,
fallbacks: Vec::with_capacity(DECRYPT_WORKER_BULK_BATCH_MAX),
authenticated_session_fallback_tx: None,
authenticated_sessions: Vec::with_capacity(DECRYPT_WORKER_BULK_BATCH_MAX),
endpoint_fallback_tx: None,
endpoint_sink: None,
endpoint_commits: Vec::with_capacity(DECRYPT_WORKER_ENDPOINT_DELIVERY_BATCH_MAX),
endpoint_deliveries: Vec::with_capacity(DECRYPT_WORKER_ENDPOINT_DELIVERY_BATCH_MAX),
direct_fallback_tx: None,
direct_commits: Vec::with_capacity(DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX),
direct_deliveries: Vec::with_capacity(DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX),
direct_data_fallback_tx: None,
direct_data: Vec::with_capacity(DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX),
}
}
fn batch_max_for(fallback_tx: &DecryptWorkerFallbackSender) -> usize {
fallback_tx
.bulk_packet_cap
.clamp(1, DECRYPT_WORKER_BULK_BATCH_MAX)
}
fn endpoint_batch_max_for(fallback_tx: &DecryptWorkerFallbackSender) -> usize {
fallback_tx
.bulk_packet_cap
.clamp(1, DECRYPT_WORKER_ENDPOINT_DELIVERY_BATCH_MAX)
}
fn direct_batch_max_for(fallback_tx: &DecryptWorkerFallbackSender) -> usize {
fallback_tx
.bulk_packet_cap
.clamp(1, DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX)
}
fn push_output(&mut self, output: DecryptWorkerOutput) {
if output.is_batchable_bulk_plaintext() {
self.flush_authenticated_sessions();
self.flush_endpoint();
self.flush_direct();
self.flush_direct_data();
let DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
} = output;
debug_assert!(direct_delivery.is_none());
let DecryptWorkerEvent::Plaintext(fallback) = event else {
unreachable!("checked batchable plaintext output")
};
if self
.fallback_tx
.as_ref()
.is_some_and(|current| !current.same_channels(&fallback_tx))
{
self.flush();
}
if self.fallback_tx.is_none() {
self.fallback_tx = Some(fallback_tx);
}
let batch_max = Self::batch_max_for(
self.fallback_tx
.as_ref()
.expect("fallback sender set before batching plaintext"),
);
self.fallbacks.push(fallback);
if self.fallbacks.len() >= batch_max {
self.flush();
}
return;
}
if output.is_batchable_authenticated_session() {
self.flush_plaintext();
self.flush_endpoint();
self.flush_direct();
self.flush_direct_data();
let DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
} = output;
debug_assert!(direct_delivery.is_none());
let DecryptWorkerEvent::AuthenticatedSession(session) = event else {
unreachable!("checked batchable authenticated session output")
};
if self
.authenticated_session_fallback_tx
.as_ref()
.is_some_and(|current| !current.same_channels(&fallback_tx))
{
self.flush_authenticated_sessions();
}
if self.authenticated_session_fallback_tx.is_none() {
self.authenticated_session_fallback_tx = Some(fallback_tx);
}
let batch_max = Self::batch_max_for(
self.authenticated_session_fallback_tx
.as_ref()
.expect("fallback sender set before batching authenticated sessions"),
);
self.authenticated_sessions.push(session);
if self.authenticated_sessions.len() >= batch_max {
self.flush_authenticated_sessions();
}
return;
}
if output.is_batchable_direct_endpoint() {
self.flush_plaintext();
self.flush_authenticated_sessions();
self.flush_direct();
self.flush_direct_data();
let DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
} = output;
let DecryptWorkerEvent::DirectSessionCommit(commit) = event else {
unreachable!("checked batchable direct endpoint commit output")
};
let Some(direct_delivery) = direct_delivery else {
unreachable!("checked batchable direct endpoint delivery")
};
let Ok((sink, delivery)) = direct_delivery.into_endpoint_data() else {
unreachable!("checked batchable endpoint delivery")
};
let same_fallback = self
.endpoint_fallback_tx
.as_ref()
.is_none_or(|current| current.same_channels(&fallback_tx));
let same_endpoint = self
.endpoint_sink
.as_ref()
.is_none_or(|current| current.same_endpoint_event_channel(&sink));
if !same_fallback || !same_endpoint {
self.flush_endpoint();
}
if self.endpoint_fallback_tx.is_none() {
self.endpoint_fallback_tx = Some(fallback_tx);
}
if self.endpoint_sink.is_none() {
self.endpoint_sink = Some(sink);
}
let batch_max = Self::endpoint_batch_max_for(
self.endpoint_fallback_tx
.as_ref()
.expect("fallback sender set before batching direct endpoint completions"),
);
self.endpoint_commits.push(commit);
self.endpoint_deliveries.push(delivery);
if self.endpoint_commits.len() >= batch_max {
self.flush_endpoint();
}
return;
}
if output.is_batchable_direct_ipv6() {
self.flush_plaintext();
self.flush_authenticated_sessions();
self.flush_endpoint();
self.flush_direct_data();
let DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
} = output;
let DecryptWorkerEvent::DirectSessionCommit(commit) = event else {
unreachable!("checked batchable direct IPv6 commit output")
};
let Some(direct_delivery) = direct_delivery else {
unreachable!("checked batchable direct IPv6 delivery")
};
if self
.direct_fallback_tx
.as_ref()
.is_some_and(|current| !current.same_channels(&fallback_tx))
{
self.flush_direct();
}
if self.direct_fallback_tx.is_none() {
self.direct_fallback_tx = Some(fallback_tx);
}
let batch_max = Self::direct_batch_max_for(
self.direct_fallback_tx
.as_ref()
.expect("fallback sender set before batching direct completions"),
);
self.direct_commits.push(commit);
self.direct_deliveries.push(direct_delivery);
if self.direct_commits.len() >= batch_max {
self.flush_direct();
}
return;
}
if output.is_batchable_direct_data() {
self.flush_plaintext();
self.flush_authenticated_sessions();
self.flush_endpoint();
self.flush_direct();
let DecryptWorkerOutput {
fallback_tx,
event,
direct_delivery,
} = output;
debug_assert!(direct_delivery.is_none());
let DecryptWorkerEvent::DirectSessionData(direct) = event else {
unreachable!("checked batchable direct data output")
};
if self
.direct_data_fallback_tx
.as_ref()
.is_some_and(|current| !current.same_channels(&fallback_tx))
{
self.flush_direct_data();
}
if self.direct_data_fallback_tx.is_none() {
self.direct_data_fallback_tx = Some(fallback_tx);
}
let batch_max = Self::direct_batch_max_for(
self.direct_data_fallback_tx
.as_ref()
.expect("fallback sender set before batching direct data"),
);
self.direct_data.push(direct);
if self.direct_data.len() >= batch_max {
self.flush_direct_data();
}
return;
}
self.flush();
let _ = output.send();
}
fn flush(&mut self) {
self.flush_plaintext();
self.flush_authenticated_sessions();
self.flush_endpoint();
self.flush_direct();
self.flush_direct_data();
}
fn flush_plaintext(&mut self) {
if self.fallbacks.is_empty() {
return;
}
let _t_flush =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptWorkerOutputFlush);
let Some(fallback_tx) = self.fallback_tx.take() else {
return;
};
let event = if self.fallbacks.len() == 1 {
DecryptWorkerEvent::Plaintext(self.fallbacks.pop().expect("checked single fallback"))
} else {
let fallbacks = std::mem::replace(
&mut self.fallbacks,
Vec::with_capacity(DECRYPT_WORKER_BULK_BATCH_MAX),
);
DecryptWorkerEvent::PlaintextBatch(fallbacks)
};
let _ = fallback_tx.send(event);
}
fn flush_authenticated_sessions(&mut self) {
if self.authenticated_sessions.is_empty() {
return;
}
let _t_flush =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptWorkerOutputFlush);
let Some(fallback_tx) = self.authenticated_session_fallback_tx.take() else {
self.authenticated_sessions.clear();
return;
};
let event = if self.authenticated_sessions.len() == 1 {
DecryptWorkerEvent::AuthenticatedSession(
self.authenticated_sessions
.pop()
.expect("checked single authenticated session"),
)
} else {
let sessions = std::mem::replace(
&mut self.authenticated_sessions,
Vec::with_capacity(DECRYPT_WORKER_BULK_BATCH_MAX),
);
DecryptWorkerEvent::AuthenticatedSessionBatch(sessions)
};
let _ = fallback_tx.send(event);
}
fn flush_endpoint(&mut self) {
if self.endpoint_commits.is_empty() {
return;
}
let _t_flush =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptWorkerOutputFlush);
let Some(fallback_tx) = self.endpoint_fallback_tx.take() else {
return;
};
let Some(sink) = self.endpoint_sink.take() else {
self.endpoint_commits.clear();
self.endpoint_deliveries.clear();
return;
};
let Some(endpoint_event_tx) = sink.endpoint_event_sender().cloned() else {
self.endpoint_commits.clear();
self.endpoint_deliveries.clear();
return;
};
let event = if self.endpoint_commits.len() == 1 {
DecryptWorkerEvent::DirectSessionCommit(
self.endpoint_commits
.pop()
.expect("checked single direct endpoint commit"),
)
} else {
let commits = std::mem::replace(
&mut self.endpoint_commits,
Vec::with_capacity(DECRYPT_WORKER_ENDPOINT_DELIVERY_BATCH_MAX),
);
DecryptWorkerEvent::DirectSessionCommitBatch(commits)
};
if !fallback_tx.send(event) {
self.endpoint_deliveries.clear();
return;
}
let count = self.endpoint_deliveries.len();
if count == 0 {
return;
}
let queued_at = crate::perf_profile::stamp();
let endpoint_event = if count == 1 {
let delivery = self
.endpoint_deliveries
.pop()
.expect("checked single endpoint delivery");
NodeEndpointEvent::Data {
source_peer: delivery.source_peer,
payload: delivery.payload,
queued_at,
}
} else {
let messages = std::mem::replace(
&mut self.endpoint_deliveries,
Vec::with_capacity(DECRYPT_WORKER_ENDPOINT_DELIVERY_BATCH_MAX),
);
NodeEndpointEvent::DataBatch {
messages,
queued_at,
}
};
let _t_deliver =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::EndpointDeliver);
if let Err(error) = endpoint_event_tx.send(endpoint_event) {
debug!(
error = %error,
messages = count,
"Failed to deliver worker-decoded endpoint data batch"
);
}
}
fn flush_direct(&mut self) {
if self.direct_commits.is_empty() {
return;
}
let _t_flush =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptWorkerOutputFlush);
let Some(fallback_tx) = self.direct_fallback_tx.take() else {
self.direct_commits.clear();
self.direct_deliveries.clear();
return;
};
let event = if self.direct_commits.len() == 1 {
DecryptWorkerEvent::DirectSessionCommit(
self.direct_commits
.pop()
.expect("checked single direct commit"),
)
} else {
let commits = std::mem::replace(
&mut self.direct_commits,
Vec::with_capacity(DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX),
);
DecryptWorkerEvent::DirectSessionCommitBatch(commits)
};
if !fallback_tx.send(event) {
self.direct_deliveries.clear();
return;
}
for delivery in self.direct_deliveries.drain(..) {
delivery.deliver();
}
}
fn flush_direct_data(&mut self) {
if self.direct_data.is_empty() {
return;
}
let _t_flush =
crate::perf_profile::Timer::start(crate::perf_profile::Stage::DecryptWorkerOutputFlush);
let Some(fallback_tx) = self.direct_data_fallback_tx.take() else {
self.direct_data.clear();
return;
};
let event = if self.direct_data.len() == 1 {
DecryptWorkerEvent::DirectSessionData(
self.direct_data
.pop()
.expect("checked single direct data"),
)
} else {
let direct_data = std::mem::replace(
&mut self.direct_data,
Vec::with_capacity(DECRYPT_WORKER_DIRECT_DELIVERY_BATCH_MAX),
);
DecryptWorkerEvent::DirectSessionDataBatch(direct_data)
};
let _ = fallback_tx.send(event);
}
}