Add per-member multicast source filtering

.github/buildomat/jobs/test.sh

@@ -95,6 +95,9 @@ pfexec /input/xde/work/test/multicast_multi_sub --nocapture --test-threads=1
 pfexec chmod +x /input/xde/work/test/multicast_validation
 pfexec /input/xde/work/test/multicast_validation --nocapture --test-threads=1
 
+pfexec chmod +x /input/xde/work/test/multicast_source_filter
+pfexec /input/xde/work/test/multicast_source_filter --nocapture --test-threads=1
+
 banner "teardown"
 # Ensure full driver teardown is exercised after tests complete
 pfexec rem_drv xde

.github/buildomat/jobs/xde.sh

@@ -17,6 +17,7 @@
 #:   "=/work/test/multicast_rx",
 #:   "=/work/test/multicast_multi_sub",
 #:   "=/work/test/multicast_validation",
+#:   "=/work/test/multicast_source_filter",
 #:   "=/work/xde.conf",
 #: ]
 #:
@@ -134,8 +135,14 @@ multicast_validation_test=$(
     cargo build -q --test multicast_validation --message-format=json |\
     jq -r "select(.profile.test == true) | .filenames[]"
 )
+cargo build --test multicast_source_filter
+multicast_source_filter_test=$(
+    cargo build -q --test multicast_source_filter --message-format=json |\
+    jq -r "select(.profile.test == true) | .filenames[]"
+)
 mkdir -p /work/test
 cp $loopback_test /work/test/loopback
 cp $multicast_rx_test /work/test/multicast_rx
 cp $multicast_multi_sub_test /work/test/multicast_multi_sub
 cp $multicast_validation_test /work/test/multicast_validation
+cp $multicast_source_filter_test /work/test/multicast_source_filter

bench/Cargo.toml

@@ -41,3 +41,4 @@ harness = false
 [[bench]]
 name = "xde"
 harness = false
+

bench/benches/userland.rs

@@ -8,6 +8,7 @@
 
 use criterion::BenchmarkId;
 use criterion::Criterion;
+use criterion::Throughput;
 use criterion::criterion_group;
 use criterion::criterion_main;
 use opte::engine::packet::Packet;
@@ -24,6 +25,11 @@ use opte_bench::packet::TestCase;
 use opte_bench::packet::ULP_FAST_PATH;
 use opte_bench::packet::ULP_SLOW_PATH;
 use opte_test_utils::*;
+use oxide_vpc::api::IpAddr;
+use oxide_vpc::api::Ipv4Addr;
+use oxide_vpc::api::Ipv6Addr;
+use oxide_vpc::api::SourceFilter;
+use std::collections::BTreeSet;
 use std::hint::black_box;
 
 // Top level runner. Specifies packet classes.
@@ -218,7 +224,111 @@ pub fn test_handle<M: MeasurementInfo + 'static>(
     );
 }
 
