WIP async read write implementation

burrow/Cargo.toml

@@ -39,6 +39,7 @@ ip_network = "0.4.0"
 async-trait = "0.1.74"
 async-channel = "1.9"
 schemars = "0.8"
+futures = "0.3.28"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 caps = "0.5.5"

burrow/src/main.rs

@@ -13,6 +13,7 @@ use tracing_subscriber::{prelude::*, FmtSubscriber, EnvFilter};
 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
 use tun::TunInterface;
 
+
 mod daemon;
 mod wireguard;
 

burrow/src/wireguard/iface.rs

@@ -1,4 +1,5 @@
 use std::{net::IpAddr, rc::Rc};
+use std::sync::Arc;
 
 use anyhow::Error;
 use async_trait::async_trait;
@@ -11,6 +12,7 @@ use tokio::{
     task::{self, JoinHandle},
 };
 use tun::tokio::TunInterface;
+use futures::future::join_all;
 
 use super::{noise::Tunnel, pcb, Peer, PeerPcb};
 
@@ -32,7 +34,7 @@ impl PacketInterface for tun::tokio::TunInterface {
 }
 
 struct IndexedPcbs {
-    pcbs: Vec<PeerPcb>,
+    pcbs: Vec<Arc<Mutex<PeerPcb>>>,
     allowed_ips: IpNetworkTable<usize>,
 }
 
@@ -49,7 +51,7 @@ impl IndexedPcbs {
         for allowed_ip in pcb.allowed_ips.iter() {
             self.allowed_ips.insert(allowed_ip.clone(), idx);
         }
-        self.pcbs.insert(idx, pcb);
+        self.pcbs.insert(idx, Arc::new(Mutex::new(pcb)));
     }
 
     pub fn find(&mut self, addr: IpAddr) -> Option<usize> {
@@ -57,8 +59,8 @@ impl IndexedPcbs {
         Some(idx)
     }
 
-    pub fn connect(&mut self, idx: usize, handle: JoinHandle<()>) {
-        self.pcbs[idx].handle = Some(handle);
+    pub async fn connect(&mut self, idx: usize, handle: JoinHandle<()>) {
+        self.pcbs[idx].lock().await.handle = Some(handle);
     }
 }
 
@@ -72,8 +74,8 @@ impl FromIterator<PeerPcb> for IndexedPcbs {
 }
 
 pub struct Interface {
-    tun: Rc<Mutex<TunInterface>>,
-    pcbs: Rc<Mutex<IndexedPcbs>>,
+    tun: Arc<Mutex<TunInterface>>,
+    pcbs: Arc<Mutex<IndexedPcbs>>,
 }
 
 impl Interface {
@@ -84,21 +86,23 @@ impl Interface {
             .map(|peer| PeerPcb::new(peer))
             .collect::<Result<_, _>>()?;
 
-        let tun = Rc::new(Mutex::new(tun));
-        let pcbs = Rc::new(Mutex::new(pcbs));
+        let tun = Arc::new(Mutex::new(tun));
+        let pcbs = Arc::new(Mutex::new(pcbs));
         Self { tun, pcbs }
     }
 
     pub async fn run(self) {
-        let pcbs = self.pcbs;
-        let tun = self.tun;
+        let pcbs = self.pcbs.clone();
+        let tun = self.tun.clone();
         log::info!("starting interface");
 
         let outgoing = async move {
             loop {
+                log::debug!("starting loop...");
                 let mut buf = [0u8; 3000];
 
                 let mut tun = tun.lock().await;
+                log::debug!("awaiting read...");
                 let src = match tun.recv(&mut buf[..]).await {
                     Ok(len) => &buf[..len],
                     Err(e) => {
@@ -127,7 +131,7 @@ impl Interface {
 
                 log::debug!("found peer:{}", idx);
 
-                match pcbs.pcbs[idx].send(src).await {
+                match pcbs.pcbs[idx].lock().await.send(src).await {
                     Ok(..) => {
                         log::debug!("sent packet to peer {}", dst_addr);
                     }
@@ -135,26 +139,32 @@ impl Interface {
                         log::error!("failed to send packet {}", e);
                         continue
                     },
-                }
-
-                let mut recv_buf = [0;1500];
-                match pcbs.pcbs[idx].recv(&mut recv_buf[..]).await {
-                    Ok(siz) => {
-                        log::info!("received {} bytes from peer",siz);
-                        log::debug!("bytes: {:?}", &recv_buf[..siz]);
-                    },
-                    Err(e) => {
-                        log::error!("failed to receive packet {}", e);
-                        continue
-                    }
-                }
+                };
             }
         };
 
+
+
         task::LocalSet::new()
             .run_until(async move {
-                let outgoing = task::spawn_local(outgoing);
-                join!(outgoing);
+                let mut tsks = vec![];
+                let tun = self.tun.clone();
+                let outgoing = tokio::task::spawn(outgoing);
+                tsks.push(outgoing);
+                {
+                    let pcbs = self.pcbs.lock().await;
+                    for i in 0..pcbs.pcbs.len(){
+                        let pcb = pcbs.pcbs[i].clone();
+                        let tun = tun.clone();
+                        let tsk = async move{
+                            pcb.lock().await.run(tun).await.unwrap();
+                        };
+                        tsks.push(tokio::task::spawn(tsk));
+                    }
+                    log::debug!("spawned read tasks");
+                }
+                log::debug!("preparing to join..");
+                join_all(tsks).await;
             })
             .await;
     }

burrow/src/wireguard/pcb.rs

@@ -1,9 +1,12 @@
 use std::net::SocketAddr;
+use std::rc::Rc;
+use std::sync::Arc;
 
 use anyhow::Error;
 use fehler::throws;
 use ip_network::IpNetwork;
 use tokio::{net::UdpSocket, task::JoinHandle};
+use tokio::sync::Mutex;
 
 use super::{
     iface::PacketInterface,
@@ -44,8 +47,9 @@ impl PeerPcb {
         Ok(())
     }
 
-    pub async fn run(&self, interface: Box<&dyn PacketInterface>) -> Result<(), Error> {
+    pub async fn run(&self, interface: Arc<Mutex<impl PacketInterface>>) -> Result<(), Error> {
         let mut buf = [0u8; 3000];
+        log::debug!("starting read loop for pcb...");
         loop {
             let Some(socket) = self.socket.as_ref() else {
                 continue