net/acceptor: replace Mutex<Option<Weak<Session>>> with LazyWeak<Session>

src/net/acceptor.rs

@@ -19,16 +19,16 @@
 use std::sync::Arc;
 
 use log::error;
-use smol::{lock::Mutex, Executor};
+use smol::Executor;
 use url::Url;
 
 use super::{
     channel::{Channel, ChannelPtr},
-    session::SessionWeakPtr,
+    session::SessionWeakPtr2,
     transport::{Listener, PtListener},
 };
 use crate::{
-    system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
+    system::{LazyWeak, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
     Error, Result,
 };
 
@@ -39,7 +39,7 @@ pub type AcceptorPtr = Arc<Acceptor>;
 pub struct Acceptor {
     channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
     task: StoppableTaskPtr,
-    pub session: Mutex<Option<SessionWeakPtr>>,
+    pub(in crate::net) session: SessionWeakPtr2,
 }
 
 impl Acceptor {
@@ -48,7 +48,7 @@ impl Acceptor {
         Arc::new(Self {
             channel_subscriber: Subscriber::new(),
             task: StoppableTask::new(),
-            session: Mutex::new(None),
+            session: LazyWeak::new(),
         })
     }
 
@@ -86,7 +86,8 @@ impl Acceptor {
         loop {
             match listener.next().await {
                 Ok((stream, url)) => {
-                    let session = self.session.lock().await.clone().unwrap();
+                    let session = self.session.upgrade();
+                    let session = Arc::downgrade(&session);
                     let channel = Channel::new(stream, url, session).await;
                     self.channel_subscriber.notify(Ok(channel)).await;
                 }

src/net/session/inbound_session.rs

@@ -118,8 +118,7 @@ impl InboundSession {
         info!(target: "net::inbound_session", "[P2P] Starting Inbound session #{} on {}", index, accept_addr);
         // Generate a new acceptor for this inbound session
         let acceptor = Acceptor::new();
-        let parent = Arc::downgrade(&self);
-        *acceptor.session.lock().await = Some(parent);
+        acceptor.session.init(self.clone());
 
         // Start listener
         let result = acceptor.clone().start(accept_addr, ex).await;

src/net/session/mod.rs

@@ -23,7 +23,7 @@ use log::debug;
 use smol::Executor;
 
 use super::{channel::ChannelPtr, p2p::P2pPtr, protocol::ProtocolVersion};
-use crate::Result;
+use crate::{system::LazyWeak, Result};
 
 pub mod inbound_session;
 pub use inbound_session::{InboundSession, InboundSessionPtr};
@@ -43,6 +43,7 @@ pub const SESSION_SEED: SessionBitFlag = 0b1000;
 pub const SESSION_ALL: SessionBitFlag = 0b1111;
 
 pub type SessionWeakPtr = Weak<dyn Session + Send + Sync + 'static>;
+pub type SessionWeakPtr2 = LazyWeak<dyn Session + Send + Sync + 'static>;
 
 /// Removes channel from the list of connected channels when a stop signal
 /// is received.

src/system/lazy_weak.rs

@@ -71,9 +71,9 @@ use std::sync::{Arc, OnceLock, Weak};
 ///     }
 /// }
 /// ```
-pub struct LazyWeak<Parent>(OnceLock<Weak<Parent>>);
+pub struct LazyWeak<Parent: ?Sized>(OnceLock<Weak<Parent>>);
 
-impl<Parent> LazyWeak<Parent> {
+impl<Parent: ?Sized> LazyWeak<Parent> {
     /// Create an empty `LazyWeak`, which must immediately be followed by `weak.init()`.
     pub fn new() -> Self {
         Self(OnceLock::new())