feat: introduce tokio compact StreamExt and remove AsyncOpShm (#4)

Given the strong coupling between `AsyncReadShm` & `AsyncWriteShm` and
`Stream`, and the need to explicitly call related `Stream` interfaces
(such as `close` and `reuse`) after using the interfaces of AsyncOpShm,
abstracting those traits is unnecessary.

However, for compatibility with `AsyncRead` and `AsyncWrite` of TokIO,
we added `StreamExt` to provide implementations for `AsyncRead` and
`AsyncWrite`. But since explicit calls to `Stream`'s interfaces are
still required in this case, we added detailed documentation to
`StreamExt` to remind users of this.

So we removed `AsyncReadShm` and `AsyncWriteShm` and Introduced
`StreamExt` which implements `AsyncRead` and `AsyncWrite` of TokIO.

Signed-off-by: Yu Li <[email protected]>

Author: yukiiiteru

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "shmipc"
-version = "0.0.1"
+version = "0.1.0"
 edition = "2024"
 authors = ["Volo Team <[email protected]>"]
 license = "Apache-2.0"
@@ -35,20 +35,13 @@ rand = "0.9"
 tokio-scoped = "0.2"
 tracing-subscriber = "0.3"
 
-
-[profile.dev]
-overflow-checks = false
-
 [profile.release]
 opt-level = 3
 debug = true
 rpath = false
 lto = true
-debug-assertions = false
 codegen-units = 1
 panic = 'unwind'
-incremental = false
-overflow-checks = false
 
 [workspace]
 members = ["examples"]

Cargo.toml

@@ -16,10 +16,12 @@ use std::os::unix::net::SocketAddr;
 
 use criterion::{Criterion, criterion_group, criterion_main};
 use shmipc::{
-    AsyncReadShm, AsyncWriteShm, BufferReader, BufferSlice, Error, LinkedBuffer, Listener,
-    SessionManager, SessionManagerConfig, Stream,
+    Error, Listener,
+    buffer::{BufferReader, BufferSlice, LinkedBuffer},
     config::SizePercentPair,
     consts::MemMapType,
+    session::{SessionManager, SessionManagerConfig},
+    stream::Stream,
     transport::{DefaultUnixConnect, DefaultUnixListen},
 };
 use tokio::{
@@ -93,7 +95,7 @@ fn criterion_benchmark(c: &mut Criterion) {
                         loop {
                             tokio::select! {
                                 r = server.accept() => {
-                                    let mut stream = r.unwrap().unwrap();
+                                    let mut stream = r.unwrap();
                                     tokio::spawn(async move {
                                         loop {
                                             if !must_read(&mut stream, size).await {

benches/bench.rs

@@ -16,6 +16,14 @@ tracing-subscriber = "0.3"
 name = "greeter_client"
 path = "src/hello_world/greeter_client.rs"
 
+[[bin]]
+name = "greeter_client_tokio"
+path = "src/hello_world/greeter_client_tokio.rs"
+
 [[bin]]
 name = "greeter_server"
 path = "src/hello_world/greeter_server.rs"
+
+[[bin]]
+name = "greeter_server_tokio"
+path = "src/hello_world/greeter_server_tokio.rs"

examples/Cargo.toml

@@ -15,7 +15,7 @@
 use std::os::unix::net::SocketAddr;
 
 use shmipc::{
-    AsyncReadShm, AsyncWriteShm, SessionManager, SessionManagerConfig,
+    session::{SessionManager, SessionManagerConfig},
     transport::DefaultUnixConnect,
 };
 

examples/src/hello_world/greeter_client.rs

@@ -0,0 +1,65 @@
+// Copyright 2025 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::os::unix::net::SocketAddr;
+
+use shmipc::{
+    compact::StreamExt,
+    session::{SessionManager, SessionManagerConfig},
+    transport::DefaultUnixConnect,
+};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt::init();
+    let dir = std::env::current_dir().unwrap();
+    let uds_path = SocketAddr::from_pathname(dir.join("../ipc_test.sock")).unwrap();
+
+    let mut conf = SessionManagerConfig::new();
+    conf.config_mut().mem_map_type = shmipc::consts::MemMapType::MemMapTypeMemFd;
+    conf.config_mut().share_memory_path_prefix = "/dev/shm/client.ipc.shm".to_string();
+    #[cfg(target_os = "macos")]
+    {
+        conf.config.share_memory_path_prefix = "/tmp/client.ipc.shm".to_string();
+        conf.config.queue_path = "/tmp/client.ipc.shm_queue".to_string();
+    }
+
+    let sm = SessionManager::new(conf, DefaultUnixConnect, uds_path)
+        .await
+        .unwrap();
+    let stream = sm.get_stream().unwrap();
+    let request_msg = "client say hello world!!!";
+
+    let mut conn = StreamExt::new(stream);
+
+    println!(
+        "size: {}",
+        conn.write(request_msg.as_bytes()).await.unwrap(),
+    );
+    conn.flush().await.unwrap();
+
+    const EXPECTED_RESP: &str = "server hello world!!!";
+    let mut buf = vec![0; 4096];
+
+    conn.read_exact(&mut buf[..EXPECTED_RESP.len()])
+        .await
+        .unwrap();
+    println!(
+        "client stream receive response {}",
+        str::from_utf8(&buf[..EXPECTED_RESP.len()]).unwrap()
+    );
+    conn.inner().reuse().await;
+    sm.close().await;
+}

examples/src/hello_world/greeter_client_tokio.rs

@@ -15,7 +15,7 @@
 use std::os::unix::net::SocketAddr;
 
 use nix::unistd::unlink;
-use shmipc::{AsyncReadShm, AsyncWriteShm, Listener, config::Config, transport::DefaultUnixListen};
+use shmipc::{Listener, config::Config, stream::Stream, transport::DefaultUnixListen};
 
 #[tokio::main]
 async fn main() {
@@ -32,19 +32,37 @@ async fn main() {
     )
     .await
     .unwrap();
-    let mut stream = ln.accept().await.unwrap().unwrap();
-    let req_msg = stream
-        .read_bytes("client say hello world!!!".len())
-        .await
-        .unwrap();
-    println!(
-        "server receive request {}",
-        String::from_utf8(req_msg.to_vec()).unwrap()
-    );
 
-    let resp_msg = "server hello world!!!";
-    stream.write_bytes(resp_msg.as_bytes()).unwrap();
-    stream.flush(true).await.unwrap();
+    loop {
+        let stream = match ln.accept().await {
+            Ok(stream) => stream,
+            Err(e) => {
+                eprintln!("failed to accept conn, err: {e}");
+                continue;
+            }
+        };
+        tokio::spawn(handle_stream(stream));
+    }
+}
+
+async fn handle_stream(mut stream: Stream) {
+    loop {
+        let req_msg = match stream.read_bytes("client say hello world!!!".len()).await {
+            Ok(msg) => msg,
+            Err(e) => {
+                eprintln!("failed to read msg, err: {e}");
+                break;
+            }
+        };
+        println!(
+            "server receive request {}",
+            String::from_utf8(req_msg.to_vec()).unwrap()
+        );
+
+        let resp_msg = "server hello world!!!";
+        stream.write_bytes(resp_msg.as_bytes()).unwrap();
+        stream.flush(true).await.unwrap();
+    }
     tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
     stream.close().await.unwrap();
 }

examples/src/hello_world/greeter_server.rs

@@ -0,0 +1,76 @@
+// Copyright 2025 CloudWeGo Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::os::unix::net::SocketAddr;
+
+use nix::unistd::unlink;
+use shmipc::{
+    Listener, compact::StreamExt, config::Config, stream::Stream, transport::DefaultUnixListen,
+};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+#[tokio::main]
+async fn main() {
+    tracing_subscriber::fmt::init();
+    let dir = std::env::current_dir().unwrap();
+    let binding = dir.join("../ipc_test.sock");
+    let uds_path = binding.to_str().unwrap();
+    _ = unlink(uds_path);
+
+    let mut ln = Listener::new(
+        DefaultUnixListen,
+        SocketAddr::from_pathname(binding).unwrap(),
+        Config::default(),
+    )
+    .await
+    .unwrap();
+
+    loop {
+        let stream = match ln.accept().await {
+            Ok(stream) => stream,
+            Err(e) => {
+                eprintln!("failed to accept conn, err: {e}");
+                continue;
+            }
+        };
+        tokio::spawn(handle_stream(stream));
+    }
+}
+
+async fn handle_stream(stream: Stream) {
+    const EXPECTED_REQ: &str = "client say hello world!!!";
+    let mut buf = vec![0; 4096];
+    let mut conn = StreamExt::new(stream);
+
+    loop {
+        match conn.read_exact(&mut buf[..EXPECTED_REQ.len()]).await {
+            Ok(len) => println!("read {len}"),
+            Err(e) => {
+                eprintln!("failed to read msg, err: {e}");
+                break;
+            }
+        }
+
+        println!(
+            "server receive request {}",
+            str::from_utf8(&buf[..EXPECTED_REQ.len()]).unwrap()
+        );
+
+        let resp_msg = "server hello world!!!";
+        conn.write_all(resp_msg.as_bytes()).await.unwrap();
+        conn.flush().await.unwrap();
+    }
+    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    conn.shutdown().await.unwrap();
+}

examples/src/hello_world/greeter_server_tokio.rs

@@ -23,6 +23,20 @@ pub enum Buf<'shm> {
 }
 
 impl Buf<'_> {
+    pub const fn len(&self) -> usize {
+        match self {
+            Self::Shm(s) => s.len(),
+            Self::Exm(b) => b.len(),
+        }
+    }
+
+    pub const fn is_empty(&self) -> bool {
+        match self {
+            Self::Shm(s) => s.is_empty(),
+            Self::Exm(b) => b.is_empty(),
+        }
+    }
+
     /// # Safety
     ///
     /// The caller must ensure that the shm buf is valid before the return value is out of usage.
@@ -217,3 +231,21 @@ impl PartialOrd<Buf<'_>> for &str {
         <[u8] as PartialOrd<[u8]>>::partial_cmp(self.as_bytes(), other)
     }
 }
+
+impl bytes::Buf for Buf<'_> {
+    fn remaining(&self) -> usize {
+        self.len()
+    }
+
+    fn chunk(&self) -> &[u8] {
+        self.as_ref()
+    }
+
+    fn advance(&mut self, cnt: usize) {
+        // no need to care cnt and len, `bytes::Buf` for `&[u8]` and `Bytes` will process it
+        match self {
+            Self::Shm(s) => bytes::Buf::advance(s, cnt),
+            Self::Exm(b) => bytes::Buf::advance(b, cnt),
+        }
+    }
+}

src/buffer/buf.rs

@@ -438,15 +438,14 @@ mod tests {
     }
 
     #[test]
+    #[should_panic]
     fn test_create_free_buffer_list() {
-        assert!(
-            BufferList::create(
-                4294967295,
-                4294967295,
-                &MmapOptions::new().len(1).map_anon().unwrap(),
-                4294967279
-            )
-            .is_err()
+        BufferList::create(
+            4294967295,
+            4294967295,
+            &MmapOptions::new().len(1).map_anon().unwrap(),
+            4294967279,
         )
+        .unwrap_err();
     }
 }