added working p2p network

Cargo.toml

@@ -29,11 +29,22 @@ bitvec = "0.18"
 bimap = "0.5.2"
 
 hex = "0.4.2"
+num_enum = "0.5.0"
 
 simplelog = "0.7.4"
 clap = "3.0.0-beta.1"
 failure = "0.1.8"
 failure_derive = "0.1.8"
+log = "0.4"
+ctrlc = "3.1.7"
+
+smol = "1.2.4"
+futures = "0.3.5"
+async-channel = "1.4.2"
+async-executor = "1.4.0"
+async-dup = "1.1.0"
+async-std = "1.6.2"
+easy-parallel = "3.1.0"
 
 [[bin]]
 name = "sha256"
@@ -107,3 +118,7 @@ path = "src/bin/mint.rs"
 name = "jubjub"
 path = "src/bin/jubjub.rs"
 
+[[bin]]
+name = "dfi"
+path = "src/bin/dfi.rs"
+

src/async_serial.rs

@@ -0,0 +1,112 @@
+use futures::prelude::*;
+
+use crate::endian;
+use crate::error::{Error, Result};
+use crate::net::net::AsyncTcpStream;
+use crate::serial::VarInt;
+
+impl VarInt {
+    pub async fn encode_async(&self, stream: &mut AsyncTcpStream) -> Result<usize> {
+        match self.0 {
+            0..=0xFC => {
+                AsyncWriteExt::write_u8(stream, self.0 as u8).await?;
+                Ok(1)
+            }
+            0xFD..=0xFFFF => {
+                AsyncWriteExt::write_u8(stream, 0xFD).await?;
+                AsyncWriteExt::write_u16(stream, self.0 as u16).await?;
+                Ok(3)
+            }
+            0x10000..=0xFFFFFFFF => {
+                AsyncWriteExt::write_u8(stream, 0xFE).await?;
+                AsyncWriteExt::write_u32(stream, self.0 as u32).await?;
+                Ok(5)
+            }
+            _ => {
+                AsyncWriteExt::write_u8(stream, 0xFF).await?;
+                AsyncWriteExt::write_u64(stream, self.0 as u64).await?;
+                Ok(9)
+            }
+        }
+    }
+
+    pub async fn decode_async(stream: &mut AsyncTcpStream) -> Result<Self> {
+        let n = AsyncReadExt::read_u8(stream).await?;
+        match n {
+            0xFF => {
+                let x = AsyncReadExt::read_u64(stream).await?;
+                if x < 0x100000000 {
+                    Err(Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x))
+                }
+            }
+            0xFE => {
+                let x = AsyncReadExt::read_u32(stream).await?;
+                if x < 0x10000 {
+                    Err(Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x as u64))
+                }
+            }
+            0xFD => {
+                let x = AsyncReadExt::read_u16(stream).await?;
+                if x < 0xFD {
+                    Err(Error::NonMinimalVarInt)
+                } else {
+                    Ok(VarInt(x as u64))
+                }
+            }
+            n => Ok(VarInt(n as u64)),
+        }
+    }
+}
+
+macro_rules! async_encoder_fn {
+    ($name:ident, $val_type:ty, $writefn:ident) => {
+        #[inline]
+        pub async fn $name(stream: &mut AsyncTcpStream, v: $val_type) -> Result<()> {
+            stream
+                .write_all(&endian::$writefn(v))
+                .await
+                .map_err(Error::Io)
+        }
+    };
+}
+
+macro_rules! async_decoder_fn {
+    ($name:ident, $val_type:ty, $readfn:ident, $byte_len: expr) => {
+        pub async fn $name(stream: &mut AsyncTcpStream) -> Result<$val_type> {
+            assert_eq!(::std::mem::size_of::<$val_type>(), $byte_len); // size_of isn't a constfn in 1.22
+            let mut val = [0; $byte_len];
+            stream.read_exact(&mut val[..]).await.map_err(Error::Io)?;
+            Ok(endian::$readfn(&val))
+        }
+    };
+}
+
+pub struct AsyncReadExt {}
+
+impl AsyncReadExt {
+    async_decoder_fn!(read_u64, u64, slice_to_u64_le, 8);
+    async_decoder_fn!(read_u32, u32, slice_to_u32_le, 4);
+    async_decoder_fn!(read_u16, u16, slice_to_u16_le, 2);
+
+    pub async fn read_u8(stream: &mut AsyncTcpStream) -> Result<u8> {
+        let mut slice = [0u8; 1];
+        stream.read_exact(&mut slice).await?;
+        Ok(slice[0])
+    }
+}
+
+pub struct AsyncWriteExt {}
+
+impl AsyncWriteExt {
+    async_encoder_fn!(write_u64, u64, u64_to_array_le);
+    async_encoder_fn!(write_u32, u32, u32_to_array_le);
+    async_encoder_fn!(write_u16, u16, u16_to_array_le);
+
+    pub async fn write_u8(stream: &mut AsyncTcpStream, v: u8) -> Result<()> {
+        stream.write_all(&[v]).await.map_err(Error::Io)
+    }
+}

