add router stats and spawn with delay

Author: rkuhn

Cargo.lock

@@ -3,15 +3,21 @@ members = ["cli", "core", "machine", "macros", "nat", "router", "."]
 
 [package]
 name = "netsim-embed"
-version = "0.8.0"
+version = "0.9.0"
 authors = ["David Craven <david@craven.ch>"]
 edition = "2018"
 description = "Network simulator."
 license = "MIT"
 repository = "https://github.com/ipfs-rust/netsim-embed"
 
 [features]
-ipc = ["dep:libtest-mimic", "dep:anyhow", "dep:ipc-channel", "dep:netsim-embed-macros", "dep:serde"]
+ipc = [
+    "dep:libtest-mimic",
+    "dep:anyhow",
+    "dep:ipc-channel",
+    "dep:netsim-embed-macros",
+    "dep:serde",
+]
 
 [dependencies]
 anyhow = { version = "1.0.40", optional = true }

Cargo.toml

@@ -1,10 +1,18 @@
-use futures::channel::{mpsc, oneshot};
-use futures::future::{poll_fn, FutureExt};
-use futures::stream::{FuturesUnordered, StreamExt};
+use futures::{
+    channel::{mpsc, oneshot},
+    future::{poll_fn, FutureExt},
+    stream::{FuturesUnordered, StreamExt},
+};
 use libpacket::ipv4::Ipv4Packet;
 use netsim_embed_core::{Ipv4Route, Plug};
-use std::net::Ipv4Addr;
-use std::task::Poll;
+use std::{
+    net::Ipv4Addr,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+    task::Poll,
+};
 
 #[derive(Debug)]
 #[allow(clippy::enum_variant_names)]
