nuke old networking subsystem

src/bin/dfi.rs

@@ -9,8 +9,7 @@ use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use sapvi::net;
-use sapvi::{Channel, Result, SeedProtocol, ServerProtocol};
+use sapvi::{net, Result};
 
 use std::net::TcpListener;
 
@@ -160,6 +159,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     Ok(())
 }
 
+/*
 async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
     let connections = Arc::new(Mutex::new(HashMap::new()));
 
@@ -226,7 +226,6 @@ async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<
         client_slots.push(client);
     }
 
-    /*
     let rpc = RpcInterface::new();
     let http = listen(
         executor.clone(),
@@ -240,7 +239,7 @@ async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<
     rpc.quit_recv.recv().await?;
 
     http_task.cancel().await;
-    */
+
     match server_task {
         None => {}
         Some(server_task) => {
@@ -249,6 +248,7 @@ async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<
     }
     Ok(())
 }
+*/
 
 struct ProgramOptions {
     network_settings: net::Settings,

src/lib.rs

@@ -15,11 +15,7 @@ pub mod vm_serial;
 
 pub use crate::bls_extensions::BlsStringConversion;
 pub use crate::error::{Error, Result};
-pub use crate::net::messages::{select_event, send_message, sleep};
 pub use crate::net::p2p::P2p;
-pub use crate::net::protocol::client_protocol::Channel;
-pub use crate::net::protocol::seed_protocol::SeedProtocol;
-pub use crate::net::protocol::server_protocol::ServerProtocol;
 pub use crate::serial::{Decodable, Encodable};
 pub use crate::vm::{
     AllocType, ConstraintInstruction, CryptoOperation, VariableIndex, VariableRef, ZKVMCircuit,

src/net/messages.rs

@@ -366,28 +366,6 @@ pub async fn send_message<W: AsyncWrite + Unpin>(stream: &mut W, message: Messag
     send_packet(stream, packet).await
 }
 
-// Eventloop event
-pub enum Event {
-    // Message to be sent from event queue
-    Send(Message),
-    // Received message to process by protocol
-    Receive(Message),
-    // Connection ping-pong timeout
-    Timeout,
-}
-
-pub async fn select_event(
-    stream: &mut AsyncTcpStream,
-    send_rx: &async_channel::Receiver<Message>,
-    inactivity_timer: &InactivityTimer,
-) -> Result<Event> {
-    Ok(futures::select! {
-        message = send_rx.recv().fuse() => Event::Send(message?),
-        message = receive_message(stream).fuse() => Event::Receive(message?),
-        _ = inactivity_timer.wait_for_wakeup().fuse() => Event::Timeout
-    })
-}
-
 pub async fn sleep(seconds: u64) {
     Timer::after(Duration::from_secs(seconds)).await;
 }

src/net/mod.rs

@@ -8,7 +8,6 @@ pub mod message_subscriber;
 pub mod messages;
 pub mod hosts;
 pub mod p2p;
-pub mod protocol;
 pub mod protocols;
 pub mod proxy;
 pub mod sessions;

src/net/p2p.rs

@@ -41,7 +41,11 @@ impl P2p {
 
     /// Synchronize the blockchain and then begin long running sessions,
     /// call after start() is invoked.
-    pub fn run() {}
+    pub fn run() {
+        //let inbound = InboundSession::new(Arc::downgrade(&self));
+        //let inbound_task = inbound.start(executor.clone());
+        //inbound_task.await;
+    }
 
     pub async fn store(self: Arc<Self>, channel: ChannelPtr) {
         self.pending_channels

src/net/protocol/client_protocol.rs

@@ -1,227 +0,0 @@
-use async_std::sync::Mutex;
-use log::*;
-use rand::seq::SliceRandom;
-use smol::{Async, Executor};
-use std::net::{SocketAddr, TcpStream};
-use std::sync::atomic::AtomicU64;
-use std::sync::Arc;
-
-use crate::error::Result;
-use crate::net::messages as net;
-use crate::net::protocol::protocol_base;
-use crate::utility::{AddrsStorage, ConnectionsMap};
-
-pub struct Channel {
-    send_sx: async_channel::Sender<net::Message>,
-    send_rx: async_channel::Receiver<net::Message>,
-    connections: ConnectionsMap,
-    main_process: Mutex<Option<smol::Task<()>>>,
-
-    accept_addr: Option<SocketAddr>,
-    stored_addrs: AddrsStorage,
-}
-
-impl Channel {
-    pub fn new(
-        connections: ConnectionsMap,
-        accept_addr: Option<SocketAddr>,
-        stored_addrs: AddrsStorage,
-    ) -> Arc<Self> {
-        let (send_sx, send_rx) = async_channel::unbounded::<net::Message>();
-        Arc::new(Self {
-            send_sx,
-            send_rx,
-            connections,
-            main_process: Mutex::new(None),
-            accept_addr,
-            stored_addrs,
-        })
-    }
-
-    pub fn get_send_pipe(&self) -> async_channel::Sender<net::Message> {
-        self.send_sx.clone()
-    }
-
-    async fn fetch_random_addr(
-        self: Arc<Self>,
-        accept_addr: &Option<SocketAddr>,
-        stored_addrs: &AddrsStorage,
-        connections: &ConnectionsMap,
-    ) {
-        loop {
-            let addr = match stored_addrs.lock().await.choose(&mut rand_core::OsRng) {
-                Some(addr) => addr.clone(),
-                None => {
-                    debug!("No addresses in store. Sleeping for 2 secs before retrying...");
-                    net::sleep(2).await;
-                    continue;
-                }
-            };
-            if connections.lock().await.contains_key(&addr) {
-                continue;
-            }
-            if let Some(accept_addr) = accept_addr {
-                if addr == *accept_addr {
-                    continue;
-                }
-            }
-        }
-    }
-
-    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
-        let executor2 = executor.clone();
-        let self2 = self.clone();
-
-        *self2.main_process.lock().await = Some(executor.spawn(async move {
-            loop {
-                let addr = match self.stored_addrs.lock().await.choose(&mut rand_core::OsRng) {
-                    Some(addr) => addr.clone(),
-                    None => {
-                        debug!("No addresses in store. Sleeping for 2 secs before retrying...");
-                        net::sleep(2).await;
-                        continue;
-                    }
-                };
-                if self.connections.lock().await.contains_key(&addr) {
-                    continue;
-                }
-                if let Some(accept_addr) = self.accept_addr {
-                    if addr == accept_addr {
-                        continue;
-                    }
-                }
-
-                debug!("Attempting connect to {}", addr);
-
-                self.try_connect_process(addr, executor2.clone()).await;
-
-                // TODO: Fix this
-                net::sleep(2).await;
-            }
-        }));
-    }
-
-    pub async fn start_manual(
-        self: Arc<Self>,
-        remote_addr: SocketAddr,
-        executor: Arc<Executor<'_>>,
-    ) {
-        let executor2 = executor.clone();
-        let self2 = self.clone();
-
-        *self2.main_process.lock().await = Some(executor.spawn(async move {
-            loop {
-                for _ in 0..4 {
-                    debug!("Attempting connect to {}", remote_addr);
-
-                    self.try_connect_process(remote_addr, executor2.clone())
-                        .await;
-                }
-                net::sleep(2).await;
-            }
-        }));
-    }
-
-    pub async fn try_connect_process(&self, address: SocketAddr, executor: Arc<Executor<'_>>) {
-        match Async::<TcpStream>::connect(address.clone()).await {
-            Ok(stream) => {
-                let _ = self.handle_connect(stream, address, executor).await;
-            }
-            Err(_err) => {
-                warn!("Unable to connect to addr {:?}: {}", address, _err);
-            }
-        }
-    }
-
-    async fn handle_connect(
-        &self,
-        stream: Async<TcpStream>,
-        address: SocketAddr,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<()> {
-        debug!("Connected to {}", address);
-
-        let stream = async_dup::Arc::new(stream);
-        self.connections
-            .lock()
-            .await
-            .insert(address.clone(), self.send_sx.clone());
-
-        // Run event loop
-        match self.event_loop_process(stream, executor).await {
-            Ok(()) => {
-                warn!("Server timeout");
-            }
-            Err(err) => {
-                warn!("Server disconnected: {}", err);
-            }
-        }
-        self.connections.lock().await.remove(&address);
-        Ok(())
-    }
-
-    async fn send_addr(
-        send_sx: async_channel::Sender<net::Message>,
-        accept_addr: SocketAddr,
-    ) -> Result<()> {
-        loop {
-            send_sx
-                .send(net::Message::Addrs(net::AddrsMessage {
-                    addrs: vec![accept_addr],
-                }))
-                .await?;
-
-            net::sleep(3600).await;
-        }
-    }
-
-    pub async fn event_loop_process(
-        &self,
-        mut stream: net::AsyncTcpStream,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<()> {
-        let inactivity_timer = net::InactivityTimer::new(executor.clone());
-
-        let clock = Arc::new(AtomicU64::new(0));
-        let send_sx2 = self.send_sx.clone();
-        let clock2 = clock.clone();
-        let ping_task = executor.spawn(protocol_base::repeat_ping(send_sx2, clock2));
-
-        let mut send_addr_task = None;
-        if let Some(accept_addr) = self.accept_addr {
-            send_addr_task =
-                Some(executor.spawn(Self::send_addr(self.send_sx.clone(), accept_addr.clone())));
-        }
-
-        loop {
-            let event = net::select_event(&mut stream, &self.send_rx, &inactivity_timer).await?;
-
-            match event {
-                net::Event::Send(message) => {
-                    net::send_message(&mut stream, message).await?;
-                }
-                net::Event::Receive(message) => {
-                    inactivity_timer.reset().await?;
-                    protocol_base::protocol(
-                        message,
-                        &self.stored_addrs,
-                        &self.send_sx,
-                        Some(&clock),
-                        self.connections.clone(),
-                    )
-                    .await?;
-                }
-                net::Event::Timeout => break,
-            }
-        }
-
-        if let Some(send_addr_task) = send_addr_task {
-            send_addr_task.cancel().await;
-        }
-        ping_task.cancel().await;
-        inactivity_timer.stop().await;
-
-        // Connection timed out
-        Ok(())
-    }
-}

src/net/protocol/mod.rs

@@ -1,4 +0,0 @@
-pub mod client_protocol;
-pub mod protocol_base;
-pub mod seed_protocol;
-pub mod server_protocol;

src/net/protocol/protocol_base.rs

@@ -1,110 +0,0 @@
-use log::*;
-use std::sync::atomic::Ordering;
-
-use crate::net::messages as net;
-use crate::utility::{get_current_time, AddrsStorage, Clock, ConnectionsMap};
-use crate::Result;
-
-// Clients send repeated pings. Servers only respond with pong.
-pub async fn repeat_ping(send_sx: async_channel::Sender<net::Message>, clock: Clock) -> Result<()> {
-    debug!("ping process");
-    loop {
-        // Send ping
-        send_sx.send(net::Message::Ping).await?;
-        debug!("send Message::Ping");
-        clock.store(get_current_time(), Ordering::Relaxed);
-
-        net::sleep(5).await;
-    }
-}
-
-pub async fn protocol(
-    message: net::Message,
-    stored_addrs: &AddrsStorage,
-    send_sx: &async_channel::Sender<net::Message>,
-    clock: Option<&Clock>,
-    connections: ConnectionsMap,
-) -> Result<()> {
-    match message {
-        net::Message::Ping => {
-            send_sx.send(net::Message::Pong).await?;
-        }
-        net::Message::Pong => {
-            if let Some(clock) = clock {
-                let current_time = get_current_time();
-                let elapsed = current_time - clock.load(Ordering::Relaxed);
-                info!("Ping time: {} ms", elapsed);
-            }
-        }
-        net::Message::GetAddrs(_message) => {
-            info!("received GetAddrMessage");
-            send_sx
-                .send(net::Message::Addrs(net::AddrsMessage {
-                    addrs: stored_addrs.lock().await.to_vec(),
-                }))
-                .await?;
-        }
-        net::Message::Addrs(message) => {
-            info!("received AddrMessage");
-            let mut stored_addrs = stored_addrs.lock().await;
-            for addr in message.addrs {
-                if stored_addrs.contains(&addr) {
-                    continue;
-                }
-                info!("Added new address to storage {}", addr.to_string());
-                stored_addrs.push(addr);
-            }
-        }
-        net::Message::Sync => {
-            info!("received SyncMessage");
-            /*send_sx
-            .send(net::Message::Inv(net::InvMessage {
-                slabs_hash: slabman.get_slabs_hash(),
-            }))
-            .await?;*/
-        }
-        net::Message::Inv(_message) => {
-            info!("received inv message");
-            /*
-            let mut list_of_hash: Vec<net::CiphertextHash> = vec![];
-            for slab in message.slabs_hash.iter() {
-                /*if !slabman.get_slabs_hash().contains(slab) {
-                    list_of_hash.push(slab.clone());
-                }*/
-            }
-            send_sx
-                .send(net::Message::GetSlabs(net::GetSlabsMessage {
-                    slabs_hash: list_of_hash,
-                }))
-                .await?;
-            */
-        }
-
-        net::Message::GetSlabs(message) => {
-            info!("received GetSlabs message.");
-            for _slab_hash in message.slabs_hash {
-                /*let slab = slabman.get_slab(&slab_hash);
-                if let Some(slab) = slab {
-                    send_sx.send(net::Message::Slab(slab.clone())).await?;
-                }*/
-            }
-        }
-        net::Message::Slab(message) => {
-            let _slab = net::SlabMessage {
-                nonce: message.nonce,
-                ciphertext: message.ciphertext.clone(),
-            };
-
-            // TODO:  it doesn't have to send inv message to the connection which sent the slab.
-            for (a, _send) in connections.lock().await.iter() {
-                println!("send to {:?}", a);
-                /*send.send(net::Message::Inv(net::InvMessage {
-                    slabs_hash: vec![slab.cipher_hash()],
-                }))
-                .await?;*/
-            }
-        }
-        _ => {}
-    }
-    Ok(())
-}

src/net/protocol/seed_protocol.rs

@@ -1,170 +0,0 @@
-use async_std::sync::Mutex;
-use log::*;
-use smol::{Async, Executor};
-use std::net::{SocketAddr, TcpStream};
-use std::sync::atomic::{AtomicU64, Ordering};
-use std::sync::Arc;
-
-use crate::error::Result;
-use crate::net::messages as net;
-use crate::net::protocol::protocol_base;
-use crate::utility::{get_current_time, AddrsStorage};
-
-type Clock = Arc<AtomicU64>;
-
-pub struct SeedProtocol {
-    send_sx: async_channel::Sender<net::Message>,
-    send_rx: async_channel::Receiver<net::Message>,
-    main_process: Mutex<Option<smol::Task<()>>>,
-
-    seed_addr: SocketAddr,
-    accept_addr: Option<SocketAddr>,
-    stored_addrs: AddrsStorage,
-}
-
-#[derive(PartialEq)]
-enum ProtocolSignal {
-    Waiting,
-    Finished,
-    Timeout,
-}
-
-impl SeedProtocol {
-    pub fn new(
-        seed_addr: SocketAddr,
-        accept_addr: Option<SocketAddr>,
-        stored_addrs: AddrsStorage,
-    ) -> Arc<Self> {
-        let (send_sx, send_rx) = async_channel::unbounded::<net::Message>();
-        Arc::new(Self {
-            send_sx,
-            send_rx,
-            main_process: Mutex::new(None),
-            seed_addr,
-            accept_addr,
-            stored_addrs,
-        })
-    }
-
-    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
-        let executor2 = executor.clone();
-        let self2 = self.clone();
-
-        *self2.main_process.lock().await = Some(executor.spawn(async move {
-            match Async::<TcpStream>::connect(self.seed_addr).await {
-                Ok(stream) => {
-                    let _ = self.handle_connect(stream, executor2).await;
-                }
-                Err(err) => {
-                    warn!("Unable to connect to seed {}: {}", self.seed_addr, err)
-                }
-            }
-        }));
-    }
-
-    pub async fn await_finish(self: Arc<Self>) {
-        let mut process = self.main_process.lock().await;
-        if let Some(process) = &mut *process {
-            process.await;
-        }
-    }
-
-    async fn handle_connect(
-        &self,
-        stream: Async<TcpStream>,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<()> {
-        if let Some(accept_addr) = self.accept_addr {
-            self.send_sx
-                .send(net::Message::Addrs(net::AddrsMessage {
-                    addrs: vec![accept_addr],
-                }))
-                .await?;
-        }
-
-        self.send_sx
-            .send(net::Message::GetAddrs(net::GetAddrsMessage {}))
-            .await?;
-
-        let stream = async_dup::Arc::new(stream);
-
-        // Run event loop
-        match self.event_loop_process(stream, executor).await {
-            Ok(ProtocolSignal::Finished) => {
-                info!("Seed node queried successfully: {}", self.seed_addr);
-            }
-            Ok(ProtocolSignal::Timeout) => {
-                warn!("Seed node timeout: {}", self.seed_addr);
-            }
-            Ok(_) => {
-                unreachable!();
-            }
-            Err(err) => {
-                warn!("Seed disconnected: {} {}", self.seed_addr, err);
-            }
-        }
-        Ok(())
-    }
-
-    async fn event_loop_process(
-        &self,
-        mut stream: net::AsyncTcpStream,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<ProtocolSignal> {
-        let inactivity_timer = net::InactivityTimer::new(executor.clone());
-
-        let clock = Arc::new(AtomicU64::new(0));
-        let _ping_task = executor.spawn(protocol_base::repeat_ping(
-            self.send_sx.clone(),
-            clock.clone(),
-        ));
-
-        loop {
-            let event = net::select_event(&mut stream, &self.send_rx, &inactivity_timer).await?;
-
-            match event {
-                net::Event::Send(message) => {
-                    net::send_message(&mut stream, message).await?;
-                }
-                net::Event::Receive(message) => {
-                    inactivity_timer.reset().await?;
-                    let signal = self.protocol(message, &clock).await?;
-
-                    if signal == ProtocolSignal::Finished {
-                        return Ok(ProtocolSignal::Finished);
-                    }
-                }
-                net::Event::Timeout => return Ok(ProtocolSignal::Timeout),
-            }
-        }
-
-        // These aren't needed since drop() cancels tasks anyway
-        //ping_task.cancel().await;
-        //inactivity_timer.stop().await;
-    }
-
-    async fn protocol(&self, message: net::Message, clock: &Clock) -> Result<ProtocolSignal> {
-        match message {
-            net::Message::Pong => {
-                let current_time = get_current_time();
-                let elapsed = current_time - clock.load(Ordering::Relaxed);
-                info!("Ping time: {} ms", elapsed);
-            }
-            net::Message::Addrs(message) => {
-                info!("received AddrMessage");
-                let mut stored_addrs = self.stored_addrs.lock().await;
-                for addr in message.addrs {
-                    if !stored_addrs.contains(&addr) {
-                        stored_addrs.push(addr);
-                        info!("Added new address to storage {}", addr.to_string());
-                    }
-                }
-
-                return Ok(ProtocolSignal::Finished);
-            }
-            _ => {}
-        }
-
-        Ok(ProtocolSignal::Waiting)
-    }
-}

src/net/protocol/server_protocol.rs

@@ -1,108 +0,0 @@
-use log::*;
-use smol::{Async, Executor};
-use std::net::{SocketAddr, TcpListener};
-use std::sync::Arc;
-
-//use super::protocol;
-use crate::error::Result;
-use crate::net::messages as net;
-use crate::net::protocol::protocol_base;
-use crate::utility::{AddrsStorage, ConnectionsMap};
-
-pub struct ServerProtocol {
-    send_sx: async_channel::Sender<net::Message>,
-    send_rx: async_channel::Receiver<net::Message>,
-    connections: ConnectionsMap,
-
-    accept_addr: SocketAddr,
-    stored_addrs: AddrsStorage,
-}
-
-impl ServerProtocol {
-    pub fn new(
-        connections: ConnectionsMap,
-        accept_addr: SocketAddr,
-        stored_addrs: AddrsStorage,
-    ) -> Arc<Self> {
-        let (send_sx, send_rx) = async_channel::unbounded::<net::Message>();
-        Arc::new(Self {
-            send_sx,
-            send_rx,
-            connections,
-
-            accept_addr,
-            stored_addrs,
-        })
-    }
-
-    pub fn get_send_pipe(&self) -> async_channel::Sender<net::Message> {
-        self.send_sx.clone()
-    }
-
-    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
-        let listener = Async::<TcpListener>::bind(self.accept_addr)?;
-        info!("Listening on {}", listener.get_ref().local_addr()?);
-
-        loop {
-            let (stream, peer_addr) = listener.accept().await?;
-            info!("Accepted client: {}", peer_addr);
-            let stream = async_dup::Arc::new(stream);
-
-            self.connections
-                .lock()
-                .await
-                .insert(peer_addr, self.send_sx.clone());
-
-            let executor2 = executor.clone();
-            let self2 = self.clone();
-
-            executor
-                .spawn(async move {
-                    match self2.clone().event_loop_process(stream, executor2).await {
-                        Ok(()) => {
-                            warn!("Peer {} timeout", peer_addr);
-                        }
-                        Err(err) => {
-                            warn!("Peer {} disconnected: {}", peer_addr, err);
-                        }
-                    }
-                    self2.connections.lock().await.remove(&peer_addr);
-                })
-                .detach();
-        }
-    }
-
-    pub async fn event_loop_process(
-        self: Arc<Self>,
-        mut stream: net::AsyncTcpStream,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<()> {
-        let inactivity_timer = net::InactivityTimer::new(executor.clone());
-
-        loop {
-            let event = net::select_event(&mut stream, &self.send_rx, &inactivity_timer).await?;
-
-            match event {
-                net::Event::Send(message) => {
-                    net::send_message(&mut stream, message).await?;
-                }
-                net::Event::Receive(message) => {
-                    inactivity_timer.reset().await?;
-                    protocol_base::protocol(
-                        message,
-                        &self.stored_addrs,
-                        &self.send_sx,
-                        None,
-                        self.connections.clone(),
-                    )
-                    .await?;
-                }
-                net::Event::Timeout => break,
-            }
-        }
-
-        inactivity_timer.stop().await;
-        // Connection timed out
-        Ok(())
-    }
-}

src/net/protocols/protocol_ping.rs

@@ -39,6 +39,8 @@ impl ProtocolPing {
 
             // Wait for pong, check nonce matches
             let _pong_msg = pong_sub.receive().await?;
+            // TODO: fix pong enum
+            //let _pong_msg = receive_message!(pong_sub, messages::Message::Pong);
             // TODO: add nonce check here
         }
     }

src/net/protocols/protocol_seed.rs

@@ -29,7 +29,6 @@ impl ProtocolSeed {
         self.channel.clone().send(get_addr).await?;
 
         // Receive addresses
-        // TODO: turn this into a macro
         let addrs_msg = receive_message!(addr_sub, messages::Message::Addrs);
         self.hosts.store(addrs_msg.addrs.clone()).await;