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

netstack3_device/queue/
tx.rs
