netstack3_device/queue/
tx.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5//! TX device queues.
6
7use core::convert::Infallible as Never;
8
9use alloc::vec::Vec;
10
11use derivative::Derivative;
12use log::trace;
13use netstack3_base::sync::Mutex;
14use netstack3_base::{
15    ChecksumOffloadResult, ChecksumOffloadSpec, Device, DeviceIdContext, ErrorAndSerializer,
16    NetworkSerializationContext, NetworkSerializer, TxMetadata,
17};
18use packet::{Buf, FragmentedBuffer as _, NoReuseBufferProvider, ReusableBuffer, new_buf_vec};
19
20use crate::internal::base::{DeviceBufferBindingsTypes, DeviceSendFrameError};
21use crate::internal::queue::{
22    DequeueState, DeviceBufferSpec, EnqueueResult, TransmitQueueFrameError, fifo,
23};
24use crate::internal::socket::{DeviceSocketHandler, ParseSentFrameError, SentFrame};
25
26/// State associated with a device transmit queue.
27#[derive(Derivative)]
28#[derivative(Default(bound = "Allocator: Default"))]
29pub struct TransmitQueueState<Meta, Buffer, Allocator> {
30    pub(super) allocator: ByteSliceAllocator<Allocator, Buffer>,
31    pub(super) queue: Option<fifo::Queue<Meta, Buffer>>,
32    pub(super) checksum_offload_spec: ChecksumOffloadSpec,
33}
34
35/// Holds queue and dequeue state for the transmit queue.
36pub struct TransmitQueue<Meta, Buffer, Allocator> {
37    /// The state for dequeued packets that will be handled.
38    ///
39    /// See `queue` for lock ordering.
40    pub(crate) deque: Mutex<DequeueState<Meta, Buffer>>,
41    /// A queue of to-be-transmitted packets protected by a lock.
42    ///
43    /// Lock ordering: `deque` must be locked before `queue` is locked when both
44    /// are needed at the same time.
45    pub(crate) queue: Mutex<TransmitQueueState<Meta, Buffer, Allocator>>,
46}
47
48impl<Meta, Buffer, Allocator> TransmitQueue<Meta, Buffer, Allocator> {
49    pub(crate) fn new(allocator: Allocator, checksum_offload_spec: ChecksumOffloadSpec) -> Self {
50        Self {
51            deque: Mutex::new(DequeueState::default()),
52            queue: Mutex::new(TransmitQueueState {
53                allocator: ByteSliceAllocator::new(allocator),
54                queue: None,
55                checksum_offload_spec,
56            }),
57        }
58    }
59
60    pub(crate) fn tx_offload_spec(&self) -> ChecksumOffloadSpec {
61        self.queue.lock().checksum_offload_spec.clone()
62    }
63}
64
65/// The bindings context for the transmit queue.
66pub trait TransmitQueueBindingsContext<DeviceId>: DeviceBufferBindingsTypes {
67    /// Signals to bindings that TX frames are available and ready to be sent
68    /// over the device.
69    ///
70    /// Implementations must make sure that the API call to handle queued
71    /// packets is scheduled to be called as soon as possible so that enqueued
72    /// TX frames are promptly handled.
73    fn wake_tx_task(&mut self, device_id: &DeviceId);
74}
75
76/// A trait for metadata that is attached to a frame in the queue.
77pub trait TxQueuePacketMetadataCommon {
78    /// Sets the checksum offload result.
79    fn set_checksum_offload_result(&mut self, result: Option<ChecksumOffloadResult>);
80}
81
82impl<T: TxMetadata> TxQueuePacketMetadataCommon for T {
83    fn set_checksum_offload_result(&mut self, result: Option<ChecksumOffloadResult>) {
84        TxMetadata::set_checksum_offload_result(self, result);
85    }
86}
87
88/// Basic definitions for a transmit queue.
89pub trait TransmitQueueCommon<D: Device, C>: DeviceIdContext<D> {
90    /// The metadata associated with every packet in the queue.
91    type Meta: TxQueuePacketMetadataCommon;
92
93    /// The context given to `send_frame` when dequeueing.
94    type DequeueContext;
95
96    /// Parses an outgoing frame for packet socket delivery.
97    fn parse_outgoing_frame<'a, 'b>(
98        buf: &'a [u8],
99        meta: &'a Self::Meta,
100    ) -> Result<SentFrame<&'a [u8]>, ParseSentFrameError>;
101}
102
103/// The execution context for a transmit queue.
104pub trait TransmitQueueContext<D: DeviceBufferSpec<BC>, BC>: TransmitQueueCommon<D, BC> {
105    /// Calls `cb` with mutable access to the queue state.
106    fn with_transmit_queue_mut<
107        O,
108        F: FnOnce(&mut TransmitQueueState<Self::Meta, D::TxBuffer, D::TxAllocator>) -> O,
109    >(
110        &mut self,
111        device_id: &Self::DeviceId,
112        cb: F,
113    ) -> O;
114
115    /// Calls `cb` with immutable access to the queue state.
116    fn with_transmit_queue<
117        O,
118        F: FnOnce(&TransmitQueueState<Self::Meta, D::TxBuffer, D::TxAllocator>) -> O,
119    >(
120        &mut self,
121        device_id: &Self::DeviceId,
122        cb: F,
123    ) -> O;
124
125    /// Send a frame out the device.
126    ///
127    /// This method may not block - if the device is not ready, an appropriate
128    /// error must be returned.
129    fn send_frame(
130        &mut self,
131        bindings_ctx: &mut BC,
132        device_id: &Self::DeviceId,
133        dequeue_context: Option<&mut Self::DequeueContext>,
134        meta: Self::Meta,
135        buf: D::TxBuffer,
136    ) -> Result<(), DeviceSendFrameError>;
137}
138
139/// The core execution context for dequeueing TX frames from the transmit queue.
140pub trait TransmitDequeueContext<D: DeviceBufferSpec<BC>, BC>: TransmitQueueContext<D, BC> {
141    /// The inner context providing dequeuing.
142    type TransmitQueueCtx<'a>: TransmitQueueContext<
143            D,
144            BC,
145            Meta = Self::Meta,
146            DequeueContext = Self::DequeueContext,
147            DeviceId = Self::DeviceId,
148        > + DeviceSocketHandler<D, BC>;
149
150    /// Calls the function with the TX deque state and the TX queue context.
151    fn with_dequed_packets_and_tx_queue_ctx<
152        O,
153        F: FnOnce(&mut DequeueState<Self::Meta, D::TxBuffer>, &mut Self::TransmitQueueCtx<'_>) -> O,
154    >(
155        &mut self,
156        device_id: &Self::DeviceId,
157        cb: F,
158    ) -> O;
159}
160
161/// The configuration for a transmit queue.
162pub enum TransmitQueueConfiguration {
163    /// No queue.
164    None,
165    /// FiFo queue.
166    Fifo,
167}
168
169/// An implementation of a transmit queue that stores egress frames.
170pub trait TransmitQueueHandler<D: Device, BC>: TransmitQueueCommon<D, BC> {
171    /// Queues a frame for transmission.
172    fn queue_tx_frame<S>(
173        &mut self,
174        bindings_ctx: &mut BC,
175        device_id: &Self::DeviceId,
176        meta: Self::Meta,
177        body: S,
178    ) -> Result<usize, TransmitQueueFrameError<S>>
179    where
180        S: NetworkSerializer,
181        S::Buffer: ReusableBuffer;
182}
183
184pub(super) fn deliver_to_device_sockets<
185    D: DeviceBufferSpec<BC>,
186    BC: TransmitQueueBindingsContext<CC::DeviceId>,
187    CC: TransmitQueueCommon<D, BC> + DeviceSocketHandler<D, BC>,
188>(
189    core_ctx: &mut CC,
190    bindings_ctx: &mut BC,
191    device_id: &CC::DeviceId,
192    buffer: &D::TxBuffer,
193    meta: &CC::Meta,
194) {
195    buffer.with_bytes(|b| {
196        let mut handle_parsed = |bytes: &[u8]| match CC::parse_outgoing_frame(bytes, meta) {
197            Ok(sent_frame) => DeviceSocketHandler::handle_frame(
198                core_ctx,
199                bindings_ctx,
200                device_id,
201                sent_frame.into(),
202                bytes,
203            ),
204            Err(ParseSentFrameError) => {
205                trace!("failed to parse outgoing frame on {:?} ({} bytes)", device_id, bytes.len())
206            }
207        };
208
209        if let Some(bytes) = b.try_get_contiguous() {
210            handle_parsed(bytes)
211        } else {
212            handle_parsed(&b.to_flattened_vec())
213        }
214    })
215}
216
217impl EnqueueResult {
218    fn maybe_wake_tx<D, BC: TransmitQueueBindingsContext<D>>(
219        self,
220        bindings_ctx: &mut BC,
221        device_id: &D,
222    ) {
223        match self {
224            Self::QueuePreviouslyWasOccupied => (),
225            Self::QueueWasPreviouslyEmpty => bindings_ctx.wake_tx_task(device_id),
226        }
227    }
228}
229
230enum EnqueueStatus<Meta, Buffer> {
231    NotAttempted(Meta, Buffer),
232    Attempted,
233}
234
235// Extracted to a function without the generic serializer parameter to ease code
236// generation.
237fn insert_and_notify<
238    D: DeviceBufferSpec<BC>,
239    BC: TransmitQueueBindingsContext<CC::DeviceId>,
240    CC: TransmitQueueContext<D, BC> + DeviceSocketHandler<D, BC>,
241>(
242    bindings_ctx: &mut BC,
243    device_id: &CC::DeviceId,
244    inserter: Option<fifo::QueueTxInserter<'_, CC::Meta, D::TxBuffer>>,
245    meta: CC::Meta,
246    body: D::TxBuffer,
247) -> EnqueueStatus<CC::Meta, D::TxBuffer> {
248    match inserter {
249        // No TX queue so send the frame immediately.
250        None => EnqueueStatus::NotAttempted(meta, body),
251        Some(inserter) => {
252            inserter.insert(meta, body).maybe_wake_tx(bindings_ctx, device_id);
253            EnqueueStatus::Attempted
254        }
255    }
256}
257
258// Extracted to a function without the generic serializer parameter to ease code
259// generation.
260fn handle_post_enqueue<
261    D: DeviceBufferSpec<BC>,
262    BC: TransmitQueueBindingsContext<CC::DeviceId>,
263    CC: TransmitQueueContext<D, BC> + DeviceSocketHandler<D, BC>,
264>(
265    core_ctx: &mut CC,
266    bindings_ctx: &mut BC,
267    device_id: &CC::DeviceId,
268    status: EnqueueStatus<CC::Meta, D::TxBuffer>,
269) -> Result<(), DeviceSendFrameError> {
270    match status {
271        EnqueueStatus::NotAttempted(meta, body) => {
272            // TODO(https://fxbug.dev/42077654): Deliver the frame to packet
273            // sockets and to the device atomically.
274            deliver_to_device_sockets(core_ctx, bindings_ctx, device_id, &body, &meta);
275            // Send the frame while not holding the TX queue exclusively to
276            // not block concurrent senders from making progress.
277            core_ctx.send_frame(bindings_ctx, device_id, None, meta, body)
278        }
279        EnqueueStatus::Attempted => Ok(()),
280    }
281}
282
283impl<
284    D: DeviceBufferSpec<BC>,
285    BC: TransmitQueueBindingsContext<CC::DeviceId>,
286    CC: TransmitQueueContext<D, BC> + DeviceSocketHandler<D, BC>,
287> TransmitQueueHandler<D, BC> for CC
288{
289    fn queue_tx_frame<S>(
290        &mut self,
291        bindings_ctx: &mut BC,
292        device_id: &CC::DeviceId,
293        mut meta: CC::Meta,
294        body: S,
295    ) -> Result<usize, TransmitQueueFrameError<S>>
296    where
297        S: NetworkSerializer,
298        S::Buffer: ReusableBuffer,
299    {
300        let (len, result) = self.with_transmit_queue_mut(
301            device_id,
302            |TransmitQueueState { allocator, queue, checksum_offload_spec }| {
303                let queue_len = queue.as_ref().map_or(0, |q| q.len());
304                let inserter = match queue {
305                    None => None,
306                    Some(q) => match q.tx_inserter() {
307                        Some(i) => Some(i),
308                        None => return Err(TransmitQueueFrameError::QueueFull(body)),
309                    },
310                };
311                let mut context = NetworkSerializationContext::new(checksum_offload_spec.clone());
312                let body = body
313                    .serialize_outer(
314                        &mut context,
315                        NoReuseBufferProvider(BufferAllocAdaptor {
316                            allocator: &mut *allocator,
317                            queue_len,
318                        }),
319                    )
320                    .map_err(|(e, serializer)| {
321                        TransmitQueueFrameError::SerializeError(ErrorAndSerializer {
322                            serializer,
323                            error: e.map_alloc(
324                                |_: <D::TxAllocator as TxBufferAllocator<D::TxBuffer>>::Error| (),
325                            ),
326                        })
327                    })?;
328                meta.set_checksum_offload_result(context.csum_offload_result());
329                let len = body.len();
330                let result = insert_and_notify::<_, _, CC>(
331                    bindings_ctx,
332                    device_id,
333                    inserter,
334                    meta,
335                    allocator.get_buffer(),
336                );
337                Ok((len, result))
338            },
339        )?;
340
341        handle_post_enqueue(self, bindings_ctx, device_id, result)
342            .map(|()| len)
343            .map_err(TransmitQueueFrameError::NoQueue)
344    }
345}
346
347/// Allocator for Tx buffers to be stored in Tx queue.
348pub trait TxBufferAllocator<Buffer> {
349    /// Error for allocating a Tx buffer.
350    type Error;
351
352    /// Allocate Tx buffer. This method is only called while the TX queue is
353    /// locked. `queue_len` is the current length of the TX queue.
354    fn alloc(&mut self, len: usize, queue_len: usize) -> Result<Buffer, Self::Error>;
355}
356
357/// Turns a `TxBufferAllocator` into a buffer allocator for `Buf<&mut [u8]>`.
358/// The caller can serialize into the returned `Buf<&mut [u8]>` and retrieve
359/// the buffer after serialization. This avoids the binary bloat by not adding
360/// serialization code for different buffer types.
361#[derive(Derivative)]
362#[derivative(Default(bound = "A: Default"))]
363pub(super) struct ByteSliceAllocator<A, B> {
364    allocator: A,
365    buffer: Option<B>,
366}
367
368impl<A, B> ByteSliceAllocator<A, B> {
369    fn new(allocator: A) -> Self {
370        Self { allocator, buffer: None }
371    }
372}
373
374impl<A: TxBufferAllocator<B>, B: AsMut<[u8]>> ByteSliceAllocator<A, B> {
375    /// Returns a borrowed slice that the caller can serialize into.
376    fn alloc_buf_slice(
377        &mut self,
378        len: usize,
379        queue_len: usize,
380    ) -> Result<Buf<&mut [u8]>, A::Error> {
381        let old = self.buffer.replace(self.allocator.alloc(len, queue_len)?);
382        debug_assert!(old.is_none());
383        // The allocator may allocate a buffer larger than requested to fulfill
384        // the minimum tx length. We cap it at `len`.
385        Ok(Buf::new(&mut self.buffer.as_mut().expect("must be set").as_mut()[..len], ..))
386    }
387
388    /// This is intended to obtain the ownership of the buffer and store it
389    /// into the tx queue.
390    fn get_buffer(&mut self) -> B {
391        self.buffer.take().expect("must be allocated")
392    }
393}
394
395struct BufferAllocAdaptor<'a, A, B> {
396    allocator: &'a mut ByteSliceAllocator<A, B>,
397    queue_len: usize,
398}
399
400impl<'a, A, B> packet::BufferAlloc<Buf<&'a mut [u8]>> for BufferAllocAdaptor<'a, A, B>
401where
402    B: AsMut<[u8]>,
403    A: TxBufferAllocator<B>,
404{
405    type Error = A::Error;
406
407    fn alloc(self, len: usize) -> Result<Buf<&'a mut [u8]>, Self::Error> {
408        self.allocator.alloc_buf_slice(len, self.queue_len)
409    }
410}
411
412/// An allocator of [`Buf<Vec<u8>>`] .
413#[derive(Default)]
414pub struct BufVecU8Allocator;
415
416impl TxBufferAllocator<Buf<Vec<u8>>> for BufVecU8Allocator {
417    type Error = Never;
418
419    fn alloc(&mut self, len: usize, _qlen: usize) -> Result<Buf<Vec<u8>>, Self::Error> {
420        new_buf_vec(len)
421    }
422}
423
424#[cfg(test)]
425mod tests {
426    use super::*;
427
428    use alloc::vec;
429
430    use assert_matches::assert_matches;
431    use net_declare::net_mac;
432    use net_types::ethernet::Mac;
433    use netstack3_base::testutil::{
434        FakeBindingsCtx, FakeCoreCtx, FakeLinkDevice, FakeLinkDeviceId, FakeTxMetadata,
435    };
436
437    impl<BT> DeviceBufferSpec<BT> for FakeLinkDevice {
438        type TxBuffer = packet::Buf<Vec<u8>>;
439        type TxAllocator = BufVecU8Allocator;
440    }
441    use netstack3_base::{
442        ContextPair, CounterContext, CtxPair, ResourceCounterContext, WorkQueueReport,
443    };
444    use test_case::test_case;
445
446    use crate::DeviceCounters;
447    use crate::internal::queue::api::TransmitQueueApi;
448    use crate::internal::queue::{BatchSize, MAX_TX_QUEUED_LEN};
449    use crate::internal::socket::{EthernetFrame, Frame};
450
451    #[derive(Default)]
452    struct FakeTxQueueState {
453        queue: TransmitQueueState<FakeTxMetadata, Buf<Vec<u8>>, BufVecU8Allocator>,
454        transmitted_packets: Vec<(Buf<Vec<u8>>, Option<DequeueContext>)>,
455        no_buffers: bool,
456        stack_wide_device_counters: DeviceCounters,
457        per_device_counters: DeviceCounters,
458    }
459
460    #[derive(Default)]
461    struct FakeTxQueueBindingsCtxState {
462        woken_tx_tasks: Vec<FakeLinkDeviceId>,
463        delivered_to_sockets: Vec<Frame<Vec<u8>>>,
464    }
465
466    type FakeCoreCtxImpl = FakeCoreCtx<FakeTxQueueState, FakeTxMetadata, FakeLinkDeviceId>;
467    type FakeBindingsCtxImpl = FakeBindingsCtx<(), (), FakeTxQueueBindingsCtxState, ()>;
468
469    impl TransmitQueueBindingsContext<FakeLinkDeviceId> for FakeBindingsCtxImpl {
470        fn wake_tx_task(&mut self, device_id: &FakeLinkDeviceId) {
471            self.state.woken_tx_tasks.push(device_id.clone())
472        }
473    }
474
475    const SRC_MAC: Mac = net_mac!("AA:BB:CC:DD:EE:FF");
476    const DEST_MAC: Mac = net_mac!("FF:EE:DD:CC:BB:AA");
477
478    #[derive(Copy, Clone, Debug, Eq, PartialEq)]
479    struct DequeueContext;
480
481    impl TransmitQueueCommon<FakeLinkDevice, FakeBindingsCtxImpl> for FakeCoreCtxImpl {
482        type DequeueContext = DequeueContext;
483        type Meta = FakeTxMetadata;
484
485        fn parse_outgoing_frame<'a, 'b>(
486            buf: &'a [u8],
487            _meta: &'b Self::Meta,
488        ) -> Result<SentFrame<&'a [u8]>, ParseSentFrameError> {
489            Ok(fake_sent_ethernet_with_body(buf))
490        }
491    }
492
493    fn fake_sent_ethernet_with_body<B>(body: B) -> SentFrame<B> {
494        SentFrame::Ethernet(EthernetFrame {
495            src_mac: SRC_MAC,
496            dst_mac: DEST_MAC,
497            ethertype: None,
498            body_offset: 0,
499            body,
500        })
501    }
502
503    /// A trait providing a shortcut to instantiate a [`TransmitQueueApi`] from a context.
504    trait TransmitQueueApiExt: ContextPair + Sized {
505        fn transmit_queue_api<D>(&mut self) -> TransmitQueueApi<D, &mut Self> {
506            TransmitQueueApi::new(self)
507        }
508    }
509
510    impl<O> TransmitQueueApiExt for O where O: ContextPair + Sized {}
511
512    impl TransmitQueueContext<FakeLinkDevice, FakeBindingsCtxImpl> for FakeCoreCtxImpl {
513        fn with_transmit_queue<
514            O,
515            F: FnOnce(&TransmitQueueState<FakeTxMetadata, Buf<Vec<u8>>, BufVecU8Allocator>) -> O,
516        >(
517            &mut self,
518            &FakeLinkDeviceId: &FakeLinkDeviceId,
519            cb: F,
520        ) -> O {
521            cb(&self.state.queue)
522        }
523
524        fn with_transmit_queue_mut<
525            O,
526            F: FnOnce(&mut TransmitQueueState<FakeTxMetadata, Buf<Vec<u8>>, BufVecU8Allocator>) -> O,
527        >(
528            &mut self,
529            &FakeLinkDeviceId: &FakeLinkDeviceId,
530            cb: F,
531        ) -> O {
532            cb(&mut self.state.queue)
533        }
534
535        fn send_frame(
536            &mut self,
537            _bindings_ctx: &mut FakeBindingsCtxImpl,
538            &FakeLinkDeviceId: &FakeLinkDeviceId,
539            dequeue_context: Option<&mut DequeueContext>,
540            _meta: FakeTxMetadata,
541            buf: Buf<Vec<u8>>,
542        ) -> Result<(), DeviceSendFrameError> {
543            let FakeTxQueueState { transmitted_packets, no_buffers, .. } = &mut self.state;
544            if *no_buffers {
545                Err(DeviceSendFrameError::NoBuffers)
546            } else {
547                Ok(transmitted_packets.push((buf, dequeue_context.map(|c| *c))))
548            }
549        }
550    }
551
552    impl ResourceCounterContext<FakeLinkDeviceId, DeviceCounters> for FakeCoreCtxImpl {
553        fn per_resource_counters<'a>(
554            &'a self,
555            _resource: &'a FakeLinkDeviceId,
556        ) -> &'a DeviceCounters {
557            &self.state.per_device_counters
558        }
559    }
560
561    impl CounterContext<DeviceCounters> for FakeCoreCtxImpl {
562        fn counters(&self) -> &DeviceCounters {
563            &self.state.stack_wide_device_counters
564        }
565    }
566
567    impl TransmitDequeueContext<FakeLinkDevice, FakeBindingsCtxImpl> for FakeCoreCtxImpl {
568        type TransmitQueueCtx<'a> = Self;
569
570        fn with_dequed_packets_and_tx_queue_ctx<
571            O,
572            F: FnOnce(
573                &mut DequeueState<FakeTxMetadata, Buf<Vec<u8>>>,
574                &mut Self::TransmitQueueCtx<'_>,
575            ) -> O,
576        >(
577            &mut self,
578            &FakeLinkDeviceId: &FakeLinkDeviceId,
579            cb: F,
580        ) -> O {
581            cb(&mut DequeueState::default(), self)
582        }
583    }
584
585    impl DeviceSocketHandler<FakeLinkDevice, FakeBindingsCtxImpl> for FakeCoreCtxImpl {
586        fn handle_frame(
587            &mut self,
588            bindings_ctx: &mut FakeBindingsCtxImpl,
589            _device: &Self::DeviceId,
590            frame: Frame<&[u8]>,
591            _whole_frame: &[u8],
592        ) {
593            bindings_ctx.state.delivered_to_sockets.push(frame.cloned())
594        }
595    }
596
597    #[test]
598    fn noqueue() {
599        let mut ctx = CtxPair::with_core_ctx(FakeCoreCtxImpl::default());
600
601        let body = Buf::new(vec![0], ..);
602
603        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
604        assert_eq!(
605            TransmitQueueHandler::queue_tx_frame(
606                core_ctx,
607                bindings_ctx,
608                &FakeLinkDeviceId,
609                FakeTxMetadata,
610                body.clone(),
611            ),
612            Ok(body.len())
613        );
614        let FakeTxQueueBindingsCtxState { woken_tx_tasks, delivered_to_sockets } =
615            &bindings_ctx.state;
616        assert_matches!(&woken_tx_tasks[..], &[]);
617        assert_eq!(
618            delivered_to_sockets,
619            &[Frame::Sent(fake_sent_ethernet_with_body(body.as_ref().into()))]
620        );
621        assert_eq!(core::mem::take(&mut core_ctx.state.transmitted_packets), [(body, None)]);
622
623        // Should not have any frames waiting to be transmitted since we have no
624        // queue.
625        assert_eq!(
626            ctx.transmit_queue_api().transmit_queued_frames(
627                &FakeLinkDeviceId,
628                BatchSize::default(),
629                &mut DequeueContext,
630            ),
631            Ok(WorkQueueReport::AllDone),
632        );
633
634        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
635        assert_matches!(&bindings_ctx.state.woken_tx_tasks[..], &[]);
636        assert_eq!(core::mem::take(&mut core_ctx.state.transmitted_packets), []);
637    }
638
639    #[test_case(BatchSize::MAX)]
640    #[test_case(BatchSize::MAX/2)]
641    fn fifo_queue_and_dequeue(batch_size: usize) {
642        let mut ctx = CtxPair::with_core_ctx(FakeCoreCtxImpl::default());
643
644        ctx.transmit_queue_api()
645            .set_configuration(&FakeLinkDeviceId, TransmitQueueConfiguration::Fifo);
646
647        for _ in 0..2 {
648            let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
649            for i in 0..MAX_TX_QUEUED_LEN {
650                let body = Buf::new(vec![i as u8], ..);
651                assert_eq!(
652                    TransmitQueueHandler::queue_tx_frame(
653                        core_ctx,
654                        bindings_ctx,
655                        &FakeLinkDeviceId,
656                        FakeTxMetadata,
657                        body
658                    ),
659                    Ok(1)
660                );
661                // We should only ever be woken up once when the first packet
662                // was enqueued.
663                assert_eq!(bindings_ctx.state.woken_tx_tasks, [FakeLinkDeviceId]);
664            }
665
666            let body = Buf::new(vec![131], ..);
667            assert_eq!(
668                TransmitQueueHandler::queue_tx_frame(
669                    core_ctx,
670                    bindings_ctx,
671                    &FakeLinkDeviceId,
672                    FakeTxMetadata,
673                    body.clone(),
674                ),
675                Err(TransmitQueueFrameError::QueueFull(body))
676            );
677
678            let FakeTxQueueBindingsCtxState { woken_tx_tasks, delivered_to_sockets } =
679                &mut bindings_ctx.state;
680            // We should only ever be woken up once when the first packet
681            // was enqueued.
682            assert_eq!(core::mem::take(woken_tx_tasks), [FakeLinkDeviceId]);
683            // No frames should be delivered to packet sockets before transmit.
684            assert_eq!(core::mem::take(delivered_to_sockets), &[]);
685
686            assert!(MAX_TX_QUEUED_LEN > batch_size);
687            for i in (0..(MAX_TX_QUEUED_LEN - batch_size)).step_by(batch_size) {
688                assert_eq!(
689                    ctx.transmit_queue_api().transmit_queued_frames(
690                        &FakeLinkDeviceId,
691                        BatchSize::new_saturating(batch_size),
692                        &mut DequeueContext
693                    ),
694                    Ok(WorkQueueReport::Pending),
695                );
696                assert_eq!(
697                    core::mem::take(&mut ctx.core_ctx.state.transmitted_packets),
698                    (i..i + batch_size)
699                        .map(|i| (Buf::new(vec![i as u8], ..), Some(DequeueContext)))
700                        .collect::<Vec<_>>()
701                );
702            }
703
704            assert_eq!(
705                ctx.transmit_queue_api().transmit_queued_frames(
706                    &FakeLinkDeviceId,
707                    BatchSize::new_saturating(batch_size),
708                    &mut DequeueContext
709                ),
710                Ok(WorkQueueReport::AllDone),
711            );
712
713            let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
714            assert_eq!(
715                core::mem::take(&mut core_ctx.state.transmitted_packets),
716                (batch_size * (MAX_TX_QUEUED_LEN / batch_size - 1)..MAX_TX_QUEUED_LEN)
717                    .map(|i| (Buf::new(vec![i as u8], ..), Some(DequeueContext)))
718                    .collect::<Vec<_>>()
719            );
720            // Should not have woken up the TX task since the queue should be
721            // empty.
722            let FakeTxQueueBindingsCtxState { woken_tx_tasks, delivered_to_sockets } =
723                &mut bindings_ctx.state;
724            assert_matches!(&core::mem::take(woken_tx_tasks)[..], &[]);
725
726            // The queue should now be empty so the next iteration of queueing
727            // `MAX_TX_QUEUED_FRAMES` packets should succeed.
728            assert_eq!(
729                core::mem::take(delivered_to_sockets),
730                (0..MAX_TX_QUEUED_LEN)
731                    .map(|i| Frame::Sent(fake_sent_ethernet_with_body(vec![i as u8])))
732                    .collect::<Vec<_>>()
733            );
734        }
735    }
736
737    #[test]
738    fn dequeue_error() {
739        let mut ctx = CtxPair::with_core_ctx(FakeCoreCtxImpl::default());
740
741        ctx.transmit_queue_api()
742            .set_configuration(&FakeLinkDeviceId, TransmitQueueConfiguration::Fifo);
743
744        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
745        let body = Buf::new(vec![0], ..);
746        assert_eq!(
747            TransmitQueueHandler::queue_tx_frame(
748                core_ctx,
749                bindings_ctx,
750                &FakeLinkDeviceId,
751                FakeTxMetadata,
752                body.clone(),
753            ),
754            Ok(body.len())
755        );
756        assert_eq!(core::mem::take(&mut bindings_ctx.state.woken_tx_tasks), [FakeLinkDeviceId]);
757        assert_eq!(core_ctx.state.transmitted_packets, []);
758
759        core_ctx.state.no_buffers = true;
760        assert_eq!(
761            ctx.transmit_queue_api().transmit_queued_frames(
762                &FakeLinkDeviceId,
763                BatchSize::default(),
764                &mut DequeueContext
765            ),
766            Err(DeviceSendFrameError::NoBuffers),
767        );
768        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
769        assert_eq!(core_ctx.state.transmitted_packets, []);
770        let FakeTxQueueBindingsCtxState { woken_tx_tasks, delivered_to_sockets } =
771            &bindings_ctx.state;
772        assert_matches!(&woken_tx_tasks[..], &[]);
773        // Frames were delivered to packet sockets before the device was found
774        // to not be ready.
775        assert_eq!(
776            delivered_to_sockets,
777            &[Frame::Sent(fake_sent_ethernet_with_body(body.as_ref().into()))]
778        );
779
780        core_ctx.state.no_buffers = false;
781        assert_eq!(
782            ctx.transmit_queue_api().transmit_queued_frames(
783                &FakeLinkDeviceId,
784                BatchSize::default(),
785                &mut DequeueContext
786            ),
787            Ok(WorkQueueReport::AllDone),
788        );
789        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
790        assert_matches!(&bindings_ctx.state.woken_tx_tasks[..], &[]);
791        // The packet that failed to dequeue is dropped.
792        assert_eq!(core::mem::take(&mut core_ctx.state.transmitted_packets), []);
793    }
794
795    #[test_case(true; "device no buffers")]
796    #[test_case(false; "device has buffers")]
797    fn drain_before_noqueue(no_buffers: bool) {
798        let mut ctx = CtxPair::with_core_ctx(FakeCoreCtxImpl::default());
799
800        ctx.transmit_queue_api()
801            .set_configuration(&FakeLinkDeviceId, TransmitQueueConfiguration::Fifo);
802
803        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
804        let body = Buf::new(vec![0], ..);
805        assert_eq!(
806            TransmitQueueHandler::queue_tx_frame(
807                core_ctx,
808                bindings_ctx,
809                &FakeLinkDeviceId,
810                FakeTxMetadata,
811                body.clone(),
812            ),
813            Ok(body.len())
814        );
815        assert_eq!(core::mem::take(&mut bindings_ctx.state.woken_tx_tasks), [FakeLinkDeviceId]);
816        assert_eq!(core_ctx.state.transmitted_packets, []);
817
818        core_ctx.state.no_buffers = no_buffers;
819        ctx.transmit_queue_api()
820            .set_configuration(&FakeLinkDeviceId, TransmitQueueConfiguration::None);
821
822        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
823        let FakeTxQueueBindingsCtxState { woken_tx_tasks, delivered_to_sockets } =
824            &bindings_ctx.state;
825        assert_matches!(&woken_tx_tasks[..], &[]);
826        assert_eq!(
827            delivered_to_sockets,
828            &[Frame::Sent(fake_sent_ethernet_with_body(body.as_ref().into()))]
829        );
830        if no_buffers {
831            assert_eq!(core_ctx.state.transmitted_packets, []);
832        } else {
833            assert_eq!(core::mem::take(&mut core_ctx.state.transmitted_packets), [(body, None)]);
834        }
835    }
836
837    #[test]
838    fn count() {
839        let mut ctx = CtxPair::with_core_ctx(FakeCoreCtxImpl::default());
840        assert_eq!(ctx.transmit_queue_api().count(&FakeLinkDeviceId), None);
841
842        ctx.transmit_queue_api()
843            .set_configuration(&FakeLinkDeviceId, TransmitQueueConfiguration::Fifo);
844
845        assert_eq!(ctx.transmit_queue_api().count(&FakeLinkDeviceId), Some(0));
846
847        let CtxPair { core_ctx, bindings_ctx } = &mut ctx;
848        let body = Buf::new(vec![0], ..);
849        assert_eq!(
850            TransmitQueueHandler::queue_tx_frame(
851                core_ctx,
852                bindings_ctx,
853                &FakeLinkDeviceId,
854                FakeTxMetadata,
855                body,
856            ),
857            Ok(1)
858        );
859
860        assert_eq!(ctx.transmit_queue_api().count(&FakeLinkDeviceId), Some(1));
861    }
862}
netstack3_device/queue/tx.rs

netstack3_device/queue/
tx.rs