@@ -20,13 +28,43 @@ pub struct Ipv4Router {
     #[allow(unused)]
     addr: Ipv4Addr,
     ctrl: mpsc::UnboundedSender<RouterCtrl>,
+    counters: Arc<Counters>,
+}
+
+#[derive(Debug, Default)]
+struct Counters {
+    forwarded: AtomicUsize,
+    invalid: AtomicUsize,
+    disabled: AtomicUsize,
+    unroutable: AtomicUsize,
 }
 
 impl Ipv4Router {
     pub fn new(addr: Ipv4Addr) -> Self {
         let (tx, rx) = mpsc::unbounded();
-        router(addr, rx);
-        Self { addr, ctrl: tx }
+        let counters = Arc::new(Counters::default());
+        router(addr, Arc::clone(&counters), rx);
+        Self {
+            addr,
+            ctrl: tx,
+            counters,
+        }
+    }
+
+    pub fn forwarded(&self) -> usize {
+        self.counters.forwarded.load(Ordering::Relaxed)
+    }
+
+    pub fn invalid(&self) -> usize {
+        self.counters.invalid.load(Ordering::Relaxed)
+    }
+
+    pub fn disabled(&self) -> usize {
+        self.counters.disabled.load(Ordering::Relaxed)
+    }
+
+    pub fn unroutable(&self) -> usize {
+        self.counters.unroutable.load(Ordering::Relaxed)
     }
 
     pub fn add_connection(&self, id: usize, plug: Plug, routes: Vec<Ipv4Route>) {
@@ -56,7 +94,7 @@ impl Ipv4Router {
     }
 }
 
-fn router(addr: Ipv4Addr, mut ctrl: mpsc::UnboundedReceiver<RouterCtrl>) {
+fn router(addr: Ipv4Addr, counters: Arc<Counters>, mut ctrl: mpsc::UnboundedReceiver<RouterCtrl>) {
     async_global_executor::spawn(async move {
         let mut conns = vec![];
         loop {
@@ -87,7 +125,7 @@ fn router(addr: Ipv4Addr, mut ctrl: mpsc::UnboundedReceiver<RouterCtrl>) {
                     None => break,
                 },
                 incoming = incoming(&mut conns).fuse() => match incoming {
-                    (_, Some(packet)) => forward_packet(addr, &mut conns, packet),
+                    (_, Some(packet)) => forward_packet(addr, &counters, &mut conns, packet),
                     (i, None) => { conns.swap_remove(i); }
                 }
             }
@@ -111,12 +149,14 @@ async fn incoming(conns: &mut [(usize, Plug, Vec<Ipv4Route>, bool)]) -> (usize,
 
 fn forward_packet(
     addr: Ipv4Addr,
+    counters: &Counters,
     conns: &mut [(usize, Plug, Vec<Ipv4Route>, bool)],
     bytes: Vec<u8>,
 ) {
     let packet = if let Some(packet) = Ipv4Packet::new(&bytes) {
         packet
     } else {
+        counters.invalid.fetch_add(1, Ordering::Relaxed);
         log::info!("router {}: dropping invalid ipv4 packet", addr);
         return;
     };
@@ -130,8 +170,10 @@ fn forward_packet(
         for route in routes {
             if route.dest().contains(dest) || dest.is_broadcast() || dest.is_multicast() {
                 if !*en {
+                    counters.disabled.fetch_add(1, Ordering::Relaxed);
                     log::trace!("router {}: route {:?} disabled", addr, route);
                 } else {
+                    counters.forwarded.fetch_add(1, Ordering::Relaxed);
                     log::trace!("router {}: routing packet on route {:?}", addr, route);
                     tx.unbounded_send(bytes.clone());
                     forwarded = true;
@@ -141,6 +183,7 @@ fn forward_packet(
     }
     if !forwarded {
         let src = packet.get_source();
+        counters.unroutable.fetch_add(1, Ordering::Relaxed);
         log::debug!(
             "router {}: dropping unroutable packet from {} to {}",
             addr,

router/src/lib.rs

@@ -1,6 +1,5 @@
 use async_process::Command;
 use futures::prelude::*;
-pub use libpacket::*;
 use netsim_embed_core::*;
 pub use netsim_embed_core::{DelayBuffer, Ipv4Range, Protocol};
 pub use netsim_embed_machine::{unshare_user, Machine, MachineId, Namespace};
@@ -72,7 +71,12 @@ where
     }
 
     #[cfg(feature = "ipc")]
-    pub async fn spawn<M: MachineFn>(&mut self, _machine: M, arg: M::Arg) -> MachineId {
+    pub async fn spawn<M: MachineFn>(
+        &mut self,
+        _machine: M,
+        arg: M::Arg,
+        delay: Option<DelayBuffer>,
+    ) -> MachineId {
         use ipc_channel::ipc;
         let id = M::id();
         let (server, server_name) = ipc::IpcOneShotServer::<ipc::IpcSender<M::Arg>>::new().unwrap();
@@ -82,7 +86,7 @@ where
             &format!("{id}"),
             &server_name,
         ]);
-        let machine = self.spawn_machine(command, None).await;
+        let machine = self.spawn_machine(command, delay).await;
         let (_, ipc) = async_global_executor::spawn_blocking(|| server.accept())
             .await
             .unwrap();
@@ -235,6 +239,22 @@ impl Network {
         self.range
     }
 
+    pub fn num_forwarded(&self) -> usize {
+        self.router.forwarded()
+    }
+
+    pub fn num_invalid(&self) -> usize {
+        self.router.invalid()
+    }
+
+    pub fn num_disabled(&self) -> usize {
+        self.router.disabled()
+    }
+
+    pub fn num_unroutable(&self) -> usize {
+        self.router.unroutable()
+    }
+
     pub fn unique_addr(&mut self) -> Ipv4Addr {
         let addr = self.range.address_for(self.device);
         self.device += 1;

src/lib.rs

@@ -15,7 +15,7 @@ fn add((left, right, sender): (IpcReceiver<usize>, IpcReceiver<usize>, IpcSender
 fn can_send_one() {
     let mut s = netsim_embed::Netsim::<String, String>::new();
     let (sender, receiver) = ipc_channel::ipc::channel().unwrap();
-    let _ = async_std::task::block_on(s.spawn(send_one, sender));
+    let _ = async_std::task::block_on(s.spawn(send_one, sender, None));
     assert_eq!(1, receiver.recv().unwrap());
 }
 
@@ -24,13 +24,13 @@ fn one_plus_one_makes_two() {
         let mut s = netsim_embed::Netsim::<String, String>::new();
 
         let (sender1, receiver1) = ipc_channel::ipc::channel::<usize>().unwrap();
-        let _ = s.spawn(send_one, sender1).await;
+        let _ = s.spawn(send_one, sender1, None).await;
 
         let (sender2, receiver2) = ipc_channel::ipc::channel::<usize>().unwrap();
-        let _ = s.spawn(send_one, sender2).await;
+        let _ = s.spawn(send_one, sender2, None).await;
 
         let (sender3, receiver3) = ipc_channel::ipc::channel::<usize>().unwrap();
-        let _ = s.spawn(add, (receiver1, receiver2, sender3)).await;
+        let _ = s.spawn(add, (receiver1, receiver2, sender3), None).await;
 
         assert_eq!(2, receiver3.recv().unwrap());
     })