How to use tokio DuplexStream in spawned parallel tasks?

progquester 0
Viewed 49

Suppose I use a pair of DuplexStreams in an application in order to allow bidirectional communication between server and client.

use use tokio::io::duplex;
let (upstream, dwstream) = duplex(64*1024);

Then I use dwstream to send data to the client and read the data sent by the client from dwstream.

If all these operations are executed in one async block, of course there is no problem. But now, for concurrency, I spawn a new async task for reading the dwstream, and in another task I need to write data to the dwstream from time to time.

In order to move the dwstream to the new task, I have to create an Arc<Mutex<DuplexStream>> object and then clone it before it supports Move.

However, in order to read in the read task, the object must always be locked(), which actually makes it impossible to get the object in the write task.

let (upstream, dwstream) = duplex(64*1024);
let dwrc = Arc::new(Mutex::new(dwstream));
let dwrc1 = dwrc.clone();
// write task
tokio::spwan(async move {
   ... some operations ...
   dwrc.lock().unwrap().write_all(...).await;
});
tokio::spwan(async move {
   let data = dwrc1.lock().unwrap().read().await; // here I think dwstream is locked for a long time.
});

How can I resolve it?

rust-tokio rust
1 Answers

You're supposed to write to one stream in one task, and read from the other stream in the other task. Currently, you're writing to and reading from the same stream (dwstream) across the two tasks.

let (mut upstream, mut dwstream) = duplex(64*1024);
// write task
tokio::spawn(async move {
   upstream.write_all(b"abc").await;
});
// read task
tokio::spawn(async move {
    let mut buf = vec![];
    dwstream.read(&mut buf).await;
});

playground

PitaJ0
Related