1use 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
28pub(crate) const PACKET_QUEUE_LEN: usize = 3;
33
34const MAX_PENDING_ROUTES: usize = 1000;
40
41const PENDING_ROUTE_EXPIRATION: Duration = Duration::from_secs(10);
46
47const PENDING_ROUTE_GC_PERIOD: Duration = Duration::from_secs(10);
53
54#[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 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 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 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 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 !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 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 removed_count += queue.queue.len() as u64;
182 }
183
184 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 inspector.record_usize("NumRoutes", table.len())
204 }
205}
206
207#[derive(Debug, PartialEq)]
209pub(crate) enum QueuePacketOutcome {
210 QueuedInNewQueue,
213 QueuedInExistingQueue,
216 ExistingQueueFull,
219}
220
221#[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 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 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#[derive(Debug, PartialEq)]
274pub struct QueuedPacket<I: Ip, D: WeakDeviceIdentifier> {
275 pub(crate) device: D,
277 pub(crate) packet: ValidIpPacketBuf<I>,
279 pub(crate) frame_dst: Option<LocalFrameDestination>,
282 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#[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 pub(crate) fn parse_ip_packet_mut(&mut self) -> I::Packet<&mut [u8]> {
326 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 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 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 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 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 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 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 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 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 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 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 const_assert!(PENDING_ROUTE_GC_PERIOD.checked_sub(PENDING_ROUTE_EXPIRATION).is_some());
533 const_assert!(!PENDING_ROUTE_EXPIRATION.is_zero());
534
535 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 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 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_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 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 assert!(next_gc_time(core_ctx, bindings_ctx).is_none());
589
590 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}