#[derive(Clone)]
pub(crate) struct DecryptWorkerFallbackSender {
priority: TokioSender<DecryptWorkerEvent>,
bulk: TokioSender<DecryptWorkerEvent>,
authenticated_bulk: TokioSender<DecryptWorkerEvent>,
bulk_queued_packets: Arc<AtomicUsize>,
authenticated_bulk_queued_packets: Arc<AtomicUsize>,
bulk_packet_cap: usize,
}
pub(crate) struct DecryptWorkerFallbackReceivers {
pub(crate) priority: TokioReceiver<DecryptWorkerEvent>,
pub(crate) bulk: TokioReceiver<DecryptWorkerEvent>,
pub(crate) authenticated_bulk: TokioReceiver<DecryptWorkerEvent>,
bulk_queued_packets: Arc<AtomicUsize>,
authenticated_bulk_queued_packets: Arc<AtomicUsize>,
}
pub(crate) fn decrypt_worker_fallback_channels()
-> (DecryptWorkerFallbackSender, DecryptWorkerFallbackReceivers) {
decrypt_worker_fallback_channels_with_caps(
fallback_priority_channel_cap(),
fallback_bulk_channel_cap(),
)
}
fn decrypt_worker_fallback_channels_with_caps(
priority_cap: usize,
bulk_cap: usize,
) -> (DecryptWorkerFallbackSender, DecryptWorkerFallbackReceivers) {
let (priority_tx, priority_rx) = tokio::sync::mpsc::channel(priority_cap.max(1));
let (bulk_tx, bulk_rx) = tokio::sync::mpsc::channel(bulk_cap.max(1));
let (authenticated_bulk_tx, authenticated_bulk_rx) =
tokio::sync::mpsc::channel(bulk_cap.max(1));
let bulk_queued_packets = Arc::new(AtomicUsize::new(0));
let authenticated_bulk_queued_packets = Arc::new(AtomicUsize::new(0));
(
DecryptWorkerFallbackSender {
priority: priority_tx,
bulk: bulk_tx,
authenticated_bulk: authenticated_bulk_tx,
bulk_queued_packets: Arc::clone(&bulk_queued_packets),
authenticated_bulk_queued_packets: Arc::clone(&authenticated_bulk_queued_packets),
bulk_packet_cap: bulk_cap.max(1),
},
DecryptWorkerFallbackReceivers {
priority: priority_rx,
bulk: bulk_rx,
authenticated_bulk: authenticated_bulk_rx,
bulk_queued_packets,
authenticated_bulk_queued_packets,
},
)
}
impl DecryptWorkerFallbackSender {
fn same_channels(&self, other: &Self) -> bool {
self.priority.same_channel(&other.priority)
&& self.bulk.same_channel(&other.bulk)
&& self
.authenticated_bulk
.same_channel(&other.authenticated_bulk)
&& Arc::ptr_eq(&self.bulk_queued_packets, &other.bulk_queued_packets)
&& Arc::ptr_eq(
&self.authenticated_bulk_queued_packets,
&other.authenticated_bulk_queued_packets,
)
&& self.bulk_packet_cap == other.bulk_packet_cap
}
fn send(&self, mut event: DecryptWorkerEvent) -> bool {
let lane = decrypt_worker_event_lane(&event);
let packet_count = event.packet_count();
let drop_event = decrypt_worker_event_drop_event(&event, lane);
let bulk_lane = if matches!(lane, DecryptWorkerLane::Bulk) {
Some(decrypt_worker_event_return_bulk_lane(&event))
} else {
None
};
event.set_trace_enqueued_at(crate::perf_profile::stamp());
if let Some(bulk_lane) = bulk_lane {
let queued_packets = self.return_bulk_queued_packets(bulk_lane);
let Some(previous) = try_reserve_bulk_packets_with_previous(
queued_packets,
self.bulk_packet_cap,
packet_count,
) else {
record_decrypt_worker_return_drop_count(drop_event, lane, packet_count);
return false;
};
let queued = previous.saturating_add(packet_count);
if previous < DECRYPT_FALLBACK_BACKLOG_HIGH_WATER
&& queued >= DECRYPT_FALLBACK_BACKLOG_HIGH_WATER
{
let event = match bulk_lane {
DecryptWorkerReturnBulkLane::Fallback => {
crate::perf_profile::Event::DecryptFallbackBacklogHigh
}
DecryptWorkerReturnBulkLane::Authenticated => {
crate::perf_profile::Event::DecryptAuthenticatedBacklogHigh
}
};
crate::perf_profile::record_event(event);
}
}
let result = match lane {
DecryptWorkerLane::Priority => self.priority.try_send(event),
DecryptWorkerLane::Bulk => match bulk_lane.expect("bulk event has return bulk lane") {
DecryptWorkerReturnBulkLane::Fallback => self.bulk.try_send(event),
DecryptWorkerReturnBulkLane::Authenticated => {
self.authenticated_bulk.try_send(event)
}
},
};
match result {
Ok(()) => true,
Err(TokioTrySendError::Full(_)) => {
if let Some(bulk_lane) = bulk_lane {
release_bulk_packets(self.return_bulk_queued_packets(bulk_lane), packet_count);
}
record_decrypt_worker_return_drop_count(drop_event, lane, packet_count);
false
}
Err(TokioTrySendError::Closed(_)) => {
if let Some(bulk_lane) = bulk_lane {
release_bulk_packets(self.return_bulk_queued_packets(bulk_lane), packet_count);
}
debug!(
?lane,
"decrypt fallback receiver gone; dropping worker event"
);
false
}
}
}
fn return_bulk_queued_packets(&self, lane: DecryptWorkerReturnBulkLane) -> &Arc<AtomicUsize> {
match lane {
DecryptWorkerReturnBulkLane::Fallback => &self.bulk_queued_packets,
DecryptWorkerReturnBulkLane::Authenticated => &self.authenticated_bulk_queued_packets,
}
}
}
impl DecryptWorkerFallbackReceivers {
pub(crate) fn release_dequeued_event(&self, event: &DecryptWorkerEvent) {
if matches!(event.lane(), DecryptWorkerLane::Bulk) {
let queued_packets =
self.return_bulk_queued_packets(decrypt_worker_event_return_bulk_lane(event));
release_bulk_packets(queued_packets, event.packet_count());
}
}
#[cfg(test)]
pub(crate) fn bulk_queued_packets(&self) -> usize {
self.bulk_queued_packets.load(Ordering::Relaxed)
}
#[cfg(test)]
pub(crate) fn bulk_pressure_queued_packets(&self) -> usize {
self.bulk_queued_packets()
.saturating_add(self.authenticated_bulk_queued_packets())
}
#[cfg(test)]
pub(crate) fn authenticated_bulk_queued_packets(&self) -> usize {
self.authenticated_bulk_queued_packets
.load(Ordering::Relaxed)
}
fn return_bulk_queued_packets(&self, lane: DecryptWorkerReturnBulkLane) -> &Arc<AtomicUsize> {
match lane {
DecryptWorkerReturnBulkLane::Fallback => &self.bulk_queued_packets,
DecryptWorkerReturnBulkLane::Authenticated => &self.authenticated_bulk_queued_packets,
}
}
}
fn decrypt_worker_event_lane(event: &DecryptWorkerEvent) -> DecryptWorkerLane {
match event {
DecryptWorkerEvent::AuthenticatedFmpReceive(receive) => receive.lane,
DecryptWorkerEvent::Plaintext(fallback) => fallback.lane(),
DecryptWorkerEvent::PlaintextBatch(_) => DecryptWorkerLane::Bulk,
DecryptWorkerEvent::AuthenticatedSession(session) => session.lane,
DecryptWorkerEvent::AuthenticatedSessionBatch(_) => DecryptWorkerLane::Bulk,
DecryptWorkerEvent::DirectSessionCommit(commit) => commit.lane,
DecryptWorkerEvent::DirectSessionCommitBatch(_) => DecryptWorkerLane::Bulk,
DecryptWorkerEvent::DirectSessionData(direct) => direct.lane,
DecryptWorkerEvent::DirectSessionDataBatch(_) => DecryptWorkerLane::Bulk,
DecryptWorkerEvent::FspDecryptFailure(report) => report.lane,
DecryptWorkerEvent::DecryptFailure(_) => DecryptWorkerLane::Priority,
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum DecryptWorkerReturnBulkLane {
Fallback,
Authenticated,
}
fn decrypt_worker_event_return_bulk_lane(
event: &DecryptWorkerEvent,
) -> DecryptWorkerReturnBulkLane {
match event {
DecryptWorkerEvent::AuthenticatedFmpReceive(_)
| DecryptWorkerEvent::AuthenticatedSession(_)
| DecryptWorkerEvent::AuthenticatedSessionBatch(_)
| DecryptWorkerEvent::DirectSessionCommit(_)
| DecryptWorkerEvent::DirectSessionCommitBatch(_)
| DecryptWorkerEvent::DirectSessionData(_)
| DecryptWorkerEvent::DirectSessionDataBatch(_) => {
DecryptWorkerReturnBulkLane::Authenticated
}
DecryptWorkerEvent::Plaintext(_)
| DecryptWorkerEvent::PlaintextBatch(_)
| DecryptWorkerEvent::FspDecryptFailure(_)
| DecryptWorkerEvent::DecryptFailure(_) => DecryptWorkerReturnBulkLane::Fallback,
}
}
fn decrypt_worker_event_drop_event(
event: &DecryptWorkerEvent,
lane: DecryptWorkerLane,
) -> crate::perf_profile::Event {
match event {
DecryptWorkerEvent::AuthenticatedFmpReceive(_)
| DecryptWorkerEvent::AuthenticatedSession(_)
| DecryptWorkerEvent::AuthenticatedSessionBatch(_)
| DecryptWorkerEvent::DirectSessionCommit(_)
| DecryptWorkerEvent::DirectSessionCommitBatch(_)
| DecryptWorkerEvent::DirectSessionData(_)
| DecryptWorkerEvent::DirectSessionDataBatch(_) => match lane {
DecryptWorkerLane::Priority => {
crate::perf_profile::Event::DecryptAuthenticatedSessionPriorityDropped
}
DecryptWorkerLane::Bulk => {
crate::perf_profile::Event::DecryptAuthenticatedSessionBulkDropped
}
},
DecryptWorkerEvent::Plaintext(_)
| DecryptWorkerEvent::PlaintextBatch(_)
| DecryptWorkerEvent::FspDecryptFailure(_)
| DecryptWorkerEvent::DecryptFailure(_) => match lane {
DecryptWorkerLane::Priority => {
crate::perf_profile::Event::DecryptFallbackPriorityDropped
}
DecryptWorkerLane::Bulk => crate::perf_profile::Event::DecryptFallbackBulkDropped,
},
}
}