Possible deadlock when using smol::io::split

[bradleyharden]

Hi,

I ran into a strange issue where a test of mine was hanging. Interestingly, changing from an instance of Async<serialport::TTYPort> to Async<mio_serial::SerialStream> seems to solve the problem. At first, I thought that was significant, but now I think it may just be coincidental.

While trying to create a minimal reproduction, it looks like swapping the order of smol::spawn calls influences the outcome. That fact, combined with knowledge that the ReadHalf and WriteHalf types returned by smol::io::split wrap instances of Arc<Mutex<T>>, makes me strongly suspect this is a deadlock of some kind. That's reinforced by the fact that I'm seeing slightly different behavior between my minimal example and my real code. Moreover, I've now seen variation in the behavior from day to day. Finally, I should note that the problem goes away completely if I spawn two instances of smol::io::copy. However, as mentioned in smol-rs/smol#270, one of these streams can't be implemented with smol::io::copy.

Here is the reproduction code. It's specific to my system, because it communicates with my serial device. But it could be easily adapted to some other device.

When I run this code, the blocking_serialport and async_mio_serial tests pass, but the async_serialport test hangs. However, if I move the spawn of smol::io::copy to the indicated position, all tests pass.

use std::io::{Read, Write};
use std::os::fd::AsRawFd;
use std::os::unix::net::UnixStream;
use std::time::Duration;

use smol::io::{AsyncReadExt, AsyncWriteExt};
use smol::Async;

const REQUEST: [u8; 12] = [
    0xC0, 0x00, 0x01, 0x00, 0x52, 0x00, 0x01, 0x77, 0xa0, 0x35, 0xc4, 0xC0,
];
const RESPONSE: [u8; 12] = [
    0xC0, 0x00, 0x20, 0x04, 0x81, 0x00, 0x01, 0xf9, 0xf5, 0x94, 0xaf, 0xC0,
];

fn adapter(device: impl Read + Write + AsRawFd + Send + 'static) -> Async<UnixStream> {
    let (a, b) = UnixStream::pair().unwrap();
    let b = Async::new(b).unwrap();
    let device = Async::new(device).unwrap();
    let (mut b_reader, b_writer) = smol::io::split(b);
    let (device_reader, mut device_writer) = smol::io::split(device);
    smol::spawn(smol::io::copy(device_reader, b_writer)).detach();
    smol::spawn(async move {
        let mut buf = [0; 1024];
        loop {
            println!("Starting loop");
            let n = match b_reader.read(&mut buf).await {
                Ok(n) => n,
                Err(err) => {
                    println!("{err:?}");
                    continue;
                }
            };
            println!("Read {n} bytes");
            match device_writer.write_all(&buf[..n]).await {
                Ok(()) => {}
                Err(err) => {
                    println!("{err:?}");
                    continue;
                }
            };
            println!("Wrote {n} bytes");
        }
    })
    .detach();
    // FIXME: Moving the copy spawn here allows the test to pass
    //smol::spawn(smol::io::copy(device_reader, b_writer)).detach();
    Async::new(a).unwrap()
}

fn serialport_device() -> serialport::TTYPort {
    serialport::new("/dev/ttyUSB0", 230400)
        .timeout(Duration::from_secs(1))
        .open_native()
        .unwrap()
}

fn mio_serial_device() -> mio_serial::SerialStream {
    use mio_serial::SerialPortBuilderExt;

    mio_serial::new("/dev/ttyUSB0", 230400)
        .timeout(Duration::from_secs(1))
        .open_native_async()
        .unwrap()
}

fn test_blocking(mut device: impl Read + Write) {
    let mut buf = [0; RESPONSE.len()];
    device.write_all(&REQUEST).unwrap();
    device.read_exact(&mut buf).unwrap();
    assert_eq!(buf, RESPONSE);
}

fn test_async(device: impl Read + Write + AsRawFd + Send + 'static) {
    let mut socket = adapter(device);
    let mut buf = [0; RESPONSE.len()];
    smol::block_on(async {
        socket.write_all(&REQUEST).await.unwrap();
        socket.read_exact(&mut buf).await.unwrap();
    });
    assert_eq!(buf, RESPONSE);
}

#[test]
fn blocking_serialport() {
    let device = serialport_device();
    test_blocking(device);
}

#[test]
fn async_serialport() {
    let device = serialport_device();
    test_async(device);
}

#[test]
fn async_mio_serial() {
    let device = mio_serial_device();
    test_async(device);
}