add outbound_session and detailed logging info

src/bin/dfi.rs

@@ -309,7 +309,11 @@ impl ProgramOptions {
             .to_path_buf(),
         );
 
-        let skip_seed_sync = if app.is_present("DISABLE_SEED") { true } else { false };
+        let skip_seed_sync = if app.is_present("DISABLE_SEED") {
+            true
+        } else {
+            false
+        };
 
         let rpc_port = if let Some(rpc_port) = app.value_of("RPC_PORT") {
             rpc_port.parse()?
@@ -322,12 +326,12 @@ impl ProgramOptions {
                 inbound: accept_addr,
                 outbound_connections: connection_slots,
                 connect_timeout_seconds: 10,
-                channel_handshake_seconds: 2,
+                channel_handshake_seconds: 4,
                 channel_heartbeat_seconds: 10,
                 external_addr: accept_addr,
                 peers: manual_connects,
                 seeds: seed_addrs,
-                skip_seed_sync
+                skip_seed_sync,
             },
             log_path,
             rpc_port,
@@ -340,8 +344,12 @@ fn main() -> Result<()> {
 
     let options = ProgramOptions::load()?;
 
+    let logger_config = ConfigBuilder::new()
+        .set_time_format_str("%T%.6f")
+        .build();
+
     CombinedLogger::init(vec![
-        TermLogger::new(LevelFilter::Debug, Config::default(), TerminalMode::Mixed).unwrap(),
+        TermLogger::new(LevelFilter::Debug, logger_config, TerminalMode::Mixed).unwrap(),
         WriteLogger::new(
             LevelFilter::Debug,
             Config::default(),

src/net/channel.rs

@@ -1,7 +1,6 @@
 use async_std::sync::Mutex;
 use futures::io::{ReadHalf, WriteHalf};
 use futures::AsyncReadExt;
-use futures::FutureExt;
 use log::*;
 use smol::{Async, Executor};
 
@@ -17,7 +16,7 @@ use crate::net::message_subscriber::{
 };
 use crate::net::messages;
 use crate::net::settings::SettingsPtr;
-use crate::system::{Subscriber, SubscriberPtr, Subscription};
+use crate::system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription};
 
 pub type ChannelPtr = Arc<Channel>;
 
@@ -27,6 +26,7 @@ pub struct Channel {
     address: SocketAddr,
     message_subscriber: MessageSubscriberPtr,
     stop_subscriber: SubscriberPtr<NetError>,
+    receive_task: StoppableTaskPtr,
     stopped: AtomicBool,
     settings: SettingsPtr,
 }
@@ -42,95 +42,148 @@ impl Channel {
             address,
             message_subscriber: MessageSubscriber::new(),
             stop_subscriber: Subscriber::new(),
+            receive_task: StoppableTask::new(),
             stopped: AtomicBool::new(false),
             settings,
         })
     }
 
     pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
-        executor.spawn(self.receive_loop()).detach();
+        debug!(target: "net", "Channel::start() [START, address={}]", self.address());
+        let self2 = self.clone();
+        self.receive_task.clone().start(
+            self.clone().receive_loop(),
+            // Ignore stop handler
+            |result| self2.handle_stop(result),
+            NetError::ServiceStopped,
+            executor,
+        );
+        debug!(target: "net", "Channel::start() [END, address={}]", self.address());
+    }
+
+    pub async fn stop(&self) {
+        debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
+        assert_eq!(self.stopped.load(Ordering::Relaxed), false);
+        self.stopped.store(false, Ordering::Relaxed);
+        let stop_err = Arc::new(NetError::ChannelStopped);
+        self.stop_subscriber.notify(stop_err).await;
+        self.receive_task.stop().await;
+        debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
+    }
+
+    pub async fn subscribe_stop(&self) -> Subscription<NetError> {
+        debug!(target: "net",
+            "Channel::subscribe_stop() [START, address={}]",
+            self.address()
+        );
+        // TODO: this should check the stopped status
+        // Call to receive should return ChannelStopped on newly created sub
+        let sub = self.stop_subscriber.clone().subscribe().await;
+        debug!(target: "net",
+            "Channel::subscribe_stop() [END, address={}]",
+            self.address()
+        );
+        sub
     }
 
     pub async fn send(self: Arc<Self>, message: messages::Message) -> NetResult<()> {
+        let packet_type = message.packet_type();
+        debug!(target: "net",
+            "Channel::send() [START, pkt_type={:?}, address={}]",
+            packet_type,
+            self.address()
+        );
         if self.stopped.load(Ordering::Relaxed) {
             return Err(NetError::ChannelStopped);
         }
 
         // Catch failure and stop channel, return a net error
-        match messages::send_message(&mut *self.writer.lock().await, message).await {
+        let result = match messages::send_message(&mut *self.writer.lock().await, message).await {
             Ok(()) => Ok(()),
             Err(err) => {
-                error!("Channel error {}, closing {}", err, self.address());
+                error!("Channel send error for [{}]: {}", self.address(), err);
                 self.stop().await;
                 Err(NetError::ChannelStopped)
             }
-        }
-    }
-
-    pub fn address(&self) -> SocketAddr {
-        self.address
+        };
+        debug!(target: "net",
+            "Channel::send() [END, pkt_type={:?}, address={}]",
+            packet_type,
+            self.address()
+        );
+        result
     }
 
     pub async fn subscribe_msg(
         self: Arc<Self>,
         packet_type: messages::PacketType,
     ) -> MessageSubscription {
-        self.message_subscriber.clone().subscribe(packet_type).await
+        debug!(target: "net",
+            "Channel::subscribe_msg() [START, pkt_type={:?}, address={}]",
+            packet_type,
+            self.address()
+        );
+        let sub = self.message_subscriber.clone().subscribe(packet_type).await;
+        debug!(target: "net",
+            "Channel::subscribe_msg() [END, pkt_type={:?}, address={}]",
+            packet_type,
+            self.address()
+        );
+        sub
     }
 
-    pub async fn subscribe_stop(self: Arc<Self>) -> Subscription<NetError> {
-        self.stop_subscriber.clone().subscribe().await
-    }
-
-    pub async fn stop(&self) {
-        self.stopped.store(false, Ordering::Relaxed);
-        let stop_err = Arc::new(NetError::ChannelStopped);
-        self.stop_subscriber.notify(stop_err).await;
+    pub fn address(&self) -> SocketAddr {
+        self.address
     }
 
     fn is_eof_error(err: &error::Error) -> bool {
         match err {
             error::Error::Io(io_err) => io_err.kind() == std::io::ErrorKind::UnexpectedEof,
-            _ => false
+            _ => false,
         }
     }
 
     async fn receive_loop(self: Arc<Self>) -> NetResult<()> {
-        let stop_sub = self.clone().subscribe_stop().await;
+        debug!(target: "net",
+            "Channel::receive_loop() [START, address={}]",
+            self.address()
+        );
         let reader = &mut *self.reader.lock().await;
 
         loop {
-            let message_result = futures::select! {
-                message_result = messages::receive_message(reader).fuse() => {
-                    match message_result {
-                        Ok(message) => Ok(Arc::new(message)),
-                        Err(err) => {
-                            if Self::is_eof_error(&err) {
-                                info!("Closing channel {} disconnected", self.address());
-                            } else {
-                                error!("Read error on channel: {}", err);
-                            }
-                            self.stop().await;
-                            Err(NetError::ChannelStopped)
-                        }
+            let message_result = messages::receive_message(reader).await;
+            let message = match message_result {
+                Ok(message) => Arc::new(message),
+                Err(err) => {
+                    if Self::is_eof_error(&err) {
+                        info!("Channel {} disconnected", self.address());
+                    } else {
+                        error!("Read error on channel: {}", err);
                     }
-                }
-                stop_err = stop_sub.receive().fuse() => {
-                    Err(*stop_err)
+                    debug!(target: "net",
+                        "Channel::receive_loop() stopping channel {}",
+                        self.address()
+                    );
+                    self.stop().await;
+                    return Err(NetError::ChannelStopped);
                 }
             };
 
-            // Save status before using the message
-            let stopped = message_result.is_err();
-
             // Send result to our subscribers
-            self.message_subscriber.notify(message_result).await;
+            self.message_subscriber.notify(Ok(message)).await;
+        }
+    }
 
-            // If channel is stopped, timed out or any other error then terminate loop.
-            if stopped {
-                break;
+    async fn handle_stop(self: Arc<Self>, result: NetResult<()>) {
+        debug!(target: "net", "Channel::handle_stop() [START, address={}]", self.address());
+        match result {
+            Ok(()) => panic!("Channel task should never complete without error status"),
+            Err(err) => {
+                // Send this error to all channel subscribers
+                let result = Err(err);
+                self.message_subscriber.notify(result).await;
             }
         }
-        Ok(())
+        debug!(target: "net", "Channel::handle_stop() [END, address={}]", self.address());
     }
 }

src/net/messages.rs

@@ -22,7 +22,7 @@ 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, PartialEq, Eq, Hash)]
+#[derive(IntoPrimitive, TryFromPrimitive, Copy, Clone, PartialEq, Eq, Hash, Debug)]
 #[repr(u8)]
 pub enum PacketType {
     Ping = 0,
@@ -343,16 +343,16 @@ pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet>
     // Packets have a 4 byte header of magic digits
     // This is used for network debugging
     let mut magic = [0u8; 4];
-    //debug!("reading magic...");
+    debug!(target: "net", "reading magic...");
     stream.read_exact(&mut magic).await?;
-    //debug!("read magic {:?}", magic);
+    debug!(target: "net", "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);
+    //debug!(target: "net", "read command: {}", command);
     let command = PacketType::try_from(command).map_err(|_| Error::MalformedPacket)?;
 
     let payload_len = VarInt::decode_async(stream).await?.0 as usize;
@@ -360,7 +360,7 @@ pub async fn read_packet<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Packet>
     // The message-dependent data (see message types)
     let mut payload = vec![0u8; payload_len];
     stream.read_exact(&mut payload).await?;
-    //debug!("read payload");
+    //debug!(target: "net", "read payload");
 
     Ok(Packet { command, payload })
 }
@@ -383,12 +383,12 @@ pub async fn send_packet<W: AsyncWrite + Unpin>(stream: &mut W, packet: Packet)
 pub async fn receive_message<R: AsyncRead + Unpin>(stream: &mut R) -> Result<Message> {
     let packet = read_packet(stream).await?;
     let message = Message::unpack(packet)?;
-    debug!("received Message::{}", message.name());
+    debug!(target: "net", "received Message::{}", message.name());
     Ok(message)
 }
 
 pub async fn send_message<W: AsyncWrite + Unpin>(stream: &mut W, message: Message) -> Result<()> {
-    debug!("sending Message::{}", message.name());
+    debug!(target: "net", "sending Message::{}", message.name());
     let packet = message.pack()?;
     send_packet(stream, packet).await
 }

src/net/p2p.rs

@@ -1,12 +1,14 @@
 use async_executor::Executor;
 use async_std::sync::Mutex;
+use log::*;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use crate::net::error::NetResult;
-use crate::net::sessions::{InboundSession, SeedSession};
+use crate::net::error::{NetError, NetResult};
+use crate::net::sessions::{InboundSession, OutboundSession, SeedSession};
 use crate::net::{Channel, ChannelPtr, Hosts, HostsPtr, Settings, SettingsPtr};
+use crate::system::{Subscriber, SubscriberPtr, Subscription};
 
 pub type Pending<T> = Mutex<HashMap<SocketAddr, Arc<T>>>;
 
@@ -14,6 +16,8 @@ pub type P2pPtr = Arc<P2p>;
 
 pub struct P2p {
     pending_channels: Pending<Channel>,
+    // Used internally
+    stop_subscriber: SubscriberPtr<NetError>,
     hosts: HostsPtr,
     settings: SettingsPtr,
 }
@@ -23,6 +27,7 @@ impl P2p {
         let settings = Arc::new(settings);
         Arc::new(Self {
             pending_channels: Mutex::new(HashMap::new()),
+            stop_subscriber: Subscriber::new(),
             hosts: Hosts::new(settings.clone()),
             settings,
         })
@@ -30,12 +35,15 @@ impl P2p {
 
     /// Invoke startup and seeding sequence. Call from constructing thread.
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "P2p::start() [BEGIN]");
         // Start manual connections
 
         // Start seed session
         let seed = SeedSession::new(Arc::downgrade(&self));
+        // This will block until all seed queries have finished
         seed.start(executor.clone()).await?;
 
+        debug!(target: "net", "P2p::start() [END]");
         Ok(())
     }
 
@@ -43,7 +51,19 @@ impl P2p {
     /// call after start() is invoked.
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         let inbound = InboundSession::new(Arc::downgrade(&self));
-        inbound.start(executor.clone())?;
+        inbound.clone().start(executor.clone())?;
+
+        let outbound = OutboundSession::new(Arc::downgrade(&self));
+        outbound.clone().start(executor.clone()).await?;
+
+        let stop_sub = self.subscribe_stop().await;
+        // Wait for stop signal
+        stop_sub.receive().await;
+
+        // Stop the sessions
+        inbound.stop().await;
+        outbound.stop().await;
+
         Ok(())
     }
 
@@ -71,4 +91,8 @@ impl P2p {
     pub fn hosts(&self) -> HostsPtr {
         self.hosts.clone()
     }
+
+    async fn subscribe_stop(&self) -> Subscription<NetError> {
+        self.stop_subscriber.clone().subscribe().await
+    }
 }

src/net/protocols/protocol_address.rs

@@ -1,3 +1,4 @@
+use log::*;
 use smol::Executor;
 use std::sync::Arc;
 
@@ -20,11 +21,12 @@ impl ProtocolAddress {
             channel: channel.clone(),
             hosts,
             settings,
-            jobsman: ProtocolJobsManager::new(channel),
+            jobsman: ProtocolJobsManager::new("ProtocolAddress", channel),
         })
     }
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
+        debug!(target: "net", "ProtocolAddress::start() [START]");
         self.jobsman.clone().start(executor.clone());
         self.jobsman
             .clone()
@@ -38,9 +40,11 @@ impl ProtocolAddress {
         // Send get_address message
         let get_addrs = messages::Message::GetAddrs(messages::GetAddrsMessage {});
         let _ = self.channel.clone().send(get_addrs).await;
+        debug!(target: "net", "ProtocolAddress::start() [END]");
     }
 
     async fn handle_receive_addrs(self: Arc<Self>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolAddress::handle_receive_addrs() [START]");
         let addrs_sub = self
             .channel
             .clone()
@@ -50,11 +54,13 @@ impl ProtocolAddress {
         loop {
             let addrs_msg = receive_message!(addrs_sub, messages::Message::Addrs);
 
+            debug!(target: "net", "ProtocolAddress::handle_receive_addrs() storing address in hosts");
             self.hosts.store(addrs_msg.addrs.clone()).await;
         }
     }
 
     async fn handle_receive_get_addrs(self: Arc<Self>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() [START]");
         let get_addrs_sub = self
             .channel
             .clone()
@@ -64,9 +70,12 @@ impl ProtocolAddress {
         loop {
             let _get_addrs = receive_message!(get_addrs_sub, messages::Message::GetAddrs);
 
+            debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() received GetAddrs message");
+
             let addrs = messages::Message::Addrs(messages::AddrsMessage {
                 addrs: self.hosts.load_all().await,
             });
+            debug!(target: "net", "ProtocolAddress::handle_receive_get_addrs() sending Addrs message");
             self.channel.clone().send(addrs).await?;
         }
     }

src/net/protocols/protocol_jobs_manager.rs

@@ -1,5 +1,6 @@
 use async_std::sync::Mutex;
 use futures::Future;
+use log::*;
 use smol::Task;
 use std::sync::Arc;
 
@@ -10,13 +11,15 @@ use crate::system::ExecutorPtr;
 pub type ProtocolJobsManagerPtr = Arc<ProtocolJobsManager>;
 
 pub struct ProtocolJobsManager {
+    name: &'static str,
     channel: ChannelPtr,
     tasks: Mutex<Vec<Task<NetResult<()>>>>,
 }
 
 impl ProtocolJobsManager {
-    pub fn new(channel: ChannelPtr) -> Arc<Self> {
+    pub fn new(name: &'static str, channel: ChannelPtr) -> Arc<Self> {
         Arc::new(Self {
+            name,
             channel,
             tasks: Mutex::new(Vec::new()),
         })
@@ -44,6 +47,11 @@ impl ProtocolJobsManager {
     }
 
     async fn close_all_tasks(self: Arc<Self>) {
+        debug!(target: "net",
+            "ProtocolJobsManager::close_all_tasks() [START, name={}, addr={}]",
+            self.name,
+            self.channel.address()
+        );
         let tasks = std::mem::take(&mut *self.tasks.lock().await);
         for task in tasks {
             let _ = task.cancel().await;

src/net/protocols/protocol_ping.rs

@@ -21,11 +21,12 @@ impl ProtocolPing {
         Arc::new(Self {
             channel: channel.clone(),
             settings,
-            jobsman: ProtocolJobsManager::new(channel),
+            jobsman: ProtocolJobsManager::new("ProtocolPing", channel),
         })
     }
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) {
+        debug!(target: "net", "ProtocolPing::start() [START]");
         self.jobsman.clone().start(executor.clone());
         self.jobsman
             .clone()
@@ -35,9 +36,11 @@ impl ProtocolPing {
             .clone()
             .spawn(self.reply_to_ping(), executor)
             .await;
+        debug!(target: "net", "ProtocolPing::start() [END]");
     }
 
     async fn run_ping_pong(self: Arc<Self>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolPing::run_ping_pong() [START]");
         let pong_sub = self
             .channel
             .clone()
@@ -54,6 +57,7 @@ impl ProtocolPing {
             // Send ping message
             let ping = messages::Message::Ping(messages::PingMessage { nonce });
             self.channel.clone().send(ping).await?;
+            debug!(target: "net", "ProtocolPing::run_ping_pong() send Ping message");
 
             // Wait for pong, check nonce matches
             let pong_msg = receive_message!(pong_sub, messages::Message::Pong);
@@ -62,10 +66,12 @@ impl ProtocolPing {
                 self.channel.stop().await;
                 return Err(NetError::ChannelStopped);
             }
+            debug!(target: "net", "ProtocolPing::run_ping_pong() received Pong message");
         }
     }
 
     async fn reply_to_ping(self: Arc<Self>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolPing::reply_to_ping() [START]");
         let ping_sub = self
             .channel
             .clone()
@@ -75,10 +81,12 @@ impl ProtocolPing {
         loop {
             // Wait for ping, reply with pong that has a matching nonce
             let ping = receive_message!(ping_sub, messages::Message::Ping);
+            debug!(target: "net", "ProtocolPing::reply_to_ping() received Ping message");
 
             // Send ping message
             let pong = messages::Message::Pong(messages::PongMessage { nonce: ping.nonce });
             self.channel.clone().send(pong).await?;
+            debug!(target: "net", "ProtocolPing::reply_to_ping() sent Pong reply");
         }
     }
 

src/net/protocols/protocol_seed.rs

@@ -1,3 +1,4 @@
+use log::*;
 use smol::Executor;
 use std::sync::Arc;
 
@@ -21,6 +22,7 @@ impl ProtocolSeed {
     }
 
     pub async fn start(self: Arc<Self>, _executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolSeed::start() [START]");
         let addr_sub = self
             .channel
             .clone()
@@ -38,6 +40,7 @@ impl ProtocolSeed {
         let addrs_msg = receive_message!(addr_sub, messages::Message::Addrs);
         self.hosts.store(addrs_msg.addrs.clone()).await;
 
+        debug!(target: "net", "ProtocolSeed::start() [END]");
         Ok(())
     }
 

src/net/protocols/protocol_version.rs

@@ -1,4 +1,5 @@
 use futures::FutureExt;
+use log::*;
 use smol::Executor;
 use std::sync::Arc;
 
@@ -18,32 +19,49 @@ impl ProtocolVersion {
     }
 
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolVersion::run() [START]");
         // Start timer
         // Send version, wait for verack
         // Wait for version, send verack
         // Fin.
-        futures::select! {
+        let result = futures::select! {
             _ = self.clone().exchange_versions(executor).fuse() => Ok(()),
             _ = sleep(self.settings.channel_handshake_seconds).fuse() => Err(NetError::ChannelTimeout)
-        }
+        };
+        debug!(target: "net", "ProtocolVersion::run() [END]");
+        result
     }
 
     async fn exchange_versions(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolVersion::exchange_versions() [START]");
         let send = executor.spawn(self.clone().send_version());
         let recv = executor.spawn(self.recv_version());
 
-        send.await.and(recv.await)
+        send.await.and(recv.await)?;
+        debug!(target: "net", "ProtocolVersion::exchange_versions() [END]");
+        Ok(())
     }
 
     async fn send_version(self: Arc<Self>) -> NetResult<()> {
-        let version = messages::Message::Version(messages::VersionMessage {});
+        debug!(target: "net", "ProtocolVersion::send_version() [START]");
+        let verack_sub = self
+            .channel
+            .clone()
+            .subscribe_msg(messages::PacketType::Verack)
+            .await;
 
+        let version = messages::Message::Version(messages::VersionMessage {});
         self.channel.clone().send(version).await?;
 
+        // Wait for version acknowledgement
+        let _verack_msg = verack_sub.receive().await?;
+
+        debug!(target: "net", "ProtocolVersion::send_version() [END]");
         Ok(())
     }
 
     async fn recv_version(self: Arc<Self>) -> NetResult<()> {
+        debug!(target: "net", "ProtocolVersion::recv_version() [START]");
         let version_sub = self
             .channel
             .clone()
@@ -54,6 +72,11 @@ impl ProtocolVersion {
 
         // Check the message is OK
 
+        // Send version acknowledgement
+        let verack = messages::Message::Verack(messages::VerackMessage {});
+        self.channel.clone().send(verack).await?;
+
+        debug!(target: "net", "ProtocolVersion::recv_version() [END]");
         Ok(())
     }
 }

src/net/sessions/inbound_session.rs

@@ -57,6 +57,7 @@ impl InboundSession {
 
     pub async fn stop(&self) {
         self.acceptor.stop().await;
+        self.accept_task.stop().await;
     }
 
     fn start_accept_session(

src/net/sessions/mod.rs

@@ -1,7 +1,9 @@
 pub mod inbound_session;
+pub mod outbound_session;
 pub mod seed_session;
 pub mod session;
 
 pub use inbound_session::InboundSession;
+pub use outbound_session::OutboundSession;
 pub use seed_session::SeedSession;
 pub use session::Session;

src/net/sessions/outbound_session.rs

@@ -0,0 +1,129 @@
+use async_executor::Executor;
+use async_std::sync::Mutex;
+use log::*;
+use std::net::SocketAddr;
+use std::sync::{Arc, Weak};
+
+use crate::net::error::{NetError, NetResult};
+use crate::net::protocols::{ProtocolAddress, ProtocolPing};
+use crate::net::sessions::Session;
+use crate::net::{ChannelPtr, Connector, P2p};
+use crate::system::{StoppableTask, StoppableTaskPtr};
+
+pub struct OutboundSession {
+    p2p: Weak<P2p>,
+    connect_slots: Mutex<Vec<StoppableTaskPtr>>,
+}
+
+impl OutboundSession {
+    pub fn new(p2p: Weak<P2p>) -> Arc<Self> {
+        Arc::new(Self {
+            p2p,
+            connect_slots: Mutex::new(Vec::new()),
+        })
+    }
+
+    pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        let slots_count = self.p2p().settings().outbound_connections;
+        let mut connect_slots = self.connect_slots.lock().await;
+
+        for i in 0..slots_count {
+            let task = StoppableTask::new();
+
+            task.clone().start(
+                self.clone().channel_connect_loop(i, executor.clone()),
+                // Ignore stop handler
+                |_| async {},
+                NetError::ServiceStopped,
+                executor.clone(),
+            );
+
+            connect_slots.push(task);
+        }
+
+        Ok(())
+    }
+
+    pub async fn stop(&self) {
+        let connect_slots = &*self.connect_slots.lock().await;
+
+        for slot in connect_slots {
+            slot.stop().await;
+        }
+    }
+
+    pub async fn channel_connect_loop(
+        self: Arc<Self>,
+        slot_number: u32,
+        executor: Arc<Executor<'_>>,
+    ) -> NetResult<()> {
+        let connector = Connector::new(self.p2p().settings().clone());
+
+        loop {
+            let addr = self.load_address(slot_number).await?;
+            info!("Connecting to outbound [{}]", addr);
+
+            match connector.connect(addr).await {
+                Ok(channel) => {
+                    // Blacklist goes here
+
+                    info!("Connected outbound [{}]", addr);
+
+                    let stop_sub = channel.subscribe_stop().await;
+
+                    self.clone()
+                        .register_channel(channel.clone(), executor.clone())
+                        .await?;
+
+                    self.clone()
+                        .attach_protocols(channel, executor.clone())
+                        .await?;
+
+                    // Wait for channel to close
+                    stop_sub.receive().await;
+                }
+                Err(err) => {
+                    info!("Unable to connect to outbound [{}]: {}", addr, err);
+                }
+            }
+        }
+    }
+
+    async fn load_address(&self, slot_number: u32) -> NetResult<SocketAddr> {
+        let hosts = self.p2p().hosts();
+
+        match hosts.load_single().await {
+            Some(addr) => Ok(addr),
+            None => {
+                error!(
+                    "Hosts address pool is empty. Closing connect slot #{}",
+                    slot_number
+                );
+                Err(NetError::ServiceStopped)
+            }
+        }
+    }
+
+    async fn attach_protocols(
+        self: Arc<Self>,
+        channel: ChannelPtr,
+        executor: Arc<Executor<'_>>,
+    ) -> NetResult<()> {
+        let settings = self.p2p().settings().clone();
+        let hosts = self.p2p().hosts().clone();
+
+        let protocol_ping = ProtocolPing::new(channel.clone(), settings.clone());
+        protocol_ping.start(executor.clone()).await;
+
+        let protocol_addr = ProtocolAddress::new(channel, hosts, settings);
+        protocol_addr.start(executor).await;
+
+        Ok(())
+    }
+}
+
+impl Session for OutboundSession {
+    fn p2p(&self) -> Arc<P2p> {
+        self.p2p.upgrade().unwrap()
+    }
+}

src/net/sessions/seed_session.rs

@@ -18,6 +18,7 @@ impl SeedSession {
     }
 
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
+        debug!(target: "net", "SeedSession::start() [START]");
         let settings = {
             let p2p = self.p2p.upgrade().unwrap();
             p2p.settings()
@@ -38,26 +39,32 @@ impl SeedSession {
 
         let mut tasks = Vec::new();
 
-        for seed in settings.seeds.clone() {
-            tasks.push(executor.spawn(self.clone().start_seed(seed, executor.clone())));
+        for (i, seed) in settings.seeds.iter().enumerate() {
+            tasks.push(executor.spawn(self.clone().start_seed(i, seed.clone(), executor.clone())));
         }
 
-        for task in tasks {
+        for (i, task) in tasks.into_iter().enumerate() {
             // Ignore errors
-            let _ = task.await;
+            match task.await {
+                Ok(()) => info!("Successfully queried seed #{}", i),
+                Err(err) => warn!("Seed query #{} failed for reason: {}", i, err),
+            }
         }
 
         // Seed process complete
         // TODO: check increase count of address
 
+        debug!(target: "net", "SeedSession::start() [END]");
         Ok(())
     }
 
     async fn start_seed(
         self: Arc<Self>,
+        seed_index: usize,
         seed: SocketAddr,
         executor: Arc<Executor<'_>>,
     ) -> NetResult<()> {
+        debug!(target: "net", "SeedSession::start_seed(i={}) [START]", seed_index);
         let (hosts, settings) = {
             let p2p = self.p2p.upgrade().unwrap();
             (p2p.hosts(), p2p.settings())
@@ -68,17 +75,23 @@ impl SeedSession {
             Ok(channel) => {
                 // Blacklist goes here
 
-                info!("Connected seed [{}]", seed);
+                info!("Connected seed #{} [{}]", seed_index, seed);
 
                 self.clone()
                     .register_channel(channel.clone(), executor.clone())
                     .await?;
 
                 self.attach_protocols(channel, hosts, settings, executor)
-                    .await
+                    .await?;
+
+                debug!(target: "net", "SeedSession::start_seed(i={}) [END]", seed_index);
+                Ok(())
             }
             Err(err) => {
-                info!("Failure contacting seed [{}]: {}", seed, err);
+                info!(
+                    "Failure contacting seed #{} [{}]: {}",
+                    seed_index, seed, err
+                );
                 Err(err)
             }
         }
@@ -95,6 +108,7 @@ impl SeedSession {
         protocol_ping.start(executor.clone()).await;
 
         let protocol_seed = ProtocolSeed::new(channel.clone(), hosts, settings.clone());
+        // This will block until seed process is complete
         protocol_seed.start(executor.clone()).await?;
 
         channel.stop().await;

src/net/sessions/session.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+use log::*;
 use smol::Executor;
 use std::sync::Arc;
 
@@ -8,12 +9,18 @@ use crate::net::protocols::ProtocolVersion;
 use crate::net::ChannelPtr;
 
 async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
+    debug!(target: "net", "remove_sub_on_stop() [START]");
     // Subscribe to stop events
     let stop_sub = channel.clone().subscribe_stop().await;
     // Wait for a stop event
     let _ = stop_sub.receive().await;
+    debug!(target: "net",
+        "remove_sub_on_stop(): received stop event. Removing channel {}",
+        channel.address()
+    );
     // Remove channel from p2p
     p2p.remove(channel).await;
+    debug!(target: "net", "remove_sub_on_stop() [END]");
 }
 
 #[async_trait]
@@ -23,12 +30,16 @@ pub trait Session: Sync {
         channel: ChannelPtr,
         executor: Arc<Executor<'_>>,
     ) -> NetResult<()> {
+        debug!(target: "net", "Session::register_channel() [START]");
         let handshake_task = self.perform_handshake_protocols(channel.clone(), executor.clone());
 
         // start channel
         channel.start(executor);
 
-        handshake_task.await
+        handshake_task.await?;
+
+        debug!(target: "net", "Session::register_channel() [END]");
+        Ok(())
     }
 
     async fn perform_handshake_protocols(