fs: support io_uring with tokio::fs::read

[Daksh14]

Motivation

We slowly use io uring everywhere here I made the simple change of supporting fs::read however it might not be as simple. Let me know if the unsafe I used is correct or not.

We currently use the blocking std::fs::MetaData to obtain file size for buffer capacity and extend the length of the vector according to the bytes read in the CQE. This implementation sounds good on paper to me.

Later we should implement an internal statx helper, in this PR or a seperate PR to make our uring implementation less painful to use. As this pr failed #7616

Lets put statx helper in different PR to avoid merging an inefficient read implementation given io uring is about being more efficient in file IO

Solution

Continue adopting io uring

strace on a tokio::fs::read after this change

io_uring_setup(256, {flags=0, sq_thread_cpu=0, sq_thread_idle=0, sq_entries=256, cq_entries=512, features=IORING_FEAT_SINGLE_MMAP|IORING_FEAT_NODROP|IORING_FEAT_SUBMIT_STABL
E|IORING_FEAT_RW_CUR_POS|IORING_FEAT_CUR_PERSONALITY|IORING_FEAT_FAST_POLL|IORING_FEAT_POLL_32BITS|IORING_FEAT_SQPOLL_NONFIXED|IORING_FEAT_EXT_ARG|IORING_FEAT_NATIVE_WORKERS
|IORING_FEAT_RSRC_TAGS|IORING_FEAT_CQE_SKIP|IORING_FEAT_LINKED_FILE|IORING_FEAT_REG_REG_RING|IORING_FEAT_RECVSEND_BUNDLE|IORING_FEAT_MIN_TIMEOUT|IORING_FEAT_RW_ATTR, sq_off=
{head=0, tail=4, ring_mask=16, ring_entries=24, flags=36, dropped=32, array=8256, user_addr=0}, cq_off={head=8, tail=12, ring_mask=20, ring_entries=28, overflow=44, cqes=64,
 flags=40, user_addr=0}}) = 9
mmap(NULL, 16384, PROT_READ|PROT_WRITE, MAP_SHARED|MAP_POPULATE, 9, 0x10000000) = 0xfaf0bf1e2000
mmap(NULL, 9280, PROT_READ|PROT_WRITE, MAP_SHARED|MAP_POPULATE, 9, 0) = 0xfaf0be71d000
epoll_ctl(5, EPOLL_CTL_ADD, 9, {events=EPOLLIN|EPOLLRDHUP|EPOLLET, data=0}) = 0
io_uring_enter(9, 1, 0, 0, NULL, 128)   = 1
futex(0xfaf0bf2557f0, FUTEX_WAIT_PRIVATE, 1, NULL) = 0
mmap(NULL, 2162688, PROT_NONE, MAP_PRIVATE|MAP_ANONYMOUS|MAP_STACK, -1, 0) = 0xfaf0be50d000
mprotect(0xfaf0be51d000, 2097152, PROT_READ|PROT_WRITE) = 0
rt_sigprocmask(SIG_BLOCK, ~[], [], 8)   = 0
clone3({flags=CLONE_VM|CLONE_FS|CLONE_FILES|CLONE_SIGHAND|CLONE_THREAD|CLONE_SYSVSEM|CLONE_SETTLS|CLONE_PARENT_SETTID|CLONE_CHILD_CLEARTID, child_tid=0xfaf0be71c150, parent_
tid=0xfaf0be71c150, exit_signal=0, stack=0xfaf0be50d000, stack_size=0x20e960, tls=0xfaf0be71c7a0} => {parent_tid=[746758]}, 88) = 746758
rt_sigprocmask(SIG_SETMASK, [], NULL, 8) = 0
futex(0xfaf0be71c850, FUTEX_WAKE_PRIVATE, 1) = 1
io_uring_enter(9, 1, 0, 0, NULL, 128)   = 1
futex(0xfaf0bf2557f0, FUTEX_WAIT_PRIVATE, 1, NULL) = -1 EAGAIN (Resource temporarily unavailable)
close(10)                               = 0

