Merge remote-tracking branch 'origin/develop' into develop

Yuri6037 · Yuri6037 · commit 2c90c5d8e785 · 2025-01-15T22:47:28.000+01:00
diff --git a/core/src/tcp/client.rs b/core/src/tcp/client.rs
@@ -28,19 +28,19 @@
 
 //! A basic TCP client implementation designed for long-running connections.
 
-use tokio::sync::{mpsc, watch, Semaphore};
+use crate::tcp::buffer::ChannelBuffer;
+use crate::tcp::util::{DataMsg, Network, ReadyEvent};
+use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
+use bp3d_debug::{error, trace};
 use std::future::Future;
-use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::Relaxed;
-use bp3d_debug::{error, trace};
+use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpStream, ToSocketAddrs};
 use tokio::select;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
-use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
-use crate::tcp::buffer::ChannelBuffer;
-use crate::tcp::util::{DataMsg, Network, ReadyEvent};
+use tokio::sync::{mpsc, watch, Semaphore};
 
 /// The reader trait which is supposed to handle the actual data reading loop.
 pub trait Reader {
@@ -86,7 +86,10 @@ pub trait Handler {
     /// * `net`: the network context created for this client.
     ///
     /// returns: impl Future<Output=Result<Self::Reader, Error>>+Send+Sized
-    fn connect(&mut self, net: &mut Network) -> impl Future<Output = std::io::Result<Self::Reader>> + Send;
+    fn connect(
+        &mut self,
+        net: &mut Network,
+    ) -> impl Future<Output = std::io::Result<Self::Reader>> + Send;
 
     /// Called when the client task is about to return.
     ///
@@ -96,9 +99,7 @@ pub trait Handler {
     ///
     /// returns: impl Future<Output=Result<(), Error>>+Send+Sized
     fn disconnect(&mut self, _: &mut Network) -> impl Future<Output = std::io::Result<()>> + Send {
-        async move {
-            Ok(())
-        }
+        async move { Ok(()) }
     }
 }
 
@@ -124,15 +125,15 @@ async unsafe fn handle_data(msg: DataMsg, net: &mut Network) -> std::io::Result<
 /// The main builder structure used to create a new TCP client.
 pub struct Builder<F> {
     factory: F,
-    event_queue_size: usize
+    event_queue_size: usize,
 }
 
 impl<F: Factory> Builder<F> {
     /// Creates a new TCP client from the given main event handler factory.
     pub fn new(factory: F) -> Builder<F> {
         Self {
             factory,
-            event_queue_size: 4
+            event_queue_size: 4,
         }
     }
 
@@ -212,7 +213,7 @@ impl<F: Factory> Builder<F> {
         Ok(ClientApp {
             client,
             handle,
-            reply_receiver
+            reply_receiver,
         })
     }
 }
@@ -223,7 +224,7 @@ pub struct Client<H: Handler> {
     request_sender: mpsc::Sender<H::Request>,
     data: mpsc::Sender<DataMsg>,
     reply_sender: mpsc::Sender<H::Reply>,
-    is_exiting: AtomicBool
+    is_exiting: AtomicBool,
 }
 
 impl<H: Handler> Client<H> {
@@ -293,12 +294,17 @@ impl<H: Handler> Client<H> {
         // SAFETY: It is safe to pass a pointer to msg as long as we wait for all clients to have
         // consumed the pointer before returning.
         let synchro = Semaphore::new(0);
-        if (self.data.send(DataMsg {
-            synchro: &synchro,
-            buffer: msg.as_ptr(),
-            buffer_size: msg.len(),
-            net_id: 0
-        }).await).is_err() {
+        if (self
+            .data
+            .send(DataMsg {
+                synchro: &synchro,
+                buffer: msg.as_ptr(),
+                buffer_size: msg.len(),
+                net_id: 0,
+            })
+            .await)
+            .is_err()
+        {
             return Err(crate::tcp::util::SendError::Closed);
         }
         trace!("Waiting for the async task to acknowledge");
diff --git a/core/src/tcp/server/client.rs b/core/src/tcp/server/client.rs
@@ -26,16 +26,16 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::future::Future;
+use crate::tcp::buffer::ChannelBuffer;
+use crate::tcp::util::{DataMsg, Network, ReadyEvent};
+use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
 use bp3d_debug::{error, trace};
+use std::future::Future;
 use tokio::io::AsyncWriteExt;
 use tokio::select;
+use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::watch;
-use tokio::sync::broadcast;
-use crate::tcp::buffer::ChannelBuffer;
-use crate::tcp::{NetReceiver, BYTES_CHANNEL_SIZE};
-use crate::tcp::util::{DataMsg, Network, ReadyEvent};
 
 /// Represents a client event handler.
 pub trait Handler {
@@ -56,17 +56,15 @@ pub trait Handler {
     ///
     /// returns: impl Future<Output=Result<(), Error>>+Send+Sized
     fn disconnect(&mut self, _: &mut Network) -> impl Future<Output = std::io::Result<()>> + Send {
-        async move {
-            Ok(())
-        }
+        async move { Ok(()) }
     }
 }
 
 pub(crate) struct ClientTask<'a, H> {
     pub(crate) net: &'a mut Network,
     pub(crate) handler: H,
     pub(crate) exit: watch::Receiver<()>,
-    pub(crate) broadcast: broadcast::Receiver<DataMsg>
+    pub(crate) broadcast: broadcast::Receiver<DataMsg>,
 }
 
 impl<H: Handler + Send + 'static> ClientTask<'_, H> {
diff --git a/core/src/tcp/server/server.rs b/core/src/tcp/server/server.rs
@@ -26,22 +26,22 @@
 // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+use crate::tcp::server::client::ClientTask;
+use crate::tcp::util::{DataMsg, Network};
+use bp3d_debug::{debug, trace};
 use std::future::Future;
 use std::net::Ipv4Addr;
-use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, AtomicUsize};
 use std::sync::atomic::Ordering::Relaxed;
-use bp3d_debug::{debug, trace};
+use std::sync::atomic::{AtomicBool, AtomicUsize};
+use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
 use tokio::net::{TcpListener, ToSocketAddrs};
 use tokio::select;
-use tokio::sync::{watch, Semaphore};
 use tokio::sync::broadcast;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::{SendError, TrySendError};
+use tokio::sync::{watch, Semaphore};
 use tokio::task::JoinSet;
-use crate::tcp::server::client::ClientTask;
-use crate::tcp::util::{DataMsg, Network};
 
 /// A factory trait which can be used to create the instance of the main server event handler.
 pub trait Factory {
@@ -83,7 +83,10 @@ pub trait Handler {
     /// * `net`: the network context created for this client.
     ///
     /// returns: impl Future<Output=Result<Self::ClientHandler, Error>>+Send+Sized
-    fn connect(&mut self, net: &mut Network) -> impl Future<Output = std::io::Result<Self::ClientHandler>> + Send;
+    fn connect(
+        &mut self,
+        net: &mut Network,
+    ) -> impl Future<Output = std::io::Result<Self::ClientHandler>> + Send;
 
     /// Called when a client has disconnected from the server.
     ///
@@ -96,7 +99,11 @@ pub trait Handler {
     /// * `handler`: the client event handler which was associated with the client.
     ///
     /// returns: impl Future<Output=()>+Send+Sized
-    fn disconnect(&mut self, _: &mut Network, _: Self::ClientHandler) -> impl Future<Output = ()> + Send {
+    fn disconnect(
+        &mut self,
+        _: &mut Network,
+        _: Self::ClientHandler,
+    ) -> impl Future<Output = ()> + Send {
         async move {}
     }
 }
@@ -106,7 +113,7 @@ struct ServerTask<H: Handler> {
     listener: TcpListener,
     exit_receiver: watch::Receiver<()>,
     request_receiver: mpsc::Receiver<H::Request>,
-    server: Arc<Server<H>>
+    server: Arc<Server<H>>,
 }
 
 impl<H: Handler + Send + 'static> ServerTask<H> {
@@ -271,14 +278,14 @@ impl<F: Factory> Builder<F> {
                 listener,
                 exit_receiver,
                 request_receiver,
-                server
+                server,
             };
             task.run().await
         });
         Ok(ServerApp {
             handle,
             server: motherfuckingrust,
-            reply_receiver
+            reply_receiver,
         })
     }
 }
@@ -291,7 +298,7 @@ pub struct Server<H: Handler> {
     max_clients: usize,
     request_sender: mpsc::Sender<H::Request>,
     reply_sender: mpsc::Sender<H::Reply>,
-    is_exiting: AtomicBool
+    is_exiting: AtomicBool,
 }
 
 impl<H: Handler> Server<H> {
@@ -376,10 +383,10 @@ impl<H: Handler> Server<H> {
             synchro: &synchro,
             buffer: msg.as_ptr(),
             buffer_size: msg.len(),
-            net_id
+            net_id,
         }) {
             Err(_) => return Err(crate::tcp::util::SendError::Closed),
-            Ok(v) => debug!("Broadcasting to {} client(s)", v)
+            Ok(v) => debug!("Broadcasting to {} client(s)", v),
         }
         let clients = self.cur_clients.load(Relaxed);
         trace!("Waiting for {} client(s) to acknowledge", clients);
@@ -406,10 +413,10 @@ impl<H: Handler> Server<H> {
             synchro: &synchro,
             buffer: msg.as_ptr(),
             buffer_size: msg.len(),
-            net_id: 0
+            net_id: 0,
         }) {
             Err(_) => return Err(crate::tcp::util::SendError::Closed),
-            Ok(v) => debug!("Broadcasting to {} client(s)", v)
+            Ok(v) => debug!("Broadcasting to {} client(s)", v),
         }
         let clients = self.cur_clients.load(Relaxed);
         trace!("Waiting for {} client(s) to acknowledge", clients);
diff --git a/core/src/tcp/util.rs b/core/src/tcp/util.rs
@@ -28,27 +28,27 @@
 
 //! Utility module for TCP client or server.
 
+use bp3d_debug::warning;
 use std::fmt::{Debug, Display, 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 tokio::sync::Semaphore;
 use crate::tcp::buffer::Bytes;
+use tokio::sync::Semaphore;
 
 #[derive(Clone, Debug)]
 pub(super) struct DataMsg {
     pub(super) synchro: *const Semaphore,
     pub(super) buffer: *const u8,
     pub(super) buffer_size: usize,
     #[allow(dead_code)]
-    pub(super) net_id: usize
+    pub(super) net_id: usize,
 }
 
 unsafe impl Send for DataMsg {}
@@ -60,14 +60,14 @@ pub enum SendError {
     IsExiting,
 
     /// The broadcast channel is already closed.
-    Closed
+    Closed,
 }
 
 impl Display for SendError {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         match self {
             SendError::IsExiting => f.write_str("exit requested"),
-            SendError::Closed => f.write_str("channel closed")
+            SendError::Closed => f.write_str("channel closed"),
         }
     }
 }
@@ -83,7 +83,7 @@ pub enum ReadyEvent {
     None,
 
     /// Data was submitted to the given ChannelBuffer channel.
-    Submitted
+    Submitted,
 }
 
 /// Buffered reader/writer for a TCP stream.
@@ -93,7 +93,7 @@ pub struct Network {
     reader: OwnedReadHalf,
     writer: OwnedWriteHalf,
     addr: SocketAddr,
-    id: usize
+    id: usize,
 }
 
 impl Debug for Network {
@@ -141,8 +141,14 @@ impl Network {
     /// # Errors
     ///
     /// Returns an IO error if the operation failed.
-    pub async fn ready_read<const N: usize>(&self, bytes_sender: &mpsc::Sender<Bytes<N>>) -> std::io::Result<ReadyEvent> {
-        let ev = self.reader.ready(Interest::ERROR | Interest::READABLE).await?;
+    pub async fn ready_read<const N: usize>(
+        &self,
+        bytes_sender: &mpsc::Sender<Bytes<N>>,
+    ) -> std::io::Result<ReadyEvent> {
+        let ev = self
+            .reader
+            .ready(Interest::ERROR | Interest::READABLE)
+            .await?;
         if ev.is_write_closed() || ev.is_read_closed() || ev.is_error() {
             return Ok(ReadyEvent::ConnectionLoss);
         }
@@ -170,14 +176,25 @@ impl Network {
 }
 
 impl AsyncRead for Network {
-    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+    fn poll_read(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<std::io::Result<()>> {
         unsafe { self.map_unchecked_mut(|v| &mut v.reader).poll_read(cx, buf) }
     }
 }
 
 impl AsyncWrite for Network {
-    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<Result<usize, Error>> {
-        unsafe { self.map_unchecked_mut(|v| &mut v.writer).poll_write(cx, buf) }
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        unsafe {
+            self.map_unchecked_mut(|v| &mut v.writer)
+                .poll_write(cx, buf)
+        }
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
@@ -188,8 +205,15 @@ impl AsyncWrite for Network {
         unsafe { self.map_unchecked_mut(|v| &mut v.writer).poll_shutdown(cx) }
     }
 
-    fn poll_write_vectored(self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>]) -> Poll<Result<usize, Error>> {
-        unsafe { self.map_unchecked_mut(|v| &mut v.writer).poll_write_vectored(cx, bufs) }
+    fn poll_write_vectored(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        bufs: &[IoSlice<'_>],
+    ) -> Poll<Result<usize, Error>> {
+        unsafe {
+            self.map_unchecked_mut(|v| &mut v.writer)
+                .poll_write_vectored(cx, bufs)
+        }
     }
 
     fn is_write_vectored(&self) -> bool {
diff --git a/testprog/src/client.rs b/testprog/src/client.rs
diff --git a/testprog/src/server.rs b/testprog/src/server.rs
