tokio: write_all hangs after read on a pty slave

Version tokio v1.19.2 (*)

Platform x86_64 x86_64 GNU/Linux

Description

I tried to read the data from a PTY slave and write the user input back to the pty slave. But it hangs during the first input write to the file

I tried this code:

use tokio::fs::OpenOptions;
use tokio::io::{stdin, stdout, AsyncReadExt, AsyncWriteExt};
use tokio::select;


enum EventLoopResult {
    // Input received
    Input(usize),
    // Output received
    Output(usize),
    // An error occurred
    Error(anyhow::Error),
}

// Run an interactive console session of pty slave
pub(crate) async fn run_stream(file_name: &str) -> Result<()> {

    let mut file = OpenOptions::new()
        .write(true)
        .read(true)
        .create(true)
        .open(file_name)
        .await?;

    let mut stdin_stream = stdin();
    let mut stdout_stream = stdout();


    let mut console_buf = [0u8; 256];
    let mut stdin_buf = [0u8; 256];

    loop {
        let loop_result = select! {
            bytes_read = file.read(&mut console_buf[..]) => {
                match bytes_read {
                    Err(e) => EventLoopResult::Error(e.into()),
                    Ok(b) => EventLoopResult::Output(b),
                }
            },
            bytes_read = stdin_stream.read(&mut stdin_buf[..]) => {
                match bytes_read {
                    Err(e) => EventLoopResult::Error(e.into()),
                    Ok(b) => EventLoopResult::Input(b),
                }
            }
        };

        // Process results
        match loop_result {
            EventLoopResult::Error(e) => {
                return Err(e);
            }
            EventLoopResult::Input(bytes_read) => {
                println!("Bytes read: {}", bytes_read);
                if process_input(&stdin_buf[0..bytes_read], &mut file).await? {
                    // Requested shutdown
                    break;
                }
            }
            EventLoopResult::Output(bytes_read) => {
                stdout_stream.write_all(&console_buf[0..bytes_read]).await?;
                stdout_stream.flush().await?;
            }
        };
    }

    Ok(())
}


/// Process input data
async fn process_data<T>(buffer: &[u8],  file: &mut T) -> Result<bool>
where
    T: AsyncWriteExt + Unpin,
{

    file.write_all(&[*b]).await?;
    file.flush().await?;
    Ok(true)
}

I expected to see this happen: Should continue the read and write

Instead, this happened: Blocked on write_all