netstack3_ip/multicast_forwarding/
packet_queue.rs

1// Copyright 2024 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//! Declares types and functionality related to queued multicast packets.
6
7use alloc::vec::Vec;
8use arrayvec::ArrayVec;
9use core::num::NonZeroUsize;
10use core::time::Duration;
11use derivative::Derivative;
12use lru::LruCache;
13use net_types::ip::{Ip, IpVersionMarker};
14use netstack3_base::{
15    CoreTimerContext, Inspectable, Inspector, Instant as _, LocalFrameDestination, Marks,
16    StrongDeviceIdentifier as _, WeakDeviceIdentifier,
17};
18use packet::{Buf, ParseBufferMut};
19use packet_formats::ip::IpPacket;
20use zerocopy::SplitByteSlice;
21
22use crate::IpLayerIpExt;
23use crate::internal::multicast_forwarding::{
24    MulticastForwardingBindingsContext, MulticastForwardingBindingsTypes,
25    MulticastForwardingTimerId,
26};
27use crate::multicast_forwarding::MulticastRouteKey;
28
29/// The number of packets that the stack is willing to queue for a given
30/// [`MulticastRouteKey`] while waiting for an applicable route to be installed.
31///
32/// This value is consistent with the defaults on both Netstack2 and Linux.
33pub(crate) const PACKET_QUEUE_LEN: usize = 3;
34
35/// The maximum number of pending multicast routes that can be queued.
36///
37/// The size of each entry is dominated by the packet queue (up to
38/// PACKET_QUEUE_LEN). With 1000 entries, each with 3 standard MTU packets, this
39/// limit is approximately 4.5MB.
40const MAX_PENDING_ROUTES: NonZeroUsize = NonZeroUsize::new(1000).unwrap();
41
42/// The amount of time the stack is willing to queue a packet while waiting
43/// for an applicable route to be installed.
44///
45/// This value is consistent with the defaults on both Netstack2 and Linux.
46const PENDING_ROUTE_EXPIRATION: Duration = Duration::from_secs(10);
47
48/// The minimum amount of time after a garbage-collection run across the
49/// [`MulticastForwardingPendingPackets`] table that the stack will wait before
50/// performing another garbage-collection.
51///
52/// This value is consistent with the defaults on both Netstack2 and Linux.
53const PENDING_ROUTE_GC_PERIOD: Duration = Duration::from_secs(10);
54
55/// A table of pending multicast packets that have not yet been forwarded.
56///
57/// Packets are placed in this table when, during forwarding, there is no route
58/// in the [`MulticastRouteTable`] via which to forward them. If/when such a
59/// route is installed, the packets stored here can be forwarded accordingly.
60#[derive(Derivative)]
61#[derivative(Debug(bound = ""))]
62pub struct MulticastForwardingPendingPackets<
63    I: IpLayerIpExt,
64    D: WeakDeviceIdentifier,
65    BT: MulticastForwardingBindingsTypes,
66> {
67    table: LruCache<MulticastRouteKey<I>, PacketQueue<I, D, BT>>,
68    /// Periodically triggers invocations of [`Self::run_garbage_collection`].
69    ///
70    /// All interactions with the `gc_timer` must uphold the invariant that the
71    /// timer is not scheduled if [`Self::table`] is empty.
72    ///
73    /// Note: When [`Self`] is held by [`MulticastForwardingEnabledState`], it
74    /// is lock protected, which prevents method calls on it from racing. E.g.
75    /// no overlapping calls to [`Self::try_queue_packet`], [`Self::remove`],
76    /// or [`Self::run_garbage_collection`].
77    gc_timer: BT::Timer,
78}
79
80impl<I: IpLayerIpExt, D: WeakDeviceIdentifier, BC: MulticastForwardingBindingsContext<I, D::Strong>>
81    MulticastForwardingPendingPackets<I, D, BC>
82{
83    pub(crate) fn new<CC>(bindings_ctx: &mut BC) -> Self
84    where
85        CC: CoreTimerContext<MulticastForwardingTimerId<I>, BC>,
86    {
87        Self {
88            table: LruCache::new(MAX_PENDING_ROUTES),
89            gc_timer: CC::new_timer(
90                bindings_ctx,
91                MulticastForwardingTimerId::PendingPacketsGc(IpVersionMarker::<I>::new()),
92            ),
93        }
94    }
95
96    /// Attempt to queue the packet in the pending_table.
97    ///
98    /// If the table becomes newly occupied, the GC timer is scheduled.
99    pub(crate) fn try_queue_packet<B>(
100        &mut self,
101        bindings_ctx: &mut BC,
102        key: MulticastRouteKey<I>,
103        packet: &I::Packet<B>,
104        dev: &D::Strong,
105        frame_dst: Option<LocalFrameDestination>,
106        max_fragment_len: Option<usize>,
107        marks: Marks,
108    ) -> QueuePacketOutcome
109    where
110        B: SplitByteSlice,
111    {
112        let was_empty = self.table.is_empty();
113        let outcome = if let Some(queue) = self.table.get_mut(&key) {
114            match queue
115                .try_push(|| QueuedPacket::new(dev, packet, frame_dst, max_fragment_len, marks))
116            {
117                Ok(()) => QueuePacketOutcome::QueuedInExistingQueue,
118                Err(PacketQueueFullError) => QueuePacketOutcome::ExistingQueueFull,
119            }
120        } else {
121            let mut queue = PacketQueue::new(bindings_ctx);
122            queue
123                .try_push(|| QueuedPacket::new(dev, packet, frame_dst, max_fragment_len, marks))
124                .expect("newly instantiated queue must have capacity");
125
126            let prev = self.table.put(key, queue);
127            debug_assert!(prev.is_none());
128            QueuePacketOutcome::QueuedInNewQueue
129        };
130
131        // If the table is newly non-empty, schedule the GC. The timer must not
132        // already be scheduled (given the invariants on `gc_timer`).
133        if was_empty && !self.table.is_empty() {
134            let prev = bindings_ctx.schedule_timer(PENDING_ROUTE_GC_PERIOD, &mut self.gc_timer);
135            debug_assert!(prev.is_none());
136        }
137
138        outcome
139    }
140
141    #[cfg(any(debug_assertions, test))]
142    pub(crate) fn contains(&self, key: &MulticastRouteKey<I>) -> bool {
143        self.table.contains(key)
144    }
145
146    /// Remove the key from the pending table, returning its queue of packets.
147    ///
148    /// If the table becomes newly empty, the GC timer is canceled.
149    pub(crate) fn remove(
150        &mut self,
151        key: &MulticastRouteKey<I>,
152        bindings_ctx: &mut BC,
153    ) -> Option<PacketQueue<I, D, BC>> {
154        let was_empty = self.table.is_empty();
155        let queue = self.table.pop(key);
156
157        // If the table is newly empty, cancel the GC. Note, we don't assert on
158        // the previous state of the timer, because it's possible cancelation
159        // will race with the timer firing.
160        if !was_empty && self.table.is_empty() {
161            let _: Option<BC::Instant> = bindings_ctx.cancel_timer(&mut self.gc_timer);
162        }
163
164        queue
165    }
166
167    /// Removes expired [`PacketQueue`] entries from [`Self`].
168    ///
169    /// Returns the number of packets removed as a result.
170    pub(crate) fn run_garbage_collection(&mut self, bindings_ctx: &mut BC) -> u64 {
171        let now = bindings_ctx.now();
172        let mut removed_count = 0u64;
173        let expired_keys: Vec<_> = self
174            .table
175            .iter()
176            .filter_map(
177                |(key, queue)| if queue.expires_at <= now { Some(key.clone()) } else { None },
178            )
179            .collect();
180
181        for key in expired_keys {
182            let queue = self.table.pop(&key).expect("expired key must be present");
183            // NB: "as" conversion is safe because queue_len has a maximum
184            // value of `PACKET_QUEUE_LEN`, which fits in a u64.
185            removed_count += queue.queue.len() as u64;
186        }
187
188        // If the table is still not empty, reschedule the GC. Note that we
189        // don't assert on the previous state of the timer, because it's
190        // possible that starting GC raced with a new timer being scheduled.
191        if !self.table.is_empty() {
192            let _: Option<BC::Instant> =
193                bindings_ctx.schedule_timer(PENDING_ROUTE_GC_PERIOD, &mut self.gc_timer);
194        }
195
196        removed_count
197    }
198}
199
200impl<I: IpLayerIpExt, D: WeakDeviceIdentifier, BT: MulticastForwardingBindingsTypes> Inspectable
201    for MulticastForwardingPendingPackets<I, D, BT>
202{
203    fn record<II: Inspector>(&self, inspector: &mut II) {
204        let MulticastForwardingPendingPackets { table, gc_timer: _ } = self;
205        // NB: Don't record all routes, as the size of the table may be quite
206        // large, and its contents are dictated by network traffic.
207        inspector.record_usize("NumRoutes", table.len())
208    }
209}
210
211/// Possible outcomes from calling [`MulticastForwardingPendingPackets::try_queue_packet`].
212#[derive(Debug, PartialEq)]
213pub(crate) enum QueuePacketOutcome {
214    /// The packet was successfully queued. There was no existing
215    /// [`PacketQueue`] for the given route key, so a new one was instantiated.
216    QueuedInNewQueue,
217    /// The packet was successfully queued. It was added onto an existing
218    /// [`PacketQueue`] for the given route key.
219    QueuedInExistingQueue,
220    /// The packet was not queued. There was an existing [`PacketQueue`] for the
221    /// given route key, but that queue was full.
222    ExistingQueueFull,
223}
224
225/// A queue of multicast packets that are pending the installation of a route.
226#[derive(Derivative)]
227#[derivative(Debug(bound = ""))]
228pub struct PacketQueue<I: Ip, D: WeakDeviceIdentifier, BT: MulticastForwardingBindingsTypes> {
229    queue: ArrayVec<QueuedPacket<I, D>, PACKET_QUEUE_LEN>,
230    /// The time after which the PacketQueue is allowed to be garbage collected.
231    expires_at: BT::Instant,
232}
233
234impl<I: IpLayerIpExt, D: WeakDeviceIdentifier, BC: MulticastForwardingBindingsContext<I, D::Strong>>
235    PacketQueue<I, D, BC>
236{
237    fn new(bindings_ctx: &mut BC) -> Self {
238        Self {
239            queue: Default::default(),
240            expires_at: bindings_ctx.now().panicking_add(PENDING_ROUTE_EXPIRATION),
241        }
242    }
243
244    /// Try to push a packet into the queue, returning an error when full.
245    ///
246    /// Note: the packet is taken as a builder closure, because constructing the
247    /// packet is an expensive operation (requiring a `Vec` allocation). By
248    /// taking a closure we can defer construction until we're certain the queue
249    /// has the free space to hold it.
250    fn try_push(
251        &mut self,
252        packet_builder: impl FnOnce() -> QueuedPacket<I, D>,
253    ) -> Result<(), PacketQueueFullError> {
254        if self.queue.is_full() {
255            return Err(PacketQueueFullError);
256        }
257        self.queue.push(packet_builder());
258        Ok(())
259    }
260}
261
262#[derive(Debug)]
263struct PacketQueueFullError;
264
265impl<I: Ip, D: WeakDeviceIdentifier, BT: MulticastForwardingBindingsTypes> IntoIterator
266    for PacketQueue<I, D, BT>
267{
268    type Item = QueuedPacket<I, D>;
269    type IntoIter = <ArrayVec<QueuedPacket<I, D>, PACKET_QUEUE_LEN> as IntoIterator>::IntoIter;
270    fn into_iter(self) -> Self::IntoIter {
271        let Self { queue, expires_at: _ } = self;
272        queue.into_iter()
273    }
274}
275
276/// An individual multicast packet that's queued.
277#[derive(Debug, PartialEq)]
278pub struct QueuedPacket<I: Ip, D: WeakDeviceIdentifier> {
279    /// The device on which the packet arrived.
280    pub(crate) device: D,
281    /// The packet.
282    pub(crate) packet: ValidIpPacketBuf<I>,
283    /// The link layer (L2) destination that the packet was sent to, or `None`
284    /// if the packet arrived above the link layer (e.g. a Pure IP device).
285    pub(crate) frame_dst: Option<LocalFrameDestination>,
286    /// The maximum size fragment that was used to reassemble this packet when
287    /// it ingressed the stack. None if IP reassembly was not performed.
288    pub(crate) max_fragment_len: Option<usize>,
289    /// Marks attached to the packet on ingress.
290    pub(crate) marks: Marks,
291}
292
293impl<I: IpLayerIpExt, D: WeakDeviceIdentifier> QueuedPacket<I, D> {
294    fn new<B: SplitByteSlice>(
295        device: &D::Strong,
296        packet: &I::Packet<B>,
297        frame_dst: Option<LocalFrameDestination>,
298        max_fragment_len: Option<usize>,
299        marks: Marks,
300    ) -> Self {
301        QueuedPacket {
302            device: device.downgrade(),
303            packet: ValidIpPacketBuf::new(packet),
304            frame_dst,
305            max_fragment_len,
306            marks,
307        }
308    }
309}
310
311/// A buffer containing a known-to-be valid IP packet.
312///
313/// The only constructor of this type takes an `I::Packet`, which is already
314/// parsed & validated.
315#[derive(Clone, Debug, PartialEq)]
316pub(crate) struct ValidIpPacketBuf<I: Ip> {
317    buffer: Buf<Vec<u8>>,
318    _version_marker: IpVersionMarker<I>,
319}
320
321impl<I: IpLayerIpExt> ValidIpPacketBuf<I> {
322    fn new<B: SplitByteSlice>(packet: &I::Packet<B>) -> Self {
323        Self { buffer: Buf::new(packet.to_vec(), ..), _version_marker: Default::default() }
324    }
325
326    /// Parses the internal buffer into a mutable IP Packet.
327    ///
328    /// # Panics
329    ///
330    /// This function panics if called multiple times. Parsing moves the cursor
331    /// in the underlying buffer from the start of the IP header to the start
332    /// of the IP body.
333    pub(crate) fn parse_ip_packet_mut(&mut self) -> I::Packet<&mut [u8]> {
334        // NB: Safe to unwrap here because the buffer is known to be valid.
335        self.buffer.parse_mut().unwrap()
336    }
337
338    pub(crate) fn into_inner(self) -> Buf<Vec<u8>> {
339        let Self { buffer, _version_marker } = self;
340        buffer
341    }
342}
343
344#[cfg(test)]
345mod tests {
346    use super::*;
347
348    use assert_matches::assert_matches;
349    use ip_test_macro::ip_test;
350    use netstack3_base::testutil::{
351        FakeInstant, FakeTimerCtxExt, FakeWeakDeviceId, MultipleDevicesId,
352    };
353    use netstack3_base::{
354        CounterContext, InstantContext, MarkDomain, StrongDeviceIdentifier, TimerContext,
355    };
356    use packet::ParseBuffer;
357    use static_assertions::const_assert;
358    use test_case::test_case;
359
360    use crate::internal::multicast_forwarding;
361    use crate::internal::multicast_forwarding::counters::MulticastForwardingCounters;
362    use crate::internal::multicast_forwarding::testutil::{
363        FakeBindingsCtx, FakeCoreCtx, TestIpExt,
364    };
365
366    #[ip_test(I)]
367    #[test_case(None, None; "no_metadata")]
368    #[test_case(Some(LocalFrameDestination::Multicast), None; "some_frame_dst")]
369    #[test_case(None, Some(1400); "some_max_fragment_len")]
370    #[test_case(Some(LocalFrameDestination::Multicast), Some(1400); "some_all")]
371    fn queue_packet<I: TestIpExt>(
372        frame_dst: Option<LocalFrameDestination>,
373        max_fragment_len: Option<usize>,
374    ) {
375        const DEV: MultipleDevicesId = MultipleDevicesId::A;
376        let key1 = MulticastRouteKey::new(I::SRC1, I::DST1).unwrap();
377        let key2 = MulticastRouteKey::new(I::SRC2, I::DST2).unwrap();
378        let key3 = MulticastRouteKey::new(I::SRC1, I::DST2).unwrap();
379
380        // NB: technically the packet's addresses only match `key1`, but for the
381        // sake of this test that doesn't cause problems.
382        let buf = multicast_forwarding::testutil::new_ip_packet_buf::<I>(I::SRC1, I::DST1);
383        let mut buf_ref = buf.as_ref();
384        let packet = buf_ref.parse::<I::Packet<_>>().expect("parse should succeed");
385
386        let mut bindings_ctx = FakeBindingsCtx::<I, MultipleDevicesId>::default();
387
388        let mut pending_table =
389            MulticastForwardingPendingPackets::<
390                I,
391                <MultipleDevicesId as StrongDeviceIdentifier>::Weak,
392                _,
393            >::new::<FakeCoreCtx<I, MultipleDevicesId>>(&mut bindings_ctx);
394
395        // Use a non-default value so that the `expected_packet` comparison
396        // below actually exercises round-tripping `marks` through the queue.
397        let marks = Marks::new([(MarkDomain::Mark1, 100)]);
398
399        // The first packet gets a new queue.
400        assert_eq!(
401            pending_table.try_queue_packet(
402                &mut bindings_ctx,
403                key1.clone(),
404                &packet,
405                &DEV,
406                frame_dst,
407                max_fragment_len,
408                marks,
409            ),
410            QueuePacketOutcome::QueuedInNewQueue
411        );
412        // The second - Nth packets uses the existing queue.
413        for _ in 1..PACKET_QUEUE_LEN {
414            assert_eq!(
415                pending_table.try_queue_packet(
416                    &mut bindings_ctx,
417                    key1.clone(),
418                    &packet,
419                    &DEV,
420                    frame_dst,
421                    max_fragment_len,
422                    marks,
423                ),
424                QueuePacketOutcome::QueuedInExistingQueue
425            );
426        }
427        // The Nth +1 packet is rejected.
428        assert_eq!(
429            pending_table.try_queue_packet(
430                &mut bindings_ctx,
431                key1.clone(),
432                &packet,
433                &DEV,
434                frame_dst,
435                max_fragment_len,
436                marks,
437            ),
438            QueuePacketOutcome::ExistingQueueFull
439        );
440
441        // A packet with a different key gets a new queue.
442        assert_eq!(
443            pending_table.try_queue_packet(
444                &mut bindings_ctx,
445                key2.clone(),
446                &packet,
447                &DEV,
448                frame_dst,
449                max_fragment_len,
450                marks,
451            ),
452            QueuePacketOutcome::QueuedInNewQueue
453        );
454
455        // Based on the calls above, `key1` should have a full queue, `key2`
456        // should have a queue with only 1 packet, and `key3` shouldn't have
457        // a queue.
458        let expected_packet = QueuedPacket::new(&DEV, &packet, frame_dst, max_fragment_len, marks);
459        let queue =
460            pending_table.remove(&key1, &mut bindings_ctx).expect("key1 should have a queue");
461        assert_eq!(queue.queue.len(), PACKET_QUEUE_LEN);
462        for packet in queue.queue.as_slice() {
463            assert_eq!(packet, &expected_packet);
464        }
465
466        let queue =
467            pending_table.remove(&key2, &mut bindings_ctx).expect("key2 should have a queue");
468        let packet = assert_matches!(&queue.queue[..], [p] => p);
469        assert_eq!(packet, &expected_packet);
470
471        assert_matches!(pending_table.remove(&key3, &mut bindings_ctx), None);
472    }
473
474    /// Helper to observe the next scheduled GC for the core_ctx pending table.
475    fn next_gc_time<I: TestIpExt>(
476        core_ctx: &mut FakeCoreCtx<I, MultipleDevicesId>,
477        bindings_ctx: &mut FakeBindingsCtx<I, MultipleDevicesId>,
478    ) -> Option<FakeInstant> {
479        multicast_forwarding::testutil::with_pending_table(core_ctx, |pending_table| {
480            bindings_ctx.scheduled_instant(&mut pending_table.gc_timer)
481        })
482    }
483
484    /// Helper to queue packet in the core_ctx pending table.
485    fn try_queue_packet<I: TestIpExt>(
486        core_ctx: &mut FakeCoreCtx<I, MultipleDevicesId>,
487        bindings_ctx: &mut FakeBindingsCtx<I, MultipleDevicesId>,
488        key: MulticastRouteKey<I>,
489        dev: &MultipleDevicesId,
490        frame_dst: Option<LocalFrameDestination>,
491        max_fragment_len: Option<usize>,
492    ) -> QueuePacketOutcome {
493        let buf =
494            multicast_forwarding::testutil::new_ip_packet_buf::<I>(key.src_addr(), key.dst_addr());
495        let mut buf_ref = buf.as_ref();
496        let packet = buf_ref.parse::<I::Packet<_>>().expect("parse should succeed");
497        multicast_forwarding::testutil::with_pending_table(core_ctx, |pending_table| {
498            pending_table.try_queue_packet(
499                bindings_ctx,
500                key,
501                &packet,
502                dev,
503                frame_dst,
504                max_fragment_len,
505                Marks::default(),
506            )
507        })
508    }
509
510    /// Helper to remove a packet queue in the core_ctx pending table.
511    fn remove_packet_queue<I: TestIpExt>(
512        core_ctx: &mut FakeCoreCtx<I, MultipleDevicesId>,
513        bindings_ctx: &mut FakeBindingsCtx<I, MultipleDevicesId>,
514        key: &MulticastRouteKey<I>,
515    ) -> Option<
516        PacketQueue<I, FakeWeakDeviceId<MultipleDevicesId>, FakeBindingsCtx<I, MultipleDevicesId>>,
517    > {
518        multicast_forwarding::testutil::with_pending_table(core_ctx, |pending_table| {
519            pending_table.remove(key, bindings_ctx)
520        })
521    }
522
523    /// Helper to trigger the GC.
524    fn run_gc<I: TestIpExt>(
525        core_ctx: &mut FakeCoreCtx<I, MultipleDevicesId>,
526        bindings_ctx: &mut FakeBindingsCtx<I, MultipleDevicesId>,
527    ) {
528        assert_matches!(
529            &bindings_ctx.trigger_timers_until_instant(bindings_ctx.now(), core_ctx)[..],
530            [MulticastForwardingTimerId::PendingPacketsGc(_)]
531        );
532    }
533
534    #[ip_test(I)]
535    fn garbage_collection<I: TestIpExt>() {
536        const DEV: MultipleDevicesId = MultipleDevicesId::A;
537        const FRAME_DST: Option<LocalFrameDestination> = None;
538        const MAX_FRAGMENT_LEN: Option<usize> = None;
539        let key1 = MulticastRouteKey::<I>::new(I::SRC1, I::DST1).unwrap();
540        let key2 = MulticastRouteKey::<I>::new(I::SRC2, I::DST2).unwrap();
541
542        let mut api = multicast_forwarding::testutil::new_api();
543        assert!(api.enable());
544        let (core_ctx, bindings_ctx) = api.contexts();
545
546        // NB: As written, the test requires that
547        //  1. `PENDING_ROUTE_GC_PERIOD` >= `PENDING_ROUTE_EXPIRATION`, and
548        //  2. `PENDING_ROUTE_EXPIRATION > 0`.
549        // If the values are ever changed such that that is not true, the test
550        // will need to be re-written.
551        const_assert!(PENDING_ROUTE_GC_PERIOD.checked_sub(PENDING_ROUTE_EXPIRATION).is_some());
552        const_assert!(!PENDING_ROUTE_EXPIRATION.is_zero());
553
554        // The GC shouldn't be scheduled with an empty table.
555        assert!(next_gc_time(core_ctx, bindings_ctx).is_none());
556        let counters: &MulticastForwardingCounters<I> = core_ctx.counters();
557        assert_eq!(counters.pending_table_gc.get(), 0);
558        assert_eq!(counters.pending_packet_drops_gc.get(), 0);
559
560        // Queue a packet, and expect the GC to be scheduled.
561        let expected_first_gc = bindings_ctx.now() + PENDING_ROUTE_GC_PERIOD;
562        assert_eq!(
563            try_queue_packet(
564                core_ctx,
565                bindings_ctx,
566                key1.clone(),
567                &DEV,
568                FRAME_DST,
569                MAX_FRAGMENT_LEN
570            ),
571            QueuePacketOutcome::QueuedInNewQueue
572        );
573        assert_eq!(next_gc_time(core_ctx, bindings_ctx), Some(expected_first_gc));
574
575        // Sleep until we're ready to GC, and then queue a second packet under a
576        // new key. Expect that the GC timer is still scheduled for the original
577        // instant.
578        bindings_ctx.timers.instant.sleep(PENDING_ROUTE_GC_PERIOD);
579        assert_eq!(
580            try_queue_packet(
581                core_ctx,
582                bindings_ctx,
583                key2.clone(),
584                &DEV,
585                FRAME_DST,
586                MAX_FRAGMENT_LEN
587            ),
588            QueuePacketOutcome::QueuedInNewQueue
589        );
590        assert_eq!(next_gc_time(core_ctx, bindings_ctx), Some(expected_first_gc));
591
592        // Run the GC, and verify that it was rescheduled after the fact
593        // (because `key2` still exists in the table).
594        run_gc(core_ctx, bindings_ctx);
595        let expected_second_gc = bindings_ctx.timers.instant.now() + PENDING_ROUTE_GC_PERIOD;
596        assert_eq!(next_gc_time(core_ctx, bindings_ctx), Some(expected_second_gc));
597
598        // Verify that `key1` was removed, but `key2` remains.
599        let counters: &MulticastForwardingCounters<I> = core_ctx.counters();
600        assert_eq!(counters.pending_table_gc.get(), 1);
601        assert_eq!(counters.pending_packet_drops_gc.get(), 1);
602        assert_matches!(remove_packet_queue(core_ctx, bindings_ctx, &key1), None);
603        assert_matches!(remove_packet_queue(core_ctx, bindings_ctx, &key2), Some(_));
604
605        // Now that we've explicitly removed `key2`, the table is empty and the
606        // GC should have been canceled.
607        assert!(next_gc_time(core_ctx, bindings_ctx).is_none());
608
609        // Finally, verify that if the GC clears the table, it doesn't
610        // reschedule itself.
611        assert_eq!(
612            try_queue_packet(
613                core_ctx,
614                bindings_ctx,
615                key1.clone(),
616                &DEV,
617                FRAME_DST,
618                MAX_FRAGMENT_LEN
619            ),
620            QueuePacketOutcome::QueuedInNewQueue
621        );
622        assert_eq!(next_gc_time(core_ctx, bindings_ctx), Some(expected_second_gc));
623        bindings_ctx.timers.instant.sleep(PENDING_ROUTE_GC_PERIOD);
624        run_gc(core_ctx, bindings_ctx);
625        let counters: &MulticastForwardingCounters<I> = core_ctx.counters();
626        assert_eq!(counters.pending_table_gc.get(), 2);
627        assert_eq!(counters.pending_packet_drops_gc.get(), 2);
628        assert_matches!(remove_packet_queue(core_ctx, bindings_ctx, &key1), None);
629        assert!(next_gc_time(core_ctx, bindings_ctx).is_none());
630    }
631}
netstack3_ip/multicast_forwarding/packet_queue.rs

netstack3_ip/multicast_forwarding/
packet_queue.rs
