Merge branch 'develop' of github.com:BlockProject3D/tools.net into develop

Author: Yuri6037

core/src/tcp/client.rs

@@ -29,8 +29,10 @@
 //! A basic TCP client implementation designed for long-running connections.
 
 use crate::tcp::util::buffer::{ChannelBuffer, NetReceiver};
+use crate::tcp::util::net::{Network, ReadyEvent};
 use crate::tcp::util::DataMsg;
 use crate::tcp::BYTES_CHANNEL_SIZE;
+use crate::util::barrier;
 use bp3d_debug::{error, trace};
 use std::future::Future;
 use std::sync::Arc;
@@ -39,8 +41,6 @@ use tokio::net::{TcpStream, ToSocketAddrs};
 use tokio::select;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
 use tokio::sync::{mpsc, watch};
-use crate::tcp::util::net::{Network, ReadyEvent};
-use crate::util::barrier;
 
 /// The reader trait which is supposed to handle the actual data reading loop.
 pub trait Reader {
@@ -113,7 +113,10 @@ pub trait Factory {
 }
 
 /// SAFETY: DataMsg must point to valid memory (normally ensured by Client structure).
-async unsafe fn handle_data(msg: barrier::mpsc::Lock<DataMsg>, net: &mut Network) -> std::io::Result<()> {
+async unsafe fn handle_data(
+    msg: barrier::mpsc::Lock<DataMsg>,
+    net: &mut Network,
+) -> std::io::Result<()> {
     trace!({?net} {?msg}, "Received data event");
     let slice = std::slice::from_raw_parts(msg.buffer, msg.buffer_size);
     net.write_all(slice).await?;
@@ -173,7 +176,7 @@ impl<F: Factory> Builder<F> {
             exit: exit_sender,
             request_sender,
             reply_sender,
-            data: data_sender
+            data: data_sender,
         });
         let mut handler = self.factory.start(&client);
         let handle = tokio::spawn(async move {
@@ -221,7 +224,7 @@ pub struct Client<H: Handler> {
     exit: watch::Sender<()>,
     request_sender: mpsc::Sender<H::Request>,
     data: barrier::mpsc::Sender<DataMsg>,
-    reply_sender: mpsc::Sender<H::Reply>
+    reply_sender: mpsc::Sender<H::Reply>,
 }
 
 impl<H: Handler> Client<H> {
@@ -286,11 +289,13 @@ impl<H: Handler> Client<H> {
     /// returns: true if the operation has succeeded, false otherwise.
     pub async fn send(&self, msg: &[u8]) -> Result<(), barrier::Error> {
         // SAFETY: It is safe to pass a pointer to msg thanks to the barrier synchronization.
-        self.data.send(DataMsg {
-            buffer: msg.as_ptr(),
-            buffer_size: msg.len(),
-            net_id: 0,
-        }).await
+        self.data
+            .send(DataMsg {
+                buffer: msg.as_ptr(),
+                buffer_size: msg.len(),
+                net_id: 0,
+            })
+            .await
     }
 }
 

core/src/tcp/server/client.rs

@@ -27,16 +27,16 @@
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use crate::tcp::util::buffer::{ChannelBuffer, NetReceiver};
+use crate::tcp::util::net::{Network, ReadyEvent};
 use crate::tcp::util::DataMsg;
 use crate::tcp::BYTES_CHANNEL_SIZE;
+use crate::util::barrier;
 use bp3d_debug::{error, trace};
 use std::future::Future;
 use tokio::io::AsyncWriteExt;
 use tokio::select;
 use tokio::sync::mpsc;
 use tokio::sync::watch;
-use crate::tcp::util::net::{Network, ReadyEvent};
-use crate::util::barrier;
 
 /// Represents a client event handler.
 pub trait Handler {
@@ -108,7 +108,10 @@ impl<H: Handler + Send + 'static> ClientTask<'_, H> {
 }
 
 /// SAFETY: DataMsg must point to valid memory (normally ensured by Server structure).
-async unsafe fn handle_broadcast(msg: barrier::broadcast::Lock<DataMsg>, net: &mut Network) -> std::io::Result<()> {
+async unsafe fn handle_broadcast(
+    msg: barrier::broadcast::Lock<DataMsg>,
+    net: &mut Network,
+) -> std::io::Result<()> {
     trace!({?net} {?msg}, "Received broadcast event");
     if msg.net_id == 0 || msg.net_id == net.id() {
         let slice = std::slice::from_raw_parts(msg.buffer, msg.buffer_size);

core/src/tcp/server/server.rs

@@ -27,12 +27,14 @@
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use crate::tcp::server::client::ClientTask;
+use crate::tcp::util::net::Network;
 use crate::tcp::util::DataMsg;
+use crate::util::barrier;
 use bp3d_debug::{debug, trace};
 use std::future::Future;
 use std::net::Ipv4Addr;
-use std::sync::atomic::Ordering::Relaxed;
 use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering::Relaxed;
 use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, ToSocketAddrs};
@@ -41,8 +43,6 @@ use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
 use tokio::sync::watch;
 use tokio::task::JoinSet;
-use crate::tcp::util::net::Network;
-use crate::util::barrier;
 
 /// A factory trait which can be used to create the instance of the main server event handler.
 pub trait Factory {
@@ -268,7 +268,7 @@ impl<F: Factory> Builder<F> {
             exit: exit_sender,
             max_clients: self.max_clients,
             cur_clients: AtomicUsize::new(0),
-            reply_sender
+            reply_sender,
         });
         let handler = self.factory.start(&server);
         let motherfuckingrust = server.clone();
@@ -297,7 +297,7 @@ pub struct Server<H: Handler> {
     cur_clients: AtomicUsize,
     max_clients: usize,
     request_sender: mpsc::Sender<H::Request>,
-    reply_sender: mpsc::Sender<H::Reply>
+    reply_sender: mpsc::Sender<H::Reply>,
 }
 
 impl<H: Handler> Server<H> {
@@ -373,11 +373,13 @@ impl<H: Handler> Server<H> {
     /// returns: true if the operation has succeeded, false otherwise.
     pub async fn send(&self, net_id: usize, msg: &[u8]) -> Result<(), barrier::Error> {
         // SAFETY: It is safe to pass a pointer to msg thanks to the barrier synchronization.
-        self.broadcast.send(DataMsg {
-            buffer: msg.as_ptr(),
-            buffer_size: msg.len(),
-            net_id
-        }).await?;
+        self.broadcast
+            .send(DataMsg {
+                buffer: msg.as_ptr(),
+                buffer_size: msg.len(),
+                net_id,
+            })
+            .await?;
         trace!("All clients have acknowledged");
         Ok(())
     }
@@ -391,11 +393,13 @@ impl<H: Handler> Server<H> {
     /// returns: true if the operation has succeeded, false otherwise.
     pub async fn broadcast(&self, msg: &[u8]) -> Result<(), barrier::Error> {
         // SAFETY: It is safe to pass a pointer to msg thanks to the barrier synchronization.
-        self.broadcast.send(DataMsg {
-            buffer: msg.as_ptr(),
-            buffer_size: msg.len(),
-            net_id: 0
-        }).await?;
+        self.broadcast
+            .send(DataMsg {
+                buffer: msg.as_ptr(),
+                buffer_size: msg.len(),
+                net_id: 0,
+            })
+            .await?;
         trace!("All clients have acknowledged");
         Ok(())
     }

core/src/tcp/util/mod.rs

@@ -28,8 +28,8 @@
 
 //! Utility module for TCP client or server.
 
-pub mod net;
 pub mod buffer;
+pub mod net;
 
 use std::fmt::Debug;
 

core/src/tcp/util/net.rs

@@ -28,17 +28,17 @@
 
 //! TCP network stream async reader/writer tools.
 
+use crate::tcp::util::buffer::Bytes;
+use bp3d_debug::warning;
 use std::fmt::{Debug, Formatter};
 use std::io::{Error, ErrorKind, IoSlice};
 use std::net::SocketAddr;
 use std::pin::Pin;
 use std::task::{Context, Poll};
-use bp3d_debug::warning;
 use tokio::io::{AsyncRead, AsyncWrite, Interest, ReadBuf};
 use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
 use tokio::net::TcpStream;
 use tokio::sync::mpsc;
-use crate::tcp::util::buffer::Bytes;
 
 /// The event returned by the ready function in [Network].
 pub enum ReadyEvent {

core/src/util/barrier/broadcast.rs

@@ -28,28 +28,28 @@
 
 //! A multi-client barrier synchronization primitive.
 
+use crate::util::barrier::{Error, Msg};
+use bp3d_debug::trace;
 use std::fmt::Debug;
 use std::ops::{Deref, DerefMut};
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::Relaxed;
-use bp3d_debug::trace;
 use tokio::sync::{broadcast, Semaphore};
-use crate::util::barrier::{Error, Msg};
 
 /// Represents a multi-client barrier lock.
 pub struct Lock<T> {
-    msg: Msg<T>
+    msg: Msg<T>,
 }
 
 /// Represents a multi-client sender.
 pub struct Sender<T> {
     inner: broadcast::Sender<Msg<T>>,
-    closed: AtomicBool
+    closed: AtomicBool,
 }
 
 /// Represents a multi-client receiver.
 pub struct Receiver<T> {
-    inner: broadcast::Receiver<Msg<T>>
+    inner: broadcast::Receiver<Msg<T>>,
 }
 
 impl<T> Deref for Lock<T> {
@@ -98,7 +98,7 @@ impl<T> Sender<T> {
         let semaphore = Semaphore::new(0);
         let msg = Msg {
             synchro: &semaphore,
-            inner: msg
+            inner: msg,
         };
         let count = self.inner.send(msg).map_err(|_| Error::BrokenPipe)?;
         trace!("Waiting for {} client(s) to acknowledge", count);
@@ -108,7 +108,9 @@ impl<T> Sender<T> {
 
     /// Creates and subscribes a new receiver to this sender.
     pub fn subscribe(&self) -> Receiver<T> {
-        Receiver { inner: self.inner.subscribe() }
+        Receiver {
+            inner: self.inner.subscribe(),
+        }
     }
 }
 
@@ -119,7 +121,11 @@ impl<T: Clone> Receiver<T> {
     ///
     /// This returns an [Error] if the channel was prematurely closed.
     pub async fn recv(&mut self) -> Result<Lock<T>, Error> {
-        self.inner.recv().await.map_err(|_| Error::BrokenPipe).map(|msg| Lock { msg })
+        self.inner
+            .recv()
+            .await
+            .map_err(|_| Error::BrokenPipe)
+            .map(|msg| Lock { msg })
     }
 }
 
@@ -132,5 +138,8 @@ impl<T: Clone> Receiver<T> {
 /// returns: Sender<T>
 pub fn barrier<T: Clone>(len: usize) -> Sender<T> {
     let (sender, _) = broadcast::channel(len);
-    Sender { inner: sender, closed: AtomicBool::new(false) }
+    Sender {
+        inner: sender,
+        closed: AtomicBool::new(false),
+    }
 }

core/src/util/barrier/mod.rs

@@ -41,14 +41,14 @@ pub enum Error {
     BrokenPipe,
 
     /// The channel was requested to be closed.
-    Closed
+    Closed,
 }
 
 impl Display for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Error::BrokenPipe => f.write_str("broken pipe"),
-            Error::Closed => f.write_str("closed")
+            Error::Closed => f.write_str("closed"),
         }
     }
 }
@@ -58,7 +58,7 @@ impl std::error::Error for Error {}
 #[derive(Clone)]
 struct Msg<T> {
     synchro: *const Semaphore,
-    inner: T
+    inner: T,
 }
 
 unsafe impl<T: Send> Send for Msg<T> {}

core/src/util/barrier/mpsc.rs

@@ -28,28 +28,28 @@
 
 //! A single-client barrier synchronization primitive.
 
+use crate::util::barrier::{Error, Msg};
+use bp3d_debug::trace;
 use std::fmt::Debug;
 use std::ops::{Deref, DerefMut};
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::Relaxed;
-use bp3d_debug::trace;
 use tokio::sync::{mpsc, Semaphore};
-use crate::util::barrier::{Error, Msg};
 
 /// Represents a single-client barrier lock.
 pub struct Lock<T> {
-    msg: Msg<T>
+    msg: Msg<T>,
 }
 
 /// Represents a single-client sender.
 pub struct Sender<T> {
     inner: mpsc::Sender<Msg<T>>,
-    closed: AtomicBool
+    closed: AtomicBool,
 }
 
 /// Represents a single-client receiver.
 pub struct Receiver<T> {
-    inner: mpsc::Receiver<Msg<T>>
+    inner: mpsc::Receiver<Msg<T>>,
 }
 
 impl<T> Deref for Lock<T> {
@@ -98,7 +98,7 @@ impl<T> Sender<T> {
         let semaphore = Semaphore::new(0);
         let msg = Msg {
             synchro: &semaphore,
-            inner: msg
+            inner: msg,
         };
         self.inner.send(msg).await.map_err(|_| Error::BrokenPipe)?;
         trace!("Waiting for client to acknowledge");
@@ -114,7 +114,11 @@ impl<T: Clone> Receiver<T> {
     ///
     /// This returns an [Error] if the channel was prematurely closed.
     pub async fn recv(&mut self) -> Result<Lock<T>, Error> {
-        self.inner.recv().await.ok_or(Error::BrokenPipe).map(|msg| Lock { msg })
+        self.inner
+            .recv()
+            .await
+            .ok_or(Error::BrokenPipe)
+            .map(|msg| Lock { msg })
     }
 }
 
@@ -127,5 +131,11 @@ impl<T: Clone> Receiver<T> {
 /// returns: (Sender<T>, Receiver<T>)
 pub fn barrier<T: Clone>(len: usize) -> (Sender<T>, Receiver<T>) {
     let (sender, receiver) = mpsc::channel(len);
-    (Sender { inner: sender, closed: AtomicBool::new(false) }, Receiver { inner: receiver })
+    (
+        Sender {
+            inner: sender,
+            closed: AtomicBool::new(false),
+        },
+        Receiver { inner: receiver },
+    )
 }

testprog/src/client.rs

@@ -27,10 +27,10 @@
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use bp3d_net::tcp::client::{Client, Factory, Handler, Reader};
-use std::sync::Arc;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use bp3d_net::tcp::util::buffer::NetReceiver;
 use bp3d_net::tcp::util::net::Network;
+use std::sync::Arc;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 
 #[derive(Clone)]
 pub struct EchoClient {