dev(owned): trying to have a stream and a sink at the same time using a mutex. but it deadlocks

Author: JadKHaddad

Cargo.lock

@@ -20,6 +20,8 @@ httparse = { version = "1.10.1", default-features = false }
 base64 = { version = "0.22.1", default-features = false }
 sha1 = { version = "0.10.6", default-features = false }
 heapless = { version = "0.9.1", default-features = false }
+embassy-sync = { version = "0.7.2", default-features = false }
+futures = { version = "0.3.31", default-features = false }
 
 [dev-dependencies]
 rand = { version = "0.9.1", features = ["std_rng"] }

Cargo.toml

@@ -0,0 +1,82 @@
+//! Run with
+//!
+//! ```not_rust
+//! cargo run --example owned
+//! ```
+
+use std::pin::pin;
+
+use embassy_sync::blocking_mutex::raw::NoopRawMutex;
+use embedded_io_adapters::tokio_1::FromTokio;
+use futures::{SinkExt, StreamExt};
+use rand::{SeedableRng, rngs::StdRng};
+use tokio::net::TcpStream;
+use websocketz::{Message, WebSocket, http::Header, options::ConnectOptions};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let domain = "websockets.chilkat.io";
+
+    let addr = tokio::net::lookup_host((domain, 80))
+        .await?
+        .next()
+        .ok_or("Failed to resolve domain")?;
+
+    let stream = TcpStream::connect(addr).await?;
+
+    let read_buf = &mut [0u8; 8192 * 2];
+    let write_buf = &mut [0u8; 8192 * 2];
+    let fragments_buf = &mut [0u8; 8192 * 2];
+    let rng = StdRng::from_os_rng();
+
+    let websocketz = WebSocket::connect::<16>(
+        ConnectOptions::default()
+            .with_path_unchecked("/wsChilkatEcho.ashx")
+            .with_headers(&[Header {
+                name: "Host",
+                value: domain.as_bytes(),
+            }]),
+        FromTokio::new(stream),
+        rng,
+        read_buf,
+        write_buf,
+        fragments_buf,
+    )
+    .await?;
+
+    println!(
+        "Number of framable bytes after handshake: {}",
+        websocketz.framable()
+    );
+
+    let websocketz = websocketz.owned::<1024, NoopRawMutex>();
+
+    let (stream, sink) = websocketz.split();
+    let (mut stream, mut sink) = (pin!(stream), pin!(sink));
+
+    sink.send(Message::Text("Hello, WebSocket!")).await?;
+
+    loop {
+        tokio::select! {
+            msg = stream.next() => match msg.transpose()? {
+                None => {
+                    println!("EOF");
+
+                    break;
+                }
+                Some(msg) => {
+                    println!("Received message: {msg:?}");
+                }
+            },
+            _ = tokio::time::sleep(tokio::time::Duration::from_secs(1)) => {
+                println!("Sending message...");
+
+                sink.send(Message::Text("Hello, WebSocket!")).await?; // This deadlocks
+
+                println!("Message sent!");
+            }
+        }
+    }
+
+    Ok(())
+}

examples/owned.rs