src/bin/dfi.rs

@@ -0,0 +1,162 @@
+#[macro_use]
+extern crate clap;
+use async_channel::unbounded;
+use async_dup::Arc;
+use log::*;
+use async_std::sync::Mutex;
+use async_executor::Executor;
+use easy_parallel::Parallel;
+use std::collections::HashMap;
+use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+
+use sapvi::{Result, SeedProtocol, ServerProtocol, ClientProtocol};
+
+async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
+    let connections = Arc::new(Mutex::new(HashMap::new()));
+
+    let stored_addrs = Arc::new(Mutex::new(Vec::new()));
+
+    let executor2 = executor.clone();
+    let stored_addrs2 = stored_addrs.clone();
+
+    let mut server_task = None;
+    if let Some(accept_addr) = options.accept_addr {
+        let accept_addr = accept_addr.clone();
+
+        let mut protocol = ServerProtocol::new(connections.clone());
+        server_task = Some(executor.spawn(async move {
+            protocol.start(accept_addr, stored_addrs2, executor2).await?;
+            Ok::<(), sapvi::Error>(())
+        }));
+    }
+
+    let mut seed_protocols = Vec::with_capacity(options.seed_addrs.len());
+
+    // Normally we query this from a server
+    let local_addr = options.accept_addr.clone();
+
+    for seed_addr in options.seed_addrs.iter() {
+        let mut protocol = SeedProtocol::new();
+        protocol
+            .start(seed_addr.clone(), local_addr, stored_addrs.clone(), executor.clone())
+            .await;
+        seed_protocols.push(protocol);
+    }
+
+    debug!("Waiting for seed node queries to finish...");
+
+    for seed_protocol in seed_protocols {
+        seed_protocol.await_finish().await;
+    }
+
+    debug!("Seed nodes queried.");
+
+    let accept_addr = options.accept_addr.clone();
+
+    let mut client_slots: Vec<ClientProtocol> = vec![];
+    for i in 0..options.connection_slots {
+        debug!("Starting connection slot {}", i);
+
+        let mut client = ClientProtocol::new(connections.clone());
+        client.start(accept_addr.clone(), stored_addrs.clone(), executor.clone()).await;
+        client_slots.push(client);
+    }
+
+    for remote_addr in options.manual_connects {
+        debug!("Starting connection (manual) to {}", remote_addr);
+
+        let mut client = ClientProtocol::new(connections.clone());
+        client.start_manual(remote_addr, accept_addr.clone(), stored_addrs.clone(), executor.clone()).await;
+        client_slots.push(client);
+    }
+
+    loop {
+        sapvi::sleep(2).await;
+    }
+
+    //server_task.cancel().await;
+    //Ok(())
+}
+
+struct ProgramOptions {
+    accept_addr: Option<SocketAddr>,
+    seed_addrs: Vec<SocketAddr>,
+    manual_connects: Vec<SocketAddr>,
+    connection_slots: u32
+}
+
+impl ProgramOptions {
+    fn load() -> Result<ProgramOptions> {
+        let app = clap_app!(dfi =>
+            (version: "0.1.0")
+            (author: "Amir Taaki <amir@dyne.org>")
+            (about: "Dark node")
+            (@arg ACCEPT: -a --accept +takes_value "Accept address")
+            (@arg SEED_NODES: -s --seeds ... "Seed nodes")
+            (@arg CONNECTS: -c --connect ... "Manual connections")
+            (@arg CONNECT_SLOTS: --slots +takes_value "Connection slots")
+        )
+        .get_matches();
+
+        let accept_addr = if let Some(accept_addr) = app.value_of("ACCEPT") {
+            Some(accept_addr.parse()?)
+        } else {
+            None
+        };
+
+        let mut seed_addrs: Vec<SocketAddr> = vec![];
+        if let Some(seeds) = app.values_of("SEED_NODES") {
+            for seed in seeds {
+                seed_addrs.push(seed.parse()?);
+            }
+        }
+
+        let mut manual_connects: Vec<SocketAddr> = vec![];
+        if let Some(connections) = app.values_of("CONNECTS") {
+            for connect in connections {
+                manual_connects.push(connect.parse()?);
+            }
+        }
+
+        let connection_slots = if let Some(connection_slots) = app.value_of("CONNECT_SLOTS") {
+            connection_slots.parse()?
+        } else {
+            0
+        };
+
+        Ok(ProgramOptions {
+            accept_addr,
+            seed_addrs,
+            manual_connects,
+            connection_slots
+        })
+    }
+}
+
+fn main() -> Result<()> {
+    use simplelog::*;
+    CombinedLogger::init(vec![
+        TermLogger::new(LevelFilter::Debug, Config::default(), TerminalMode::Mixed).unwrap(),
+    ])
+    .unwrap();
+
+    let options = ProgramOptions::load()?;
+
+    let ex = Arc::new(Executor::new());
+    let (signal, shutdown) = unbounded::<()>();
+    let ex2 = ex.clone();
+
+    let (_, result) = Parallel::new()
+        // Run four executor threads.
+        .each(0..3, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        // Run the main future on the current thread.
+        .finish(|| {
+            smol::future::block_on(async move {
+                start(ex2, options).await?;
+                drop(signal);
+                Ok::<(), sapvi::Error>(())
+            })
+        });
+
+    result
+}

src/bin/jubjub.rs

@@ -1,5 +1,5 @@
 use bls12_381::Scalar;
-use sapvi::{BlsStringConversion, Encodable, Decodable, ZKContract, ZKProof};
+use sapvi::{BlsStringConversion, Decodable, Encodable, ZKContract, ZKProof};
 use std::fs::File;
 use std::time::Instant;
 

src/bls_extensions.rs

@@ -1,5 +1,4 @@
 use bls12_381 as bls;
-use rand_core::{OsRng, RngCore};
 use std::io;
 
 use crate::error::{Error, Result};

src/error.rs

@@ -20,6 +20,7 @@ pub enum Error {
     NonMinimalVarInt,
     /// Parsing error
     ParseFailed(&'static str),
+    ParseIntError,
     AsyncChannelError,
     MalformedPacket,
     AddrParseError,
@@ -54,6 +55,7 @@ impl fmt::Display for Error {
             Error::Io(ref err) => fmt::Display::fmt(err, f),
             Error::NonMinimalVarInt => f.write_str("non-minimal varint"),
             Error::ParseFailed(ref err) => write!(f, "parse failed: {}", err),
+            Error::ParseIntError => f.write_str("Parse int error"),
             Error::AsyncChannelError => f.write_str("async_channel error"),
             Error::MalformedPacket => f.write_str("Malformed packet"),
             Error::AddrParseError => f.write_str("Unable to parse address"),
@@ -86,3 +88,27 @@ impl From<bellman::SynthesisError> for Error {
         Error::Groth16Error(err)
     }
 }
+
+impl<T> From<async_channel::SendError<T>> for Error {
+    fn from(_err: async_channel::SendError<T>) -> Error {
+        Error::AsyncChannelError
+    }
+}
+
+impl From<async_channel::RecvError> for Error {
+    fn from(_err: async_channel::RecvError) -> Error {
+        Error::AsyncChannelError
+    }
+}
+
+impl From<std::net::AddrParseError> for Error {
+    fn from(_err: std::net::AddrParseError) -> Error {
+        Error::AddrParseError
+    }
+}
+
+impl From<std::num::ParseIntError> for Error {
+    fn from(_err: std::num::ParseIntError) -> Error {
+        Error::ParseIntError
+    }
+}

src/lib.rs

@@ -2,15 +2,22 @@ use bellman::groth16;
 use bls12_381::{Bls12, Scalar};
 use std::collections::{HashMap, HashSet};
 
+pub mod async_serial;
 pub mod bls_extensions;
 pub mod endian;
 pub mod error;
+pub mod net;
 pub mod serial;
+pub mod utility;
 pub mod vm;
 pub mod vm_serial;
 
 pub use crate::bls_extensions::BlsStringConversion;
 pub use crate::error::{Error, Result};
+pub use crate::net::net::{select_event, send_message, sleep};
+pub use crate::net::protocol::seed_protocol::SeedProtocol;
+pub use crate::net::protocol::server_protocol::ServerProtocol;
+pub use crate::net::protocol::client_protocol::ClientProtocol;
 pub use crate::serial::{Decodable, Encodable};
 pub use crate::vm::{
     AllocType, ConstraintInstruction, CryptoOperation, VariableIndex, VariableRef, ZKVMCircuit,

src/net/mod.rs

@@ -0,0 +1,2 @@
+pub mod net;
+pub mod protocol;

src/net/net.rs

@@ -0,0 +1,379 @@
+use async_dup::Arc;
+use futures::prelude::*;
+use log::*;
+use num_enum::{IntoPrimitive, TryFromPrimitive};
+use smol::Executor;
+use smol::{Async, Timer};
+use std::convert::TryFrom;
+use std::io;
+use std::io::Cursor;
+use std::net::{SocketAddr, TcpStream};
+use std::time::Duration;
+
+use crate::async_serial::{AsyncReadExt, AsyncWriteExt};
+use crate::error::{Error, Result};
+use crate::serial::{serialize, Decodable, Encodable, VarInt};
+
+const MAGIC_BYTES: [u8; 4] = [0xd9, 0xef, 0xb6, 0x7d];
+
+pub type AsyncTcpStream = async_dup::Arc<Async<TcpStream>>;
+pub type Ciphertext = Vec<u8>;
+pub type CiphertextHash = [u8; 32];
+
+// Packets and Message because Rust doesn't allow value
+// aliasing from ADL type enums (which Message uses).
+#[derive(IntoPrimitive, TryFromPrimitive, Copy, Clone)]
+#[repr(u8)]
+pub enum PacketType {
+    Ping = 0,
+    Pong = 1,
+    GetAddrs = 2,
+    Addrs = 3,
+    Sync = 4,
+    Inv = 5,
+    GetSlabs = 6,
+    Slab = 7,
+}
+
+pub enum Message {
+    Ping,
+    Pong,
+    GetAddrs(GetAddrsMessage),
+    Addrs(AddrsMessage),
+    Sync,
+    Inv(InvMessage),
+    GetSlabs(GetSlabsMessage),
+    Slab(SlabMessage),
+}
+
+pub struct GetAddrsMessage {}
+
+pub struct GetSlabsMessage {
+    pub slabs_hash: Vec<[u8; 32]>,
+}
+
+#[derive(Clone)]
+pub struct SlabMessage {
+    pub nonce: [u8; 12],
+    pub ciphertext: Ciphertext,
+}
+
+pub struct InvMessage {
+    pub slabs_hash: Vec<[u8; 32]>,
+}
+
+pub struct AddrsMessage {
+    pub addrs: Vec<SocketAddr>,
+}
+
+impl Encodable for GetSlabsMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.slabs_hash.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for GetSlabsMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            slabs_hash: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Encodable for SlabMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.nonce.encode(&mut s)?;
+        len += self.ciphertext.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for SlabMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            nonce: Decodable::decode(&mut d)?,
+            ciphertext: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Encodable for InvMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.slabs_hash.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for InvMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            slabs_hash: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Encodable for GetAddrsMessage {
+    fn encode<S: io::Write>(&self, mut _s: S) -> Result<usize> {
+        let len = 0;
+        Ok(len)
+    }
+}
+
+impl Decodable for GetAddrsMessage {
+    fn decode<D: io::Read>(mut _d: D) -> Result<Self> {
+        Ok(Self {})
+    }
+}
+
+impl Encodable for AddrsMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.addrs.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for AddrsMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            addrs: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Message {
+    pub fn pack(&self) -> Result<Packet> {
+        match self {
+            Message::Ping => Ok(Packet {
+                command: PacketType::Ping,
+                payload: Vec::new(),
+            }),
+            Message::Pong => Ok(Packet {
+                command: PacketType::Pong,
+                payload: Vec::new(),
+            }),
+            Message::GetAddrs(message) => {
+                let mut payload = Vec::new();
+                message.encode(&mut payload)?;
+                Ok(Packet {
+                    command: PacketType::GetAddrs,
+                    payload,
+                })
+            }
+            Message::Addrs(message) => {
+                let mut payload = Vec::new();
+                message.encode(Cursor::new(&mut payload))?;
+                Ok(Packet {
+                    command: PacketType::Addrs,
+                    payload,
+                })
+            }
+            Message::Sync => {
+                let payload = Vec::new();
+                Ok(Packet {
+                    command: PacketType::Sync,
+                    payload,
+                })
+            }
+            Message::Inv(message) => {
+                let payload = serialize(message);
+                Ok(Packet {
+                    command: PacketType::Inv,
+                    payload,
+                })
+            }
+            Message::GetSlabs(message) => {
+                let payload = serialize(message);
+                Ok(Packet {
+                    command: PacketType::GetSlabs,
+                    payload,
+                })
+            }
+            Message::Slab(message) => {
+                let payload = serialize(message);
+                Ok(Packet {
+                    command: PacketType::Slab,
+                    payload,
+                })
+            }
+        }
+    }
+
+    pub fn unpack(packet: Packet) -> Result<Self> {
+        let cursor = Cursor::new(packet.payload.clone());
+        match packet.command {
+            PacketType::Ping => Ok(Self::Ping),
+            PacketType::Pong => Ok(Self::Pong),
+            PacketType::GetAddrs => Ok(Self::GetAddrs(GetAddrsMessage::decode(cursor)?)),
+            PacketType::Addrs => Ok(Self::Addrs(AddrsMessage::decode(cursor)?)),
+            PacketType::Sync => Ok(Self::Sync),
+            PacketType::Inv => Ok(Self::Inv(InvMessage::decode(cursor)?)),
+            PacketType::GetSlabs => Ok(Self::GetSlabs(GetSlabsMessage::decode(cursor)?)),
+            PacketType::Slab => Ok(Self::Slab(SlabMessage::decode(cursor)?)),
+        }
+    }
+
+    pub fn name(&self) -> &'static str {
+        match self {
+            Message::Ping => "Ping",
+            Message::Pong => "Pong",
+            Message::GetAddrs(_) => "GetAddrs",
+            Message::Addrs(_) => "Addrs",
+            Message::Sync => "Sync",
+            Message::Inv(_) => "Inv",
+            Message::GetSlabs(_) => "GetSlabs",
+            Message::Slab(_) => "Slab",
+        }
+    }
+}
+
+// Packets are the base type read from the network
+// These are converted to messages and passed to event loop
+pub struct Packet {
+    pub command: PacketType,
+    pub payload: Vec<u8>,
+}
+
+pub async fn read_packet(stream: &mut AsyncTcpStream) -> Result<Packet> {
+    // Packets have a 4 byte header of magic digits
+    // This is used for network debugging
+    let mut magic = [0u8; 4];
+    stream.read_exact(&mut magic).await?;
+    //debug!("read magic {:?}", magic);
+    if magic != MAGIC_BYTES {
+        return Err(Error::MalformedPacket);
+    }
+
+    // The type of the message
+    let command = AsyncReadExt::read_u8(stream).await?;
+    //debug!("read command: {}", command);
+    let command = PacketType::try_from(command).map_err(|_| Error::MalformedPacket)?;
+
+    let payload_len = VarInt::decode_async(stream).await?.0 as usize;
+
+    // The message-dependent data (see message types)
+    let mut payload = vec![0u8; payload_len];
+    stream.read_exact(&mut payload).await?;
+    //debug!("read payload");
+
+    Ok(Packet { command, payload })
+}
+
+pub async fn send_packet(stream: &mut AsyncTcpStream, packet: Packet) -> Result<()> {
+    stream.write_all(&MAGIC_BYTES).await?;
+
+    AsyncWriteExt::write_u8(stream, packet.command as u8).await?;
+
+    assert_eq!(std::mem::size_of::<usize>(), std::mem::size_of::<u64>());
+    VarInt(packet.payload.len() as u64)
+        .encode_async(stream)
+        .await?;
+
+    stream.write_all(&packet.payload).await?;
+
+    Ok(())
+}
+
+async fn receive_message(stream: &mut AsyncTcpStream) -> Result<Message> {
+    let packet = read_packet(stream).await?;
+    let message = Message::unpack(packet)?;
+    debug!("received Message::{}", message.name());
+    Ok(message)
+}
+
+pub async fn send_message(stream: &mut AsyncTcpStream, message: Message) -> Result<()> {
+    debug!("sending Message::{}", message.name());
+    let packet = message.pack()?;
+    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;
+}
+
+// Used for ping pong loop timer
+pub struct InactivityTimer {
+    reset_sender: async_channel::Sender<()>,
+    timeout_receiver: async_channel::Receiver<()>,
+    task: smol::Task<()>,
+}
+
+impl InactivityTimer {
+    pub fn new(executor: Arc<Executor<'_>>) -> Self {
+        let (reset_sender, reset_receiver) = async_channel::bounded::<()>(1);
+        let (timeout_sender, timeout_receiver) = async_channel::bounded::<()>(1);
+
+        let task = executor.spawn(async {
+            match Self::_start(reset_receiver, timeout_sender).await {
+                Ok(()) => {}
+                Err(err) => error!("InactivityTimer fatal error {}", err),
+            }
+        });
+
+        Self {
+            reset_sender,
+            timeout_receiver,
+            task,
+        }
+    }
+
+    pub async fn stop(self) {
+        self.task.cancel().await;
+    }
+
+    // This loop basically waits for 10 secs. If it doesn't
+    // receive a signal that something happened then it will
+    // send a timeout signal. This will wakeup the main event loop
+    // and the connection will be dropped.
+    async fn _start(
+        reset_rx: async_channel::Receiver<()>,
+        timeout_sx: async_channel::Sender<()>,
+    ) -> Result<()> {
+        loop {
+            let is_awake = futures::select! {
+                _ = reset_rx.recv().fuse() => true,
+                _ = sleep(10).fuse() => false
+            };
+
+            if !is_awake {
+                warn!("InactivityTimer timeout");
+                timeout_sx.send(()).await?;
+            }
+        }
+    }
+
+    pub async fn reset(&self) -> Result<()> {
+        self.reset_sender.send(()).await?;
+        Ok(())
+    }
+
+    pub async fn wait_for_wakeup(&self) -> Result<()> {
+        Ok(self.timeout_receiver.recv().await?)
+    }
+}

src/net/protocol/client_protocol.rs

@@ -0,0 +1,265 @@
+use std::net::{SocketAddr, TcpStream};
+use std::sync::atomic::AtomicU64;
+use log::*;
+use smol::{Async, Executor};
+use async_dup::Arc;
+use rand::seq::SliceRandom;
+
+use crate::error::Result;
+use crate::net::net;
+use crate::net::protocol::protocol_base;
+use crate::utility::{AddrsStorage, ConnectionsMap};
+
+pub struct ClientProtocol {
+    send_sx: async_channel::Sender<net::Message>,
+    send_rx: async_channel::Receiver<net::Message>,
+    connections: ConnectionsMap,
+    main_process: Option<smol::Task<()>>,
+}
+
+impl ClientProtocol {
+    pub fn new(connections: ConnectionsMap) -> Self {
+        let (send_sx, send_rx) = async_channel::unbounded::<net::Message>();
+        Self {
+            send_sx,
+            send_rx,
+            connections,
+            main_process: None
+        }
+    }
+
+    pub fn get_send_pipe(&self) -> async_channel::Sender<net::Message> {
+        self.send_sx.clone()
+    }
+
+    async fn fetch_random_addr(
+        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(
+        &mut self,
+        accept_addr: Option<SocketAddr>,
+        stored_addrs: AddrsStorage,
+        executor: Arc<Executor<'_>>
+    ) {
+        let connections = self.connections.clone();
+        let (send_sx, send_rx) = (self.send_sx.clone(), self.send_rx.clone());
+
+        let executor2 = executor.clone();
+
+        self.main_process = Some(executor.spawn(async move {
+            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; }
+                }
+
+                debug!("Attempting connect to {}", addr);
+
+                Self::try_connect_process(addr, connections.clone(), accept_addr.clone(), stored_addrs.clone(), (send_sx.clone(), send_rx.clone()), executor2.clone()).await;
+
+                // TODO: Fix this
+                net::sleep(2).await;
+            }
+        }));
+    }
+
+    pub async fn start_manual(
+        &mut self,
+        remote_addr: SocketAddr,
+        accept_addr: Option<SocketAddr>,
+        stored_addrs: AddrsStorage,
+        executor: Arc<Executor<'_>>
+        ) {
+        let connections = self.connections.clone();
+        let (send_sx, send_rx) = (self.send_sx.clone(), self.send_rx.clone());
+
+        let executor2 = executor.clone();
+
+        self.main_process = Some(executor.spawn(async move {
+            loop {
+                for _ in 0..4 {
+                    debug!("Attempting connect to {}", remote_addr);
+
+                    Self::try_connect_process(remote_addr, connections.clone(), accept_addr.clone(), stored_addrs.clone(), (send_sx.clone(), send_rx.clone()), executor2.clone()).await;
+                }
+                net::sleep(2).await;
+            }
+        }));
+    }
+
+    pub async fn try_connect_process(
+        address: SocketAddr,
+        connections: ConnectionsMap,
+        accept_addr: Option<SocketAddr>,
+        stored_addrs: AddrsStorage,
+        (send_sx, send_rx): (
+            async_channel::Sender<net::Message>,
+            async_channel::Receiver<net::Message>,
+        ),
+        executor: Arc<Executor<'_>>
+    ) {
+        match Async::<TcpStream>::connect(address.clone()).await {
+            Ok(stream) => {
+                let _ = Self::handle_connect(
+                    stream,
+                    stored_addrs.clone(),
+                    connections,
+                    address,
+                    (send_sx.clone(), send_rx.clone()),
+                    accept_addr,
+                    executor
+                )
+                .await;
+            }
+            Err(_err) => {  warn!(
+                              "Unable to connect to addr {:?}: {}",
+                             address, _err
+                             ); }
+        }
+    }
+
+    async fn handle_connect(
+        stream: Async<TcpStream>,
+        stored_addrs: AddrsStorage,
+        connections: ConnectionsMap,
+        address: SocketAddr,
+        (send_sx, send_rx): (
+            async_channel::Sender<net::Message>,
+            async_channel::Receiver<net::Message>,
+        ),
+        accept_addr: Option<SocketAddr>,
+        executor: Arc<Executor<'_>>
+    ) -> Result<()> {
+        debug!("Connected to {}", address);
+
+        let stream = async_dup::Arc::new(stream);
+        connections
+            .lock()
+            .await
+            .insert(address.clone(), send_sx.clone());
+
+        // Run event loop
+        match Self::event_loop_process(
+            stream,
+            stored_addrs,
+            (send_sx, send_rx),
+            accept_addr,
+            connections.clone(),
+            executor,
+        )
+        .await
+        {
+            Ok(()) => {
+                warn!("Server timeout");
+            }
+            Err(err) => {
+                warn!("Server disconnected: {}", err);
+            }
+        }
+        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(
+        mut stream: net::AsyncTcpStream,
+        stored_addrs: AddrsStorage,
+        (send_sx, send_rx): (
+            async_channel::Sender<net::Message>,
+            async_channel::Receiver<net::Message>,
+        ),
+        accept_addr: Option<SocketAddr>,
+        connections: ConnectionsMap,
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        let inactivity_timer = net::InactivityTimer::new(executor.clone());
+
+        let clock = Arc::new(AtomicU64::new(0));
+        let send_sx2 = 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) = accept_addr {
+            send_addr_task = Some(executor.spawn(Self::send_addr(send_sx.clone(), accept_addr)));
+        }
+
+        loop {
+            let event = net::select_event(&mut stream, &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,
+                        &stored_addrs,
+                        &send_sx,
+                        Some(&clock),
+                        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

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

src/net/protocol/protocol_base.rs

@@ -0,0 +1,109 @@
+use std::sync::atomic::Ordering;
+use log::*;
+
+use crate::net::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

@@ -0,0 +1,183 @@
+use async_dup::Arc;
+use log::*;
+use smol::{Async, Executor};
+use std::net::{SocketAddr, TcpStream};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use crate::error::Result;
+use crate::net::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: Option<smol::Task<()>>,
+}
+
+#[derive(PartialEq)]
+enum ProtocolSignal {
+    Waiting,
+    Finished,
+    Timeout
+}
+
+impl SeedProtocol {
+    pub fn new() -> Self {
+        let (send_sx, send_rx) = async_channel::unbounded::<net::Message>();
+        Self {
+            send_sx,
+            send_rx,
+            main_process: None,
+        }
+    }
+
+    pub async fn start(
+        &mut self,
+        seed_addr: SocketAddr,
+        local_addr: Option<SocketAddr>,
+        stored_addrs: AddrsStorage,
+        executor: Arc<Executor<'_>>,
+    ) {
+        let (send_sx, send_rx) = (self.send_sx.clone(), self.send_rx.clone());
+        let ex = executor.clone();
+        self.main_process = Some(ex.spawn(async move {
+            match Async::<TcpStream>::connect(seed_addr.clone()).await {
+                Ok(stream) => {
+                    let _ = Self::handle_connect(
+                        stream,
+                        &stored_addrs,
+                        seed_addr.clone(),
+                        local_addr,
+                        (send_sx.clone(), send_rx.clone()),
+                        executor.clone(),
+                    )
+                    .await;
+                }
+                Err(err) => { warn!("Unable to connect to seed {}: {}", seed_addr, err) },
+            }
+        }));
+    }
+
+    pub async fn await_finish(self) {
+        if let Some(process) = self.main_process {
+            process.await;
+        }
+    }
+
+    async fn handle_connect(
+        stream: Async<TcpStream>,
+        stored_addrs: &AddrsStorage,
+        seed_addr: SocketAddr,
+        local_addr: Option<SocketAddr>,
+        (send_sx, send_rx): (
+            async_channel::Sender<net::Message>,
+            async_channel::Receiver<net::Message>,
+        ),
+        executor: Arc<Executor<'_>>,
+    ) -> Result<()> {
+        if let Some(local_addr) = local_addr {
+            send_sx
+                .send(net::Message::Addrs(net::AddrsMessage { addrs: vec![local_addr] }))
+                .await?;
+        }
+
+        send_sx
+            .send(net::Message::GetAddrs(net::GetAddrsMessage {}))
+            .await?;
+
+        let stream = Arc::new(stream);
+
+        // Run event loop
+        match Self::event_loop_process(
+            stream,
+            stored_addrs.clone(),
+            (send_sx, send_rx),
+            executor,
+        )
+        .await
+        {
+            Ok(ProtocolSignal::Finished) => {
+                info!("Seed node queried successfully: {}", seed_addr);
+            }
+            Ok(ProtocolSignal::Timeout) => {
+                warn!("Seed node timeout: {}", seed_addr);
+            }
+            Ok(_) => { unreachable!(); }
+            Err(err) => {
+                warn!("Seed disconnected: {} {}", seed_addr, err);
+            }
+        }
+        Ok(())
+    }
+
+    async fn event_loop_process(
+        mut stream: net::AsyncTcpStream,
+        stored_addrs: AddrsStorage,
+        (send_sx, send_rx): (
+            async_channel::Sender<net::Message>,
+            async_channel::Receiver<net::Message>,
+        ),
+        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(send_sx.clone(), clock.clone()));
+
+        loop {
+            let event = net::select_event(&mut stream, &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, &stored_addrs, &send_sx, &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(
+        message: net::Message,
+        stored_addrs: &AddrsStorage,
+        _send_sx: &async_channel::Sender<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 = 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

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

src/serial.rs

@@ -1,7 +1,7 @@
 use bls12_381 as bls;
 use std::borrow::Cow;
 use std::io::{Cursor, Read, Write};
-use std::rc::Rc;
+use std::net::{IpAddr, SocketAddr};
 use std::{io, mem};
 
 use crate::endian;
@@ -411,6 +411,61 @@ macro_rules! impl_vec {
     };
 }
 impl_vec!(bls::Scalar);
+impl_vec!(SocketAddr);
+impl_vec!([u8; 32]);
+
+impl Encodable for IpAddr {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        match self {
+            IpAddr::V4(ip) => {
+                let version: u8 = 4;
+                len += version.encode(&mut s)?;
+                len += ip.octets().encode(s)?;
+            }
+            IpAddr::V6(ip) => {
+                let version: u8 = 6;
+                len += version.encode(&mut s)?;
+                len += ip.octets().encode(s)?;
+            }
+        }
+        Ok(len)
+    }
+}
+
+impl Decodable for IpAddr {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let version: u8 = Decodable::decode(&mut d)?;
+        match version {
+            4 => {
+                let addr: [u8; 4] = Decodable::decode(&mut d)?;
+                Ok(IpAddr::from(addr))
+            }
+            6 => {
+                let addr: [u8; 16] = Decodable::decode(&mut d)?;
+                Ok(IpAddr::from(addr))
+            }
+            _ => Err(Error::ParseFailed("couldn't decode IpAddr")),
+        }
+    }
+}
+
+impl Encodable for SocketAddr {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.ip().encode(&mut s)?;
+        len += self.port().encode(s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for SocketAddr {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let ip = Decodable::decode(&mut d)?;
+        let port: u16 = Decodable::decode(d)?;
+        Ok(SocketAddr::new(ip, port))
+    }
+}
 
 pub fn encode_with_size<S: io::Write>(data: &[u8], mut s: S) -> Result<usize> {
     let vi_len = VarInt(data.len() as u64).encode(&mut s)?;

src/utility.rs

@@ -0,0 +1,89 @@
+use async_dup::Arc;
+use std::collections::HashMap;
+use std::fs::OpenOptions;
+use std::io::prelude::*;
+use std::net::SocketAddr;
+use std::sync::atomic::AtomicU64;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use rand::seq::SliceRandom;
+use smol::{Executor, Task};
+
+//use crate::{net, serial, Channel, ClientProtocol, Result, SlabsManagerSafe};
+use crate::{net::net, serial, Result};
+
+pub type ConnectionsMap = async_dup::Arc<
+    async_std::sync::Mutex<HashMap<SocketAddr, async_channel::Sender<net::Message>>>,
+>;
+
+pub type AddrsStorage = async_dup::Arc<async_std::sync::Mutex<Vec<SocketAddr>>>;
+
+pub type Clock = async_dup::Arc<AtomicU64>;
+
+pub fn get_current_time() -> u64 {
+    let start = SystemTime::now();
+    let since_the_epoch = start
+        .duration_since(UNIX_EPOCH)
+        .expect("Incorrect system clock: time went backwards");
+    let in_ms =
+        since_the_epoch.as_secs() * 1000 + since_the_epoch.subsec_nanos() as u64 / 1_000_000;
+    return in_ms;
+}
+
+pub fn save_to_addrs_store(stored_addrs: &Vec<SocketAddr>) -> Result<()> {
+    let mut writer = OpenOptions::new()
+        .write(true)
+        .create(true)
+        .open("addrs.dps")?;
+    let buffer = serial::serialize(stored_addrs);
+    writer.write_all(&buffer)?;
+    Ok(())
+}
+
+pub fn load_stored_addrs() -> Result<Vec<SocketAddr>> {
+    let mut reader = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .open("addrs.dps")?;
+    let mut buffer = Vec::new();
+    reader.read_to_end(&mut buffer)?;
+    if !buffer.is_empty() {
+        let addrs: Vec<SocketAddr> = serial::deserialize(&buffer)?;
+        Ok(addrs)
+    } else {
+        Ok(vec![])
+    }
+}
+
+pub async fn start_connections_process(
+    //slabman: SlabsManagerSafe,
+    stored_addrs: Vec<SocketAddr>,
+    connections: ConnectionsMap,
+    _accept_addr: SocketAddr,
+    _channel_secret: [u8; 32],
+    executor: Arc<Executor<'_>>,
+) -> Vec<Task<()>> {
+    let mut tasks: Vec<Task<()>> = vec![];
+    for _ in 0..10 {
+        let connections_cloned = connections.clone();
+        let stored_addrs_cloned = stored_addrs.clone();
+        //let slabman_cloned = slabman.clone();
+        //let channel_secret = channel_secret.clone();
+        let task = executor.spawn(async move {
+            loop {
+                let addr = stored_addrs_cloned.choose(&mut rand::thread_rng()).unwrap();
+                if !connections_cloned.lock().await.contains_key(addr) {
+                    /*let mut protocol =
+                        ClientProtocol::new(connections_cloned.clone(), slabman_cloned.clone());
+                    protocol
+                        .start(addr.clone(), accept_addr.clone(), &channel_secret)
+                        .await;
+                        */
+                }
+            }
+        });
+        tasks.push(task);
+    }
+    tasks
+}

src/vm.rs

@@ -1,9 +1,4 @@
-use bellman::{
-    gadgets::{
-        Assignment,
-    },
-    groth16, Circuit, ConstraintSystem, SynthesisError,
-};
+use bellman::{gadgets::Assignment, groth16, Circuit, ConstraintSystem, SynthesisError};
 use bls12_381::Bls12;
 use bls12_381::Scalar;
 use ff::{Field, PrimeField};

src/vm_serial.rs

@@ -1,8 +1,7 @@
 use crate::error::{Error, Result};
 use crate::serial::{Decodable, Encodable, ReadExt, VarInt};
 use crate::vm::{
-    AllocType, ConstraintInstruction, CryptoOperation, VariableIndex, VariableRef, 
-    ZKVirtualMachine,
+    AllocType, ConstraintInstruction, CryptoOperation, VariableIndex, VariableRef, ZKVirtualMachine,
 };
 use crate::{impl_vec, ZKContract, ZKProof};
 use bellman::groth16;