feat(dataplane): pipeline rework for pkt injection

Signed-off-by: Fredi Raspall <fredi@githedgehog.com>

Author: Fredi-raspall

dataplane/src/packet_processor/egress.rs

@@ -4,15 +4,16 @@
 //! Implements an egress stage
 
 use std::net::IpAddr;
+#[allow(unused)]
 use tracing::{debug, error, trace, warn};
 
-use net::eth::Eth;
 use net::eth::ethtype::EthType;
 use net::eth::mac::{DestinationMac, SourceMac};
 use net::{
     buffer::PacketBufferMut,
     headers::{TryIpv4, TryIpv6},
 };
+use net::{eth::Eth, headers::TryIp};
 
 use net::headers::TryEthMut;
 use net::interface::InterfaceIndex;
@@ -131,10 +132,8 @@ impl Egress {
             /* do lookup on the adjacency table */
             let Some(adj) = atable.get_adjacency(addr, ifindex) else {
                 warn!("{nfi}: missing L2 info for {addr}");
-
-                /* Todo: Trigger ARP */
-
-                packet.done(DoneReason::MissL2resolution);
+                packet.get_meta_mut().set_need_arp_nd(true);
+                packet.done(DoneReason::MissL2resolution); // temporary until ARP/ND ready
                 return None;
             };
             /* get the mac from the adjacency */
@@ -152,25 +151,23 @@ impl Egress {
         }
     }
 
-    fn resolve_next_mac<Buf: PacketBufferMut>(
+    fn resolve_next_mac_ip<Buf: PacketBufferMut>(
         &self,
         ifindex: InterfaceIndex,
+        dst_ip: IpAddr,
         packet: &mut Packet<Buf>,
     ) -> Option<DestinationMac> {
-        let nfi = &self.name;
         // if packet was annotated with a next-hop address, try to resolve it using the
         // adjacency table. Otherwise, that means that the packet is directly connected
         // to us (on the same subnet). So, fetch the destination IP address and try to
         // resolve it with the adjacency table as well. If that fails, that's where the
         // ARP/ND would need to be triggered.
-        if let Some(nh_addr) = packet.get_meta().nh_addr {
-            self.get_adj_mac(packet, nh_addr, ifindex)
-        } else if let Some(destination) = packet.ip_destination() {
-            self.get_adj_mac(packet, destination, ifindex)
+        let target = if let Some(nh_addr) = packet.get_meta().nh_addr {
+            nh_addr
         } else {
-            warn!("{nfi}: could not determine packet destination IP address");
-            None
-        }
+            dst_ip
+        };
+        self.get_adj_mac(packet, target, ifindex)
     }
 
     #[inline]
@@ -180,11 +177,27 @@ impl Egress {
             packet.done(DoneReason::RouteFailure);
             return;
         };
+        let mut dst_mac = None;
+
+        /* if Ipv4/Ipv6, resolve destination mac. We can't rely on the current Eth mac since
+        we may have not overwritten it. FIXME(fredi): we should probably do that */
+        if packet.try_ip().is_some() {
+            let Some(dst_ip) = packet.ip_destination() else {
+                warn!("Failed to retrieve packet destination IP address");
+                packet.done(DoneReason::Malformed);
+                return;
+            };
+            /* resolve destination mac */
+            dst_mac = self.resolve_next_mac_ip(oif, dst_ip, packet);
+            if dst_mac.is_none() {
+                warn!("Failed to resolve mac for packet to {dst_ip}");
+                return;
+            }
+        }
 
-        /* resolve destination mac */
-        let Some(dst_mac) = self.resolve_next_mac(oif, packet) else {
-            // we could not figure out the destination MAC.
-            // resolve_next_mac() already calls packet.done()
+        /* if frame is non-ip and we don't know the dst mac, we drop the packet for the time being  */
+        let Some(dst_mac) = dst_mac else {
+            packet.done(DoneReason::MissL2resolution);
             return;
         };
 

dataplane/src/packet_processor/ingress.rs

@@ -48,9 +48,10 @@ impl Ingress {
         let ifname = &interface.name;
         match &interface.attachment {
             Some(Attachment::VRF(fibkey)) => {
+                // we should actually drop this
                 if packet.try_ip().is_none() {
-                    warn!("{nfi}: Processing of non-ip traffic on {ifname} is not supported");
-                    packet.done(DoneReason::NotIp);
+                    packet.get_meta_mut().set_local(true);
+                    packet.done(DoneReason::Delivered);
                     return;
                 }
                 let vrfid = fibkey.as_u32();
@@ -86,18 +87,21 @@ impl Ingress {
     }
 
     #[tracing::instrument(level = "trace")]
-    fn interface_ingress_eth_bcast<Buf: PacketBufferMut>(
+    fn interface_ingress_eth_bcast_mcast<Buf: PacketBufferMut>(
         &self,
-        interface: &Interface,
+        _interface: &Interface,
         packet: &mut Packet<Buf>,
     ) {
-        let nfi = self.name();
         packet.get_meta_mut().set_l2bcast(true);
-        packet.done(DoneReason::Unhandled);
-        warn!(
-            "{nfi}: Processing of broadcast ethernet frames is not supported (interface {ifname})",
-            ifname = interface.name
-        );
+        packet.get_meta_mut().set_local(true);
+        if packet.try_ip().is_none()
+            || packet
+                .ip_destination()
+                .unwrap_or_else(|| unreachable!())
+                .is_multicast()
+        {
+            packet.done(DoneReason::Delivered);
+        }
     }
 
     #[tracing::instrument(level = "trace")]
@@ -116,8 +120,8 @@ impl Ingress {
                 None => packet.done(DoneReason::NotEthernet),
                 Some(eth) => {
                     let dmac = eth.destination().inner();
-                    if dmac.is_broadcast() {
-                        self.interface_ingress_eth_bcast(interface, packet);
+                    if dmac.is_broadcast() || dmac.is_multicast() {
+                        self.interface_ingress_eth_bcast_mcast(interface, packet);
                     } else if dmac == if_mac {
                         self.interface_ingress_eth_ucast_local(interface, packet);
                     } else {
@@ -126,7 +130,7 @@ impl Ingress {
                 }
             }
         } else {
-            unreachable!();
+            unreachable!("We should not get packet without mac");
         }
     }
 
@@ -164,9 +168,7 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Ingress {
             if !packet.is_done() {
                 if let Some(iftable) = self.iftr.enter() {
                     match packet.get_meta().iif {
-                        None => {
-                            warn!("no incoming interface for packet");
-                        }
+                        None => warn!("Packet has no iff"),
                         Some(iif) => match iftable.get_interface(iif) {
                             None => {
                                 warn!("{nfi}: unknown incoming interface {iif}");
@@ -177,6 +179,8 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for Ingress {
                             }
                         },
                     }
+                } else {
+                    warn!("Unable to read interface table!");
                 }
             }
             packet.enforce()

dataplane/src/packet_processor/ipforward.rs

@@ -144,13 +144,9 @@ impl IpForwarder {
                 packet.done(DoneReason::Malformed);
             }
             None => {
-                /* send to kernel, among other options */
                 debug!("Packet should be delivered to kernel...");
-                /*
-                We can't re-inject packet on ingress, so let's disable this to avoid churn
-                packet.get_meta_mut().oif = Some(packet.get_meta().iif);
-                 */
-                packet.done(DoneReason::Unhandled);
+                packet.get_meta_mut().set_local(true);
+                packet.done(DoneReason::Delivered);
             }
         }
     }
@@ -383,7 +379,7 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for IpForwarder {
                 let vrfid = packet.get_meta_mut().vrf.take();
                 if let Some(vrfid) = vrfid {
                     self.forward_packet(&mut packet, vrfid);
-                } else {
+                } else if !packet.get_meta().local() {
                     warn!("{}: missing information to handle packet", self.name);
                 }
             }

dataplane/src/packet_processor/mod.rs

@@ -73,12 +73,11 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
     let natallocator_factory = natallocatorw.get_reader_factory();
 
     // build pkt io stages
-    let pktio_ip: PktIo<Buf> = PktIo::new(0, 10_000usize).set_name("pkt-io-ip ");
-    let pktio_egress: PktIo<Buf> = PktIo::new(10_000usize, 0).set_name("pkt-io-egress");
+    let pktio: PktIo<Buf> = PktIo::new(10_000usize, 10_000usize).set_name("pkt-io-ip ");
 
     // queues to expose
-    let puntq = pktio_ip.get_puntq().unwrap_or_else(|| unreachable!());
-    let injectq = pktio_egress.get_injectq().unwrap_or_else(|| unreachable!());
+    let puntq = pktio.get_puntq().unwrap_or_else(|| unreachable!());
+    let injectq = pktio.get_injectq().unwrap_or_else(|| unreachable!());
 
     let pipeline_builder = move || {
         // Build network functions
@@ -94,24 +93,22 @@ pub(crate) fn start_router<Buf: PacketBufferMut>(
         let stats_stage = Stats::new("stats", writer.clone());
         let flow_lookup_nf = LookupNF::new(flow_table.clone());
         let flow_expirations_nf = ExpirationsNF::new(flow_table.clone());
-        let pktio_ip_worker = pktio_ip.clone();
-        let pktio_egress_worker = pktio_egress.clone();
+        let pktio_worker = pktio.clone();
 
         // Build the pipeline for a router. The composition of the pipeline (in stages) is currently
         // hard-coded. In any pipeline, the Stats and ExpirationsNF stages should go last
         DynPipeline::new()
             .add_stage(dumper1)
             .add_stage(stage_ingress)
             .add_stage(iprouter1)
-            .add_stage(pktio_ip_worker)
             .add_stage(dst_vpcd_lookup)
             .add_stage(flow_lookup_nf)
             .add_stage(stateless_nat)
             .add_stage(stateful_nat)
             .add_stage(iprouter2)
-            .add_stage(pktio_egress_worker)
             .add_stage(stage_egress)
             .add_stage(dumper2)
+            .add_stage(pktio_worker)
             .add_stage(flow_expirations_nf)
             .add_stage(stats_stage)
     };

pkt-io/src/nf.rs

@@ -140,7 +140,7 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for PktIo<Buf> {
             match &self.puntq {
                 None => Some(packet),
                 Some(puntq) => {
-                    if packet.get_meta().local() && !packet.is_done() {
+                    if packet.get_meta().local() {
                         match puntq.push(Box::new(packet)) {
                             Ok(()) => None, // punted!
                             Err(mut packet) => {

pkt-meta/src/dst_vpcd_lookup/mod.rs

@@ -193,17 +193,13 @@ impl<Buf: PacketBufferMut> NetworkFunction<Buf> for DstVpcdLookup {
         input: Input,
     ) -> impl Iterator<Item = Packet<Buf>> + 'a {
         input.filter_map(|mut packet| {
-            if let Some(tablesr) = &self.tablesr.enter() {
-                if !packet.is_done() {
-                    // FIXME: ideally, we'd `enter` once for the whole batch. However,
-                    // this requires boxing the closures, which may be worse than
-                    // calling `enter` per packet? ... if not uglier
-
+            if !packet.get_meta().local() && !packet.is_done() && packet.try_ip().is_some() {
+                if let Some(tablesr) = &self.tablesr.enter() {
                     self.process_packet(tablesr, &mut packet);
+                } else {
+                    error!("{}: failed to read vpcd tables", self.name);
+                    packet.done(DoneReason::InternalFailure);
                 }
-            } else {
-                error!("{}: failed to read vpcd tables", self.name);
-                packet.done(DoneReason::InternalFailure);
             }
             packet.enforce()
         })