@@ -61,7 +61,7 @@ impl ReadAutoCaller {
     // TODO: see comments in on WebSocket::next()
     // TODO: delete
     #[allow(clippy::too_many_arguments)]
-    pub async fn call_next<const N: usize, F, RW, Rng>(
+    pub async fn call_owned<const N: usize, F, RW, Rng>(
         &self,
         auto: F,
         websocket: &mut WebSocketCore<'_, RW, Rng>,

src/functions.rs

@@ -1,17 +1,16 @@
-use core::cell::RefCell;
-
+use embassy_sync::{blocking_mutex::raw::RawMutex, mutex::Mutex};
 use embedded_io_async::{Read, Write};
 use framez::{
     Framed,
     state::{ReadState, ReadWriteState, WriteState},
 };
+use futures::{Sink, Stream};
 use rand::RngCore;
 
 use crate::{
     FragmentsState, Frame, FramesCodec, Message, OnFrame, OwnedMessage, WebSocketCore,
     error::{Error, ProtocolError},
     http::{Request, Response},
-    next,
     options::{AcceptOptions, ConnectOptions},
 };
 
@@ -309,6 +308,16 @@ impl<'buf, RW, Rng> WebSocket<'buf, RW, Rng> {
     pub const fn caller(&self) -> crate::functions::ReadAutoCaller {
         crate::functions::ReadAutoCaller
     }
+
+    /// TODO
+    pub fn owned<const N: usize, M>(self) -> OwnedWebSocket<'buf, N, RW, Rng, M>
+    where
+        M: RawMutex,
+    {
+        OwnedWebSocket {
+            inner: Mutex::new(self),
+        }
+    }
 }
 
 /// Read half of a WebSocket connection.
@@ -465,28 +474,25 @@ impl<'buf, RW, Rng> WebSocketWrite<'buf, RW, Rng> {
 }
 
 #[derive(Debug)]
-pub struct OwnedWebSocket<'buf, const N: usize, RW, Rng> {
-    // XXX: the idea of the refcell is to have a stream and a sink at the same time using the same core
-    // but this would panic when using select on the stream and then using the sink to send a message, because the stream would have a mutable reference to the core and then the sink would try to borrow it again.
-    // See clippy warning
-    // I think we can only have a stream or a sink at a time, mutably borrowing the core
-    core: RefCell<WebSocket<'buf, RW, Rng>>,
+pub struct OwnedWebSocket<'buf, const N: usize, RW, Rng, Mtx: RawMutex> {
+    inner: Mutex<Mtx, WebSocket<'buf, RW, Rng>>,
 }
 
-impl<'buf, const N: usize, RW, Rng> OwnedWebSocket<'buf, N, RW, Rng>
+impl<'buf, const N: usize, RW, Rng, Mtx> OwnedWebSocket<'buf, N, RW, Rng, Mtx>
 where
     RW: Read + Write,
     Rng: RngCore,
+    Mtx: RawMutex,
 {
     pub async fn next(
         &self,
     ) -> Option<Result<Result<OwnedMessage<N>, heapless::CapacityError>, Error<RW::Error>>> {
-        let mut websocket = self.core.borrow_mut();
+        let mut websocket = self.inner.lock().await;
 
         match 'next: loop {
             match websocket
                 .caller()
-                .call_next::<N, _, _, _>(websocket.auto(), &mut websocket.core)
+                .call_owned::<N, _, _, _>(websocket.auto(), &mut websocket.core)
                 .await
             {
                 Some(Ok(None)) => continue 'next,
@@ -500,4 +506,45 @@ where
             Some(Err(err)) => Some(Err(err)),
         }
     }
+
+    pub async fn send(&self, message: Message<'_>) -> Result<(), Error<RW::Error>> {
+        let mut websocket = self.inner.lock().await;
+
+        websocket.send(message).await
+    }
+
+    pub fn stream(
+        &self,
+    ) -> impl Stream<
+        Item = Result<Result<OwnedMessage<N>, heapless::CapacityError>, Error<RW::Error>>,
+    > + '_ {
+        futures::stream::unfold((self, false), move |(this, errored)| async move {
+            if errored {
+                return None;
+            }
+
+            match this.next().await {
+                Some(Ok(item)) => Some((Ok(item), (this, false))),
+                Some(Err(err)) => Some((Err(err), (this, true))),
+                None => None,
+            }
+        })
+    }
+
+    pub fn sink(&self) -> impl Sink<Message<'_>, Error = Error<RW::Error>> + '_ {
+        futures::sink::unfold(self, |this, item: Message<'_>| async move {
+            this.send(item).await?;
+
+            Ok(this)
+        })
+    }
+
+    pub fn split(
+        &self,
+    ) -> (
+        impl Stream<Item = Result<Result<OwnedMessage<N>, heapless::CapacityError>, Error<RW::Error>>>,
+        impl Sink<Message<'_>, Error = Error<RW::Error>>,
+    ) {
+        (self.stream(), self.sink())
+    }
 }