Repository: tokio-rs/tokio
File: tokio/tests/fs_uring_read.rs (Rust)

Code before:
async fn read_small_large_files() {
    let (_tmp, path) = create_large_temp_file();

    let bytes = read(path).await.unwrap();

    assert_eq!(bytes, create_buf(5000));

    let (_tmp, path) = create_small_temp_file();

    let bytes = read(path).await.unwrap();

    assert_eq!(bytes, create_buf(20));
}

#[tokio::test]
async fn cancel_op_future() {
    let (_tmp_file, path): (Vec<NamedTempFile>, Vec<PathBuf>) = create_tmp_files(1);

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
    let handle = tokio::spawn(async move {
        poll_fn(|cx| {
            let fut = read(&path[0]);

            // If io_uring is enabled (and not falling back to the thread pool),
            // the first poll should return Pending.
            let _pending = Box::pin(fut).poll_unpin(cx);

            tx.send(()).unwrap();

            Poll::<()>::Pending
        })
        .await;
    });

    // Wait for the first poll
    rx.recv().await.unwrap();

    handle.abort();

    let res = handle.await.unwrap_err();
    assert!(res.is_cancelled());
}

fn create_tmp_files(num_files: usize) -> (Vec<NamedTempFile>, Vec<PathBuf>) {
    let mut files = Vec::with_capacity(num_files);
    for _ in 0..num_files {
        let mut tmp = NamedTempFile::new().unwrap();
        let buf = vec![20; 1023];
        tmp.write_all(&buf).unwrap();
        let path = tmp.path().to_path_buf();
        files.push((tmp, path));

Review comment (suggestion):
This code currently drops the read future right away. If you want the future to stay around until the `handle.abort()` call, then you need to modify this code.
```suggestion
        let fut = read(&path[0]);
        tokio::pin!(fut);
        poll_fn(|cx| {
            // If io_uring is enabled (and not falling back to the thread pool),
            // the first poll should return Pending.
            let _pending = fut.as_mut().poll(cx);
```

Code after:
    }
}

#[tokio::test]
async fn read_small_large_files() {
    let (_tmp, path) = create_large_temp_file();

    let bytes = read(path).await.unwrap();

    assert_eq!(bytes, create_buf(5000));

    let (_tmp, path) = create_small_temp_file();

    let bytes = read(path).await.unwrap();

    assert_eq!(bytes, create_buf(20));
}

#[tokio::test]
async fn cancel_op_future() {
    let (_tmp_file, path): (Vec<NamedTempFile>, Vec<PathBuf>) = create_tmp_files(1);
    let path = path[0].clone();

    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();

    let handle = tokio::spawn(async move {
        let fut = read(path.clone());
        tokio::pin!(fut);

        poll_fn(move |_| {
            // If io_uring is enabled (and not falling back to the thread pool),
            // the first poll should return Pending.
            assert_pending!(fut.as_mut().poll(&mut Context::from_waker(Waker::noop())));
            tx.send(true).unwrap();

            Poll::<()>::Pending
        })
        .await;
    });

    // Wait for the first poll

    let val = rx.recv().await;
    assert!(val.unwrap());

    handle.abort();

    let res = handle.await.unwrap_err();
    assert!(res.is_cancelled());
}