-criterion_group!(wall, parse_and_process);
+/// Generate a source IP address for filter testing (10.0.0.x).
+fn make_src_v4(i: u32) -> IpAddr {
+    IpAddr::Ip4(Ipv4Addr::from(0x0a000000u32 + i))
+}
+
+/// Generate a source IP address for filter testing (fd00::x).
+fn make_src_v6(i: u32) -> IpAddr {
+    let mut bytes = [0u8; 16];
+    bytes[0..4].copy_from_slice(&[0xfd, 0x00, 0x00, 0x00]);
+    bytes[12..16].copy_from_slice(&i.to_be_bytes());
+    IpAddr::Ip6(Ipv6Addr::from(bytes))
+}
+
+/// Benchmark [`SourceFilter::allows`] for various filter configurations.
+fn source_filter_allows(c: &mut Criterion) {
+    let mut group = c.benchmark_group("source_filter/allows");
+    group.throughput(Throughput::Elements(1));
+
+    let src_v4 = make_src_v4(100); // Not in any source list
+    let src_v6 = make_src_v6(100);
+
+    // Fast path: EXCLUDE() with empty sources (*, G)
+    let filter_any = SourceFilter::default();
+    group.bench_function("exclude_empty_v4", |b| {
+        b.iter(|| black_box(filter_any.allows(black_box(src_v4))))
+    });
+    group.bench_function("exclude_empty_v6", |b| {
+        b.iter(|| black_box(filter_any.allows(black_box(src_v6))))
+    });
+
+    // EXCLUDE with sources: "Miss" case where source is not in exclusion list
+    for size in [1, 5, 10, 50, 100] {
+        let sources_v4: BTreeSet<_> = (0..size).map(make_src_v4).collect();
+        let filter_v4 = SourceFilter::Exclude(sources_v4);
+        group.bench_with_input(
+            BenchmarkId::new("exclude_miss_v4", size),
+            &filter_v4,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_v4)))),
+        );
+        let src_in_list_v4 = make_src_v4(0);
+        group.bench_with_input(
+            BenchmarkId::new("exclude_hit_v4", size),
+            &filter_v4,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_in_list_v4)))),
+        );
+
+        let sources_v6: BTreeSet<_> = (0..size).map(make_src_v6).collect();
+        let filter_v6 = SourceFilter::Exclude(sources_v6);
+        group.bench_with_input(
+            BenchmarkId::new("exclude_miss_v6", size),
+            &filter_v6,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_v6)))),
+        );
+        let src_in_list_v6 = make_src_v6(0);
+        group.bench_with_input(
+            BenchmarkId::new("exclude_hit_v6", size),
+            &filter_v6,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_in_list_v6)))),
+        );
+    }
+
+    // INCLUDE with sources: "Hit" case where source is in inclusion list
+    for size in [1, 5, 10, 50, 100] {
+        let sources_v4: BTreeSet<_> = (0..size).map(make_src_v4).collect();
+        let filter_v4 = SourceFilter::Include(sources_v4);
+        let src_in_list_v4 = make_src_v4(0);
+        group.bench_with_input(
+            BenchmarkId::new("include_hit_v4", size),
+            &filter_v4,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_in_list_v4)))),
+        );
+        group.bench_with_input(
+            BenchmarkId::new("include_miss_v4", size),
+            &filter_v4,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_v4)))),
+        );
+
+        let sources_v6: BTreeSet<_> = (0..size).map(make_src_v6).collect();
+        let filter_v6 = SourceFilter::Include(sources_v6);
+        let src_in_list_v6 = make_src_v6(0);
+        group.bench_with_input(
+            BenchmarkId::new("include_hit_v6", size),
+            &filter_v6,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_in_list_v6)))),
+        );
+        group.bench_with_input(
+            BenchmarkId::new("include_miss_v6", size),
+            &filter_v6,
+            |b, f| b.iter(|| black_box(f.allows(black_box(src_v6)))),
+        );
+    }
+
+    // INCLUDE() empty, rejecting all
+    let filter_none = SourceFilter::Include(BTreeSet::new());
+    group.bench_function("include_empty_v4", |b| {
+        b.iter(|| black_box(filter_none.allows(black_box(src_v4))))
+    });
+    group.bench_function("include_empty_v6", |b| {
+        b.iter(|| black_box(filter_none.allows(black_box(src_v6))))
+    });
+
+    group.finish();
+}
+
+criterion_group!(wall, parse_and_process, source_filter_allows);
 criterion_group!(
     name = alloc;
     config = new_crit(Allocs);

bin/opteadm/src/bin/opteadm.rs

@@ -43,6 +43,7 @@ use oxide_vpc::api::FirewallRule;
 use oxide_vpc::api::IpCfg;
 use oxide_vpc::api::Ipv4Cfg;
 use oxide_vpc::api::Ipv6Cfg;
+use oxide_vpc::api::McastForwardingNextHop;
 use oxide_vpc::api::McastSubscribeReq;
 use oxide_vpc::api::McastUnsubscribeAllReq;
 use oxide_vpc::api::McastUnsubscribeReq;
@@ -64,8 +65,10 @@ use oxide_vpc::api::SetMcast2PhysReq;
 use oxide_vpc::api::SetMcastForwardingReq;
 use oxide_vpc::api::SetVirt2BoundaryReq;
 use oxide_vpc::api::SetVirt2PhysReq;
+use oxide_vpc::api::SourceFilter;
 use oxide_vpc::api::TunnelEndpoint;
 use oxide_vpc::api::VpcCfg;
+use oxide_vpc::print::print_m2p;
 use oxide_vpc::print::print_mcast_fwd;
 use oxide_vpc::print::print_mcast_subs;
 use oxide_vpc::print::print_v2b;
@@ -308,6 +311,9 @@ enum Command {
     /// Dump multicast subscriptions (group -> ports on this sled)
     DumpMcastSubs,
 
+    /// Dump M2P (multicast group -> underlay multicast) mappings
+    DumpM2p,
+
     /// Subscribe a port to a multicast group
     ///
     /// Allows a port to receive multicast traffic for the specified group.
@@ -962,7 +968,11 @@ fn main() -> anyhow::Result<()> {
             let next_hop_addr = NextHopV6::new(next_hop, next_hop_vni);
             let req = SetMcastForwardingReq {
                 underlay,
-                next_hops: vec![(next_hop_addr, replication)],
+                next_hops: vec![McastForwardingNextHop {
+                    next_hop: next_hop_addr,
+                    replication,
+                    source_filter: SourceFilter::default(),
+                }],
             };
             hdl.set_mcast_fwd(&req)?;
         }
@@ -980,8 +990,16 @@ fn main() -> anyhow::Result<()> {
             print_mcast_subs(&hdl.dump_mcast_subs()?)?;
         }
 
+        Command::DumpM2p => {
+            print_m2p(&hdl.dump_m2p()?)?;
+        }
+
         Command::McastSubscribe { port, group } => {
-            let req = McastSubscribeReq { port_name: port, group };
+            let req = McastSubscribeReq {
+                port_name: port,
+                group,
+                filter: SourceFilter::default(),
+            };
             hdl.mcast_subscribe(&req)?;
         }
 

crates/opte-api/src/cmd.rs

@@ -114,6 +114,8 @@ pub enum OpteCmd {
     DumpMcastSubscriptions = 107,
     /// Unsubscribe all ports from a multicast group.
     McastUnsubscribeAll = 108,
+    /// Read out all M2P (multicast group -> underlay multicast) mappings.
+    DumpMcast2Phys = 109,
 }
 
 impl TryFrom<c_int> for OpteCmd {
@@ -155,6 +157,7 @@ impl TryFrom<c_int> for OpteCmd {
             106 => Ok(Self::ClearMcast2Phys),
             107 => Ok(Self::DumpMcastSubscriptions),
             108 => Ok(Self::McastUnsubscribeAll),
+            109 => Ok(Self::DumpMcast2Phys),
             _ => Err(()),
         }
     }