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]

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

[mox692]

Overall looks good to me.

[Darksonn]

I do not think this makes sense.

Suggested change

	Ok(size_read) => {
	*offset += size_read as u64;
	read_len -= size_read;
	if read_len == 0 || size_read == 0 {
	return Ok((size_read, r_fd, r_buf));
	}
	buf = r_buf;
	fd = r_fd;
	}
	Ok(size_read) => return Ok((size_read, r_fd, r_buf)),

The current logic means that unless the buffer fits the file exactly, we perform the final read of length zero twice:

1. We call op_read, which performs a partial read and then a length-zero EOF read. It returns a non-zero length to the caller.
2. Since the caller got a non-zero length, it now needs to call op_read again to check for EOF.

Checking for EOF twice does not make sense.

[ADD-SP]

Since the fut.as_mut().poll() registers the waker, this assertion will be polled twice. Consider introduce noop waker here.

[Daksh14]

Replaced the contex of poll_fn with a noop waker @ADD-SP