feat(transport): start working on transport metering & metrics

Author: mempirate

.gitignore

@@ -4,3 +4,5 @@
 *.svg
 *.log
 .DS_Store
+
+todo.md

Cargo.lock

@@ -55,6 +55,7 @@ thiserror = "1"
 tracing = "0.1"
 rustc-hash = "1"
 rand = "0.8"
+libc = "0.2"
 
 # networking
 quinn = "0.11.9"

Cargo.toml

@@ -34,7 +34,7 @@ type ConnectionTask<Io, Err> = Pin<Box<dyn Future<Output = Result<Io, Err>> + Se
 type ConnectionCtl<Io, Addr> = ConnectionState<Framed<Io, reqrep::Codec>, ExponentialBackoff, Addr>;
 
 /// The request socket driver. Endless future that drives
-/// the the socket forward.
+/// the socket forward.
 pub(crate) struct ReqDriver<T: Transport<A>, A: Address> {
     /// Options shared with the socket.
     pub(crate) options: Arc<ReqOptions>,

msg-socket/src/req/driver.rs

@@ -57,7 +57,8 @@ pub struct ReqOptions {
     /// Default is `None`, and the connection is flushed after every send.
     flush_interval: Option<std::time::Duration>,
     /// The maximum number of bytes that can be buffered in the session before being flushed.
-    /// This internally sets [`Framed::set_backpressure_boundary`](tokio_util::codec::Framed).
+    /// This internally sets
+    /// [`Framed::set_backpressure_boundary`](tokio_util::codec::Framed::set_backpressure_boundary).
     backpressure_boundary: usize,
     /// The maximum number of retry attempts. If `None`, the connection will retry indefinitely.
     retry_attempts: Option<usize>,
@@ -182,5 +183,6 @@ impl ReqMessage {
 /// The request socket state, shared between the backend task and the socket.
 #[derive(Debug, Default)]
 pub(crate) struct SocketState {
+    /// The socket stats.
     pub(crate) stats: SocketStats<ReqStats>,
 }

msg-socket/src/req/mod.rs

@@ -13,6 +13,7 @@ use super::{Command, DEFAULT_BUFFER_SIZE, ReqError, ReqOptions};
 use crate::{
     ConnectionState, ExponentialBackoff, ReqMessage,
     req::{SocketState, driver::ReqDriver, stats::ReqStats},
+    stats::SocketStats,
 };
 
 /// The request socket.
@@ -82,8 +83,9 @@ where
         self
     }
 
-    pub fn stats(&self) -> &ReqStats {
-        &self.state.stats.specific
+    /// Returns the socket stats.
+    pub fn stats(&self) -> &SocketStats<ReqStats> {
+        &self.state.stats
     }
 
     pub async fn request(&self, message: Bytes) -> Result<Bytes, ReqError> {

msg-socket/src/req/socket.rs

@@ -1,7 +1,9 @@
 use std::fmt::Debug;
 
+/// Statistics for a socket
 #[derive(Debug)]
 pub struct SocketStats<S> {
+    /// Socket-specific stats.
     pub(crate) specific: S,
 }
 

msg-socket/src/stats.rs

@@ -158,7 +158,7 @@ impl<A: Address> PubMessage<A> {
 }
 
 /// The subscriber socket state, shared between the backend task and the socket frontend.
-#[derive(Debug)] // Should derive default fine now
+#[derive(Debug)]
 pub(crate) struct SocketState<A: Address> {
     pub(crate) stats: SocketStats<SubStats<A>>,
 }
@@ -184,7 +184,7 @@ mod tests {
     use super::*;
 
     async fn spawn_listener() -> SocketAddr {
-        let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
+        let listener = TcpListener::bind("[::]:0").await.unwrap();
 
         let addr = listener.local_addr().unwrap();
 

msg-socket/src/sub/mod.rs

@@ -19,6 +19,8 @@ tokio.workspace = true
 tracing.workspace = true
 thiserror.workspace = true
 
+libc.workspace = true
+
 # QUIC
 quinn = { workspace = true, optional = true }
 rcgen = { workspace = true, optional = true }

msg-transport/Cargo.toml

@@ -1,5 +1,6 @@
 use std::{
     io,
+    os::fd::{AsRawFd, RawFd},
     path::PathBuf,
     pin::Pin,
     task::{Context, Poll},
@@ -51,6 +52,12 @@ pub struct IpcStream {
     stream: UnixStream,
 }
 
+impl AsRawFd for IpcStream {
+    fn as_raw_fd(&self) -> RawFd {
+        self.stream.as_raw_fd()
+    }
+}
+
 impl IpcStream {
     pub async fn connect(peer: PathBuf) -> io::Result<Self> {
         let stream = UnixStream::connect(&peer).await?;