fips_core/upper/
tun_write.rs

1use std::collections::VecDeque;
2use std::sync::{Arc, Condvar, Mutex, MutexGuard, mpsc};
3
4use crate::transport::PacketBuffer;
5
6const DEFAULT_TUN_WRITE_BULK_QUEUE_CAP: usize = 1024;
7const MAX_TUN_WRITE_BULK_QUEUE_CAP: usize = 65_536;
8
9/// Queue lane for packets waiting on the blocking TUN writer.
10#[derive(Clone, Copy, Debug, Eq, PartialEq)]
11pub(crate) enum TunWriteLane {
12    Priority,
13    Bulk,
14}
15
16#[derive(Clone, Copy, Debug, Eq, PartialEq)]
17pub(crate) enum TunWriteErrorKind {
18    Closed,
19    BulkFull,
20}
21
22#[derive(Debug)]
23pub(crate) struct TunWriteError {
24    packet: TunWritePacket,
25    kind: TunWriteErrorKind,
26}
27
28impl TunWriteError {
29    pub(crate) fn kind(&self) -> TunWriteErrorKind {
30        self.kind
31    }
32
33    pub(crate) fn into_packet(self) -> Vec<u8> {
34        self.packet.into_vec()
35    }
36}
37
38impl std::fmt::Display for TunWriteError {
39    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
40        match self.kind {
41            TunWriteErrorKind::Closed => write!(f, "TUN write channel closed"),
42            TunWriteErrorKind::BulkFull => write!(f, "TUN bulk write queue full"),
43        }
44    }
45}
46
47impl std::error::Error for TunWriteError {}
48
49#[derive(Clone, Copy, Debug, Eq, PartialEq)]
50pub(crate) struct TunWriteBatchError {
51    pub(crate) index: usize,
52    pub(crate) kind: TunWriteErrorKind,
53}
54
55#[derive(Debug)]
56struct TunWriteQueue {
57    state: Mutex<TunWriteState>,
58    ready: Condvar,
59}
60
61#[derive(Debug)]
62struct TunWriteState {
63    priority: VecDeque<TunWritePacket>,
64    bulk: VecDeque<TunWritePacket>,
65    senders: usize,
66    receiver_alive: bool,
67    bulk_capacity: usize,
68}
69
70#[derive(Debug)]
71pub(crate) enum TunWritePacket {
72    Vec(Vec<u8>),
73    Pooled(PacketBuffer),
74}
75
76impl TunWritePacket {
77    pub(crate) fn as_slice(&self) -> &[u8] {
78        match self {
79            Self::Vec(packet) => packet,
80            Self::Pooled(packet) => packet,
81        }
82    }
83
84    pub(crate) fn as_mut_slice(&mut self) -> &mut [u8] {
85        match self {
86            Self::Vec(packet) => packet,
87            Self::Pooled(packet) => packet,
88        }
89    }
90
91    pub(crate) fn len(&self) -> usize {
92        self.as_slice().len()
93    }
94
95    pub(crate) fn into_vec(self) -> Vec<u8> {
96        match self {
97            Self::Vec(packet) => packet,
98            Self::Pooled(packet) => packet.into_vec(),
99        }
100    }
101}
102
103impl From<Vec<u8>> for TunWritePacket {
104    fn from(packet: Vec<u8>) -> Self {
105        Self::Vec(packet)
106    }
107}
108
109impl From<PacketBuffer> for TunWritePacket {
110    fn from(packet: PacketBuffer) -> Self {
111        Self::Pooled(packet)
112    }
113}
114
115impl PartialEq<Vec<u8>> for TunWritePacket {
116    fn eq(&self, other: &Vec<u8>) -> bool {
117        self.as_slice() == other.as_slice()
118    }
119}
120
121impl PartialEq<TunWritePacket> for Vec<u8> {
122    fn eq(&self, other: &TunWritePacket) -> bool {
123        self.as_slice() == other.as_slice()
124    }
125}
126
127/// Channel sender for packets to be written to TUN.
128#[derive(Debug)]
129pub struct TunTx {
130    queue: Arc<TunWriteQueue>,
131}
132
133/// Channel receiver consumed by the blocking TUN writer.
134#[derive(Debug)]
135pub(crate) struct TunRx {
136    queue: Arc<TunWriteQueue>,
137}
138
139impl Clone for TunTx {
140    fn clone(&self) -> Self {
141        {
142            let mut state = self.queue.lock();
143            state.senders = state.senders.saturating_add(1);
144        }
145        Self {
146            queue: Arc::clone(&self.queue),
147        }
148    }
149}
150
151impl Drop for TunTx {
152    fn drop(&mut self) {
153        let should_notify = {
154            let mut state = self.queue.lock();
155            state.senders = state.senders.saturating_sub(1);
156            state.senders == 0
157        };
158        if should_notify {
159            self.queue.ready.notify_all();
160        }
161    }
162}
163
164impl TunTx {
165    /// Queue a priority packet for TUN delivery.
166    pub fn send(&self, packet: Vec<u8>) -> Result<(), mpsc::SendError<Vec<u8>>> {
167        self.send_with_lane(packet, TunWriteLane::Priority)
168            .map_err(|error| mpsc::SendError(error.into_packet()))
169    }
170
171    /// Queue a packet for TUN delivery, allowing bulk to shed under pressure.
172    pub(crate) fn send_with_lane(
173        &self,
174        packet: impl Into<TunWritePacket>,
175        lane: TunWriteLane,
176    ) -> Result<(), TunWriteError> {
177        let packet = packet.into();
178        let mut state = self.queue.lock();
179        if !state.receiver_alive {
180            return Err(TunWriteError {
181                packet,
182                kind: TunWriteErrorKind::Closed,
183            });
184        }
185
186        enqueue_tun_write_packet(&mut state, packet, lane)?;
187        drop(state);
188        self.queue.ready.notify_one();
189        Ok(())
190    }
191
192    pub(crate) fn send_batch_with_lanes<I, P>(&self, packets: I) -> Vec<TunWriteBatchError>
193    where
194        I: IntoIterator<Item = (P, TunWriteLane)>,
195        P: Into<TunWritePacket>,
196    {
197        let mut state = self.queue.lock();
198        let mut failures = Vec::new();
199        let mut sent = 0usize;
200        for (index, (packet, lane)) in packets.into_iter().enumerate() {
201            let packet = packet.into();
202            if !state.receiver_alive {
203                failures.push(TunWriteBatchError {
204                    index,
205                    kind: TunWriteErrorKind::Closed,
206                });
207                continue;
208            }
209            match enqueue_tun_write_packet(&mut state, packet, lane) {
210                Ok(()) => sent = sent.saturating_add(1),
211                Err(error) => failures.push(TunWriteBatchError {
212                    index,
213                    kind: error.kind,
214                }),
215            }
216        }
217        drop(state);
218        if sent > 0 {
219            self.queue.ready.notify_one();
220        }
221        failures
222    }
223}
224
225fn enqueue_tun_write_packet(
226    state: &mut TunWriteState,
227    packet: TunWritePacket,
228    lane: TunWriteLane,
229) -> Result<(), TunWriteError> {
230    match lane {
231        TunWriteLane::Priority => state.priority.push_back(packet),
232        TunWriteLane::Bulk => {
233            if state.bulk.len() >= state.bulk_capacity {
234                crate::perf_profile::record_event(crate::perf_profile::Event::TunWriteBulkDropped);
235                return Err(TunWriteError {
236                    packet,
237                    kind: TunWriteErrorKind::BulkFull,
238                });
239            }
240            let high_water = (state.bulk_capacity / 2).max(1);
241            let previous = state.bulk.len();
242            state.bulk.push_back(packet);
243            if previous < high_water && state.bulk.len() >= high_water {
244                crate::perf_profile::record_event(
245                    crate::perf_profile::Event::TunWriteBulkBacklogHigh,
246                );
247            }
248        }
249    }
250    Ok(())
251}
252
253impl TunRx {
254    pub(crate) fn recv(&self) -> Option<TunWritePacket> {
255        let mut state = self.queue.lock();
256        loop {
257            if let Some(packet) = state.priority.pop_front() {
258                return Some(packet);
259            }
260            if let Some(packet) = state.bulk.pop_front() {
261                return Some(packet);
262            }
263            if state.senders == 0 {
264                state.receiver_alive = false;
265                return None;
266            }
267            state = self
268                .queue
269                .ready
270                .wait(state)
271                .unwrap_or_else(|poisoned| poisoned.into_inner());
272        }
273    }
274
275    #[cfg(test)]
276    pub(crate) fn try_recv(&self) -> Result<Vec<u8>, mpsc::TryRecvError> {
277        self.try_recv_packet().map(TunWritePacket::into_vec)
278    }
279
280    #[cfg(test)]
281    pub(crate) fn try_recv_packet(&self) -> Result<TunWritePacket, mpsc::TryRecvError> {
282        let mut state = self.queue.lock();
283        if let Some(packet) = state.priority.pop_front() {
284            return Ok(packet);
285        }
286        if let Some(packet) = state.bulk.pop_front() {
287            return Ok(packet);
288        }
289        if state.senders == 0 {
290            state.receiver_alive = false;
291            Err(mpsc::TryRecvError::Disconnected)
292        } else {
293            Err(mpsc::TryRecvError::Empty)
294        }
295    }
296}
297
298impl Drop for TunRx {
299    fn drop(&mut self) {
300        let mut state = self.queue.lock();
301        state.receiver_alive = false;
302        state.priority.clear();
303        state.bulk.clear();
304        drop(state);
305        self.queue.ready.notify_all();
306    }
307}
308
309impl Iterator for TunRx {
310    type Item = TunWritePacket;
311
312    fn next(&mut self) -> Option<Self::Item> {
313        self.recv()
314    }
315}
316
317impl TunWriteQueue {
318    fn lock(&self) -> MutexGuard<'_, TunWriteState> {
319        self.state
320            .lock()
321            .unwrap_or_else(|poisoned| poisoned.into_inner())
322    }
323}
324
325pub(crate) fn write_channel() -> (TunTx, TunRx) {
326    write_channel_with_bulk_capacity(tun_write_bulk_queue_cap())
327}
328
329pub(crate) fn write_channel_with_bulk_capacity(bulk_capacity: usize) -> (TunTx, TunRx) {
330    let queue = Arc::new(TunWriteQueue {
331        state: Mutex::new(TunWriteState {
332            priority: VecDeque::new(),
333            bulk: VecDeque::new(),
334            senders: 1,
335            receiver_alive: true,
336            bulk_capacity: bulk_capacity.max(1),
337        }),
338        ready: Condvar::new(),
339    });
340    (
341        TunTx {
342            queue: Arc::clone(&queue),
343        },
344        TunRx { queue },
345    )
346}
347
348fn tun_write_bulk_queue_cap() -> usize {
349    std::env::var("FIPS_TUN_WRITE_BULK_QUEUE_CAP")
350        .ok()
351        .and_then(|raw| raw.parse::<usize>().ok())
352        .map(|value| value.clamp(1, MAX_TUN_WRITE_BULK_QUEUE_CAP))
353        .unwrap_or(DEFAULT_TUN_WRITE_BULK_QUEUE_CAP)
354}
fips_core/upper/tun_write.rs

fips_core/upper/
tun_write.rs