[martin-augment]

Suggested change

	let left_to_read = (size - size_read) as u32;
	let left_to_read = u32::try_from(size - size_read).unwrap_or(u32::MAX);

To properly support files bigger than 4GB.

[Daksh14]

max read size at a time is u32::MAX, we read the rest in other next iterations

[Daksh14]

In the future, if we know we're reading more than u32::MAX then we can batch 2 read requests to avoid extra syscalls

[ADD-SP]

This looks complicated, I will review it incrementally.

[mox692]

I haven't checked all of the details in read_uring.rs, but left some comments I've noticed so far.

[Darksonn]

We can handle allocation failure by error rather than panic by allocating before calling splice. Note that this doesn't allocate if size_read == 0 since the capacity is guaranteed enough in that case.

Suggested change

	r_buf.splice(back_bytes_len..back_bytes_len, temp_arr);
	r_buf.try_reserve(PROBE_SIZE)?;
	r_buf.splice(back_bytes_len..back_bytes_len, temp_arr);

[Daksh14]

@Darksonn Will this splice write in the reserved area? Because we can only index in the length and not capacity

[Darksonn]

No. The length of r_buf before the op_read is exactly back_bytes_len, so after the op_read we have back_bytes_len <= r_buf.len() which is the requirement for using splice.

[Darksonn]

This is equivalent to just moving the runtime.

Suggested change

	std::thread::spawn(move || {
	let rt: Runtime = rx.recv().unwrap();
	rt.shutdown_timeout(Duration::from_millis(300));
	done_tx.send(()).unwrap();
	});
	tx.send(rt).unwrap();
	std::thread::spawn(move || {
	rt.shutdown_timeout(Duration::from_millis(300));
	done_tx.send(()).unwrap();
	});

If you wanted an actual sleep for the loop to make some progress, then what you have now does not achieve that.

[Daksh14]

@Darksonn Should I sleep in the std::thread so the loop can make progress? Is that allowed or I shouldn't do it

[ADD-SP]

1. Poll the read(path).await manually to submit it to the kernel.
2. Then increase the semaphore.
3. Keep the spawn task pending for ever.

So that we can wait on the semaphore and then shutdown the runtime.

[ADD-SP]

Is this work for multi-thread runtime? For current thread runtime, this won't work. But I believe we can apply the similar pattern to the current thread runtime.

        rt.spawn(async move {
            let path = path[0].clone();
            let mut futs = vec![];

            // spawning a bunch of uring operations.
            for _ in 0..N {
                let path = path.clone();
                let cl = Arc::clone(&cl);
                let fut = Box::pin(read(path));
                assert_pending!(fut.poll(cx));
                futs.push(fut);
            }

            pending_forever().await;
        });

        std::thread::spawn(move || {
            rt.shutdown_timeout(Duration::from_millis(300));
            done_tx.send(()).unwrap();
        });

        done_rx.recv().unwrap();

[Darksonn]

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.

Suggested change

	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);
	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);

[Daksh14]

I have pushed a slightly different version of this

[Darksonn]

I think we can just delete this if. If the size hint is zero/missing, then we always have buf.capacity() == 0. If we delete it, then it will fall through to the loop, which will do the same thing as what this block does.

[martin-g]

Any reason why cfg_io_uring! { is not used here ?

[Daksh14]

@martin-augment its following the currently used code block macro here in tokio/src/fs/open_options.rs, I can change those in a different "refactor" PR https://github.com/tokio-rs/tokio/blob/master/tokio/src/fs/open_options.rs#L592

[martin-g]

What kind of block is meant here ?
File system blocks are usually around 4096 bytes. Here we are talking about a block of 1MiB.

Suggested change

	// Set to read max 64 blocks at time
	// Set to read max 64 MiB at time

?

[Daksh14]

@martin-augment this number came after benchmarking with io-uring against std::fs::read as its a reasonable amount of data for kernel to copy

[martin-g]

Why try_recv() ?
This may return Empty before the task has a chance to be executed.

[Daksh14]

I added an assert with recv await