Skip to main content

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