feat(pkt-io): sketch io - tap

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

Author: Fredi-raspall

Cargo.lock

@@ -10,10 +10,13 @@ license = "Apache-2.0"
 net = { workspace = true }
 pipeline = { workspace = true }
 tracectl = { workspace = true }
+interface-manager = { workspace = true }
 
 # external
 crossbeam = { workspace = true }
+futures = { workspace = true }
 linkme = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
 tracing = { workspace = true }
 
 [dev-dependencies]

pkt-io/Cargo.toml

@@ -0,0 +1,133 @@
+// SPDX-License-Identifier: Apache-2.0
+// Copyright Open Network Fabric Authors
+
+//! Local IO
+
+#![allow(unused)]
+
+use futures::future::join_all;
+use tokio::{io, spawn};
+
+use crate::PktQueue;
+use interface_manager::interface::TapDevice;
+
+use net::buffer::PacketBufferMut;
+use net::interface::InterfaceIndex;
+use net::packet::Packet;
+
+use std::cell::RefCell;
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use tokio::sync::Mutex;
+
+#[allow(unused)]
+use tracing::{debug, error, warn};
+
+/// A mapping between an interface and a tap, and the tap itself
+struct TapMapping {
+    tapid: InterfaceIndex,
+    ifid: InterfaceIndex,
+    tap: Mutex<TapDevice>,
+}
+impl TapMapping {
+    fn new(tapid: InterfaceIndex, ifid: InterfaceIndex, tap: TapDevice) -> Self {
+        Self {
+            tapid,
+            ifid,
+            tap: Mutex::new(tap),
+        }
+    }
+}
+
+/// Collection of mappings doubly indexed. We need to cope with the case that
+/// tap ids clash with interfaceids, which could happen with distinct network namespaces.
+/// We need distinct indices because the action after a lookup depends on the 'type' of key.
+struct TapMappingDb {
+    by_tapid: HashMap<InterfaceIndex, Arc<TapMapping>>,
+    by_ifid: HashMap<InterfaceIndex, Arc<TapMapping>>,
+}
+impl TapMappingDb {
+    fn add_mapping(&mut self, tapid: InterfaceIndex, ifid: InterfaceIndex, tap: TapDevice) {
+        let mapping = Arc::new(TapMapping::new(tapid, ifid, tap));
+        self.by_tapid.insert(tapid, Arc::clone(&mapping));
+        self.by_ifid.insert(ifid, mapping);
+    }
+    fn get_by_ifid(&self, id: InterfaceIndex) -> Option<&TapMapping> {
+        self.by_ifid.get(&id).map(|m| &**m)
+    }
+    fn get_by_tapid(&self, id: InterfaceIndex) -> Option<&TapMapping> {
+        self.by_tapid.get(&id).map(|m| &**m)
+    }
+}
+
+struct IoManager<Buf: PacketBufferMut> {
+    puntq: PktQueue<Buf>, // for the time being only 2 queues. May need more with multiple PktIo stages
+    injectq: PktQueue<Buf>,
+    pub(crate) mapdb: TapMappingDb,
+}
+impl<Buf: PacketBufferMut> IoManager<Buf> {
+    // handle incoming packets: fetch packets punted by pipeline and write them to the corresponding tap
+    async fn handle_incoming_pkts(&self) {
+        loop {
+            if let Some(packet) = self.puntq.pop() {
+                let Some(ifindex) = packet.get_meta().iif else {
+                    warn!("Packet has no incoming interface annotation. Dropping it...");
+                    return;
+                };
+                debug!("Got a packet from interface {ifindex}");
+
+                // lookup tap from the id of the interface it was received
+                let Some(tap) = self.mapdb.get_by_ifid(ifindex).map(|m| &m.tap) else {
+                    warn!("Could not find tap for interface {ifindex}. Dropping packet...");
+                    return;
+                };
+
+                debug!("Will write packet to tap");
+                match packet.serialize() {
+                    Err(e) => error!("Failed to serialize packet prior to tap write: {e}"),
+                    Ok(buf) => match tap.lock().await.write(buf).await {
+                        Err(e) => error!("Failed to write buffer to tap: {e}"),
+                        Ok(()) => debug!("Wrote packet to tap!"),
+                    },
+                }
+            }
+        }
+    }
+
+    // handle outgoing traffic. We read from tap and push to injection queue
+    async fn send_outgoing_packet(&self, m: &TapMapping) {
+        let mut buffer = Buf::create();
+        match m.tap.lock().await.read(&mut buffer).await {
+            Ok(size) => match Packet::new(buffer) {
+                Err(e) => error!("Failed to build packet from buffer: {e}"),
+                Ok(mut packet) => {
+                    debug!(
+                        "Got buffer of size {size} from tap {}. Target oif: {}",
+                        m.tapid, m.ifid
+                    );
+                    packet.get_meta_mut().oif = Some(m.ifid);
+
+                    // inject
+                    match self.injectq.push(Box::new(packet)) {
+                        Ok(()) => debug!("Injected packet from tap"),
+                        Err(dropped) => warn!("Could not inject packet {dropped}"),
+                    }
+                }
+            },
+            Err(e) => error!("Failure reading packet from tap: {e}"),
+        }
+    }
+
+    async fn handle_outgoing_pkts(&self, injectq: PktQueue<Buf>) {
+        // unsure if this is the best way to do this.
+        loop {
+            let x = self
+                .mapdb
+                .by_tapid
+                .values()
+                .map(|m| self.send_outgoing_packet(m));
+            join_all(x).await;
+        }
+    }
+}

pkt-io/src/io.rs

@@ -3,6 +3,7 @@
 
 //! Local packet I/O for gateway
 
+mod io;
 mod nf;
 mod tests;