add Interest::PRIORITY and ready and ready_mut functions to AsyncFd

[folkertdev]

Motivation

Closes #4885

It was impossible in tokio to await POLLPRI events. These exceptional conditions are raised under certain circumstances on unix systems (e.g. when a new device is mounted, or when a UDP socket receives messages in its error queue).

In version 8.6.0, mio adds Interest::PRIORITY, (for the linux and android operating systems).

Solution

• bumps Tokio's mio dependency to version 8.6.0
• exposes (publicly) Interest::PRIORITY
• adds priority and priority_mut functions to AsyncFd

Next steps

• I decided not to expose this functionality for other socket-like types for now. If you're working with priority events, you're doing something pretty low-level and likely are using AsyncFd already.
• I'm not sure how/if this should be tested in tokio itself (or if mio is trusted)
• I'm not sure what the policy is on bumping dependencies (and specifically mio). Everything looks fine, but I may be missing something.

[folkertdev]

I'm seeing a weird CI error here, that seems unrelated to the changes that are made. My other PR for which CI ran recently has the same problem

• https://github.com/tokio-rs/tokio/actions/runs/4478017069/jobs/7870270813?pr=5566
• https://github.com/tokio-rs/tokio/actions/runs/4477857160/jobs/7869917563?pr=5542

So I'm assuming this is because I rebased on current master?

CI Output

     Running tests/time_delay_queue.rs (target/wasm32-wasi/debug/deps/time_delay_queue-f0c21ef322a44873.wasm)

running 30 tests
test compact_change_deadline ... ok
test compact_expire_empty ... ok
test compact_remove_empty ... ok
test compact_remove_remapped_keys ... ok
test delay_queue_poll_expired_when_empty ... ok
test expire_first_key_when_reset_to_expire_earlier ... ok
test expire_second_key_when_reset_to_expire_earlier ... ok
test expires_before_last_insert ... ok
test insert_after_ready_poll ... ok
test insert_before_first_after_poll ... ok
test insert_in_past_after_poll_fires_immediately ... ok
test insert_in_past_fires_immediately ... ok
test multi_delay_at_start ... ok
test multi_immediate_delays ... ok
test multi_reset ... ok
test remove_after_compact ... ignored, FIXME: Does not seem to work with WASI
test remove_after_compact_poll ... ignored, FIXME: Does not seem to work with WASI
test remove_at_timer_wheel_threshold ... ok
test remove_entry ... ok
test remove_expired_item ... ok
Error: failed to run main module `/home/runner/work/tokio/tokio/target/wasm32-wasi/debug/deps/time_delay_queue-f0c21ef322a44873.wasm`

Caused by:
    0: failed to invoke command default
    1: error while executing at wasm backtrace:
           0: 0x14360f - <unknown>!__rust_start_panic
           1: 0x142f1d - <unknown>!rust_panic
           2: 0x142ee4 - <unknown>!std::panicking::rust_panic_with_hook::hd6f3df478dab5dc5
           3: 0x1419aa - <unknown>!std::panicking::begin_panic_handler::{{closure}}::h7be61ca999d17a10
           4: 0x1418d4 - <unknown>!std::sys_common::backtrace::__rust_end_short_backtrace::h1c7217a99e6903e8
           5: 0x142552 - <unknown>!rust_begin_unwind
           6: 0x149a0d - <unknown>!core::panicking::panic_fmt::h86214d64a55c1d57
           7: 0x15016d - <unknown>!core::panicking::panic_display::h58fb45e5567db9b3
           8: 0x14fd1d - <unknown>!core::panicking::panic_str::hef6f4fea2c6da56a
           9: 0x14fceb - <unknown>!core::option::expect_failed::he9c7eebda4091106
          10: 0x140db4 - <unknown>!<std::time::Instant as core::ops::arith::Sub<core::time::Duration>>::sub::h9d1ea7149970c4f4
          11: 0xb656b - <unknown>!<tokio::time::instant::Instant as core::ops::arith::Sub<core::time::Duration>>::sub::h8709be5f2824f4d8
          12: 0x1423c - <unknown>!time_delay_queue::repeatedly_reset_entry_inserted_as_expired::{{closure}}::h5113477f19ab55d3
          13: 0x3e836 - <unknown>!<core::pin::Pin<P> as core::future::future::Future>::poll::h4fa8cd4e55f21ac9
          14: 0x3e8ce - <unknown>!<core::pin::Pin<P> as core::future::future::Future>::poll::h642aec97e6dd8367
          15: 0x3c5c7 - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::block_on::{{closure}}::{{closure}}::{{closure}}::h1e04fe847e0eed21
          16: 0x3b926 - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::block_on::{{closure}}::{{closure}}::hda5894a20edb5dea
          17: 0x3b74d - <unknown>!tokio::runtime::scheduler::current_thread::Context::enter::h0551c0eeeb75cb0b
          18: 0x3bfd4 - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::block_on::{{closure}}::h04e2d857a61424cc
          19: 0x3ad99 - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::enter::{{closure}}::hefeb9ef4b490f0b0
          20: 0x3aceb - <unknown>!tokio::macros::scoped_tls::ScopedKey<T>::set::h768b6b08d5c1ff74
          21: 0x3bd7c - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::enter::h3549fc9053a672ee
          22: 0x3b4a4 - <unknown>!tokio::runtime::scheduler::current_thread::CoreGuard::block_on::h29331f6bd3fa002f
          23: 0x3b0df - <unknown>!tokio::runtime::scheduler::current_thread::CurrentThread::block_on::h252f5d42c4325f40
          24: 0x3aeb2 - <unknown>!tokio::runtime::runtime::Runtime::block_on::hd567519e6c4ccdf2
          25: 0x14000 - <unknown>!time_delay_queue::repeatedly_reset_entry_inserted_as_expired::h2e8d9e0d39e801f8
          26: 0x13e64 - <unknown>!time_delay_queue::repeatedly_reset_entry_inserted_as_expired::{{closure}}::hba3d5e0e60e4e59f
          27: 0x40eae - <unknown>!core::ops::function::FnOnce::call_once::h21770440c4a77129
          28: 0x78844 - <unknown>!test::__rust_begin_short_backtrace::h2f7251cabac8cc17
          29: 0x78824 - <unknown>!core::ops::function::FnOnce::call_once{{vtable.shim}}::h98ae03618d6ef851
          30: 0xa0133 - <unknown>!test::run_test::run_test_inner::hcac225d7d620b63b
          31: 0x8b209 - <unknown>!test::run_test::h2db965a4129a3fb3
          32: 0x871c7 - <unknown>!test::console::run_tests_console::h58c4d5717b25d8bf
          33: 0x9f467 - <unknown>!test::test_main::hc2411b2d7bc2431a
          34: 0x9f6b3 - <unknown>!test::test_main_static::hd744aa7211882b13
          35: 0x2c72f - <unknown>!time_delay_queue::main::h41b7fb16cef306a9
          36: 0x4107d - <unknown>!core::ops::function::FnOnce::call_once::h3c9e1736a43e028c
          37: 0x6f4ba - <unknown>!std::sys_common::backtrace::__rust_begin_short_backtrace::h47e6361a0d37dfca
          38: 0x3a287 - <unknown>!std::rt::lang_start::{{closure}}::h20bb17588a43137f
          39: 0x13a6bf - <unknown>!std::rt::lang_start_internal::h83cbd77d07ad3a3a
          40: 0x3a224 - <unknown>!std::rt::lang_start::h6ec71f408c08c5b2
          41: 0x2c753 - <unknown>!__main_void
          42: 0x1bfa - <unknown>!_start
       note: using the `WASMTIME_BACKTRACE_DETAILS=1` environment variable may show more debugging information
    2: wasm trap: wasm `unreachable` instruction executed

Stack backtrace:
   0: anyhow::error::<impl core::convert::From<E> for anyhow::Error>::from
   1: wasmtime::trap::from_runtime_box
   2: wasmtime::func::Func::call_impl
   3: wasmtime::linker::Linker<T>::module::{{closure}}::{{closure}}
   4: wasmtime::func::Func::invoke
   5: wasmtime::func::Caller<T>::with
   6: wasmtime::trampoline::func::stub_fn
   7: <unknown>
   8: <unknown>
   9: wasmtime_setjmp
  10: wasmtime_runtime::traphandlers::<impl wasmtime_runtime::traphandlers::call_thread_state::CallThreadState>::with
  11: wasmtime_runtime::traphandlers::catch_traps
  12: wasmtime::func::Func::call_impl
  13: wasmtime_cli::commands::run::RunCommand::invoke_func
  14: wasmtime_cli::commands::run::RunCommand::load_main_module
  15: wasmtime_cli::commands::run::RunCommand::execute
  16: wasmtime::main
  17: std::sys_common::backtrace::__rust_begin_short_backtrace
  18: std::rt::lang_start::{{closure}}
  19: core::ops::function::impls::<impl core::ops::function::FnOnce<A> for &F>::call_once
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/core/src/ops/function.rs:287:13
  20: std::panicking::try::do_call
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panicking.rs:483:40
  21: std::panicking::try
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panicking.rs:447:19
  22: std::panic::catch_unwind
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panic.rs:140:14
  23: std::rt::lang_start_internal::{{closure}}
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/rt.rs:148:48
  24: std::panicking::try::do_call
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panicking.rs:483:40
  25: std::panicking::try
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panicking.rs:447:19
  26: std::panic::catch_unwind
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/panic.rs:140:14
  27: std::rt::lang_start_internal
             at /rustc/001a77fac33f6560ff361ff38f661ff5f1c6bf85/library/std/src/rt.rs:148:20
  28: main
  29: <unknown>
  30: __libc_start_main
  31: <unknown>
error: test failed, to rerun pass `-p tokio-util --test time_delay_queue`

[Darksonn]

I have no idea what's up with the CI issue is, but it isn't your fault.

[Darksonn]

Would it be possible to add a test for this?

[satakuma]

I'm not sure how/if this should be tested in tokio itself

Perhaps you can take a look how mio tests POLLPRI and add a similar test using types from Tokio.

[Darksonn]

As a general note, I'm happy that you're taking a stab at this. It has been a long-standing issue.

[folkertdev]

I'm happy to, this was a fun journey into the depths of tokio and mio (something I'd wanted to do for a while).

I think I added PRIORITY in all the relevant places now. I added a test case for AsyncFd and for TcpStream, because for those I actually know how to trigger EPOLLPRI. Information on this is kind of scarce. I was under the impression that we could also listen for EPOLLPRI for a UDP socket to receive timestamping information, but I did not get that to work yet.

[satakuma]

Looks good to me, but I would like a +1 from @Darksonn as well.

[poeschel]

I have done something similar to your work here, but mine is not as "production ready" as your work. It is very well done. Thank you.
You can have a look at my code here: https://github.com/poeschel/tokio/tree/interest_prio
One thing I did differently is, I added direct priority support to the tokio TcpStream. This is maybe worth pulling in here. Priority support for the TcpStream also was in mio version 0.8.6.

And the other thing I did differently is I added a new waker for priority. (which then implies to also add the Direction::Prioriry (which sounds weird)). I thought about this and I am also not shure if it is right. But I think it is. If you have some events in the ready queue and then a priority event arrives, you want this to be handled indepently of the read events, right? This is what "priority" is for, isn't it?

[poeschel]

I think I added PRIORITY in all the relevant places now. I added a test case for AsyncFd and for TcpStream, because for those I actually know how to trigger EPOLLPRI. Information on this is kind of scarce. I was under the impression that we could also listen for EPOLLPRI for a UDP socket to receive timestamping information, but I did not get that to work yet.

Regarding a test for AsyncFd with EPOLLPRI events:
Linux gpio sysfs is able to fire EPOLLPRI events, when a gpio is configured to fire interrupts. You do not need real hardware for this, because there is a gpio_mockup driver in Linux. But the whole thing is a bit hard to setup, because most distributions do not have this driver activated, so you have to recompile your kernel and to setup and use it, you have to be a priviledged user.
Is it worth adding a test with these requirements?

[Darksonn]

Is it worth adding a test with these requirements?

Eh, maybe not. We wouldn't be able to test that in CI. However, I would like to know whether anyone has tried running the code and seen it work at least once for a sanity check.

[poeschel]

Is it worth adding a test with these requirements?

Eh, maybe not. We wouldn't be able to test that in CI. However, I would like to know whether anyone has tried running the code and seen it work at least once for a sanity check.

I can test my use-case and the unit tests I wrote for my fork on this code tomorrow.

[folkertdev]

note for the future: if we add the Ready::READ_CLOSED to mask above, we should also add it here

[folkertdev]

thanks for your input @poeschel!

re. pulling in more stuff: I'm open to it, but also think a separate later PR would be fine.

[poeschel]

Is it worth adding a test with these requirements?

Eh, maybe not. We wouldn't be able to test that in CI. However, I would like to know whether anyone has tried running the code and seen it work at least once for a sanity check.

I can test my use-case and the unit tests I wrote for my fork on this code tomorrow.

As promised, I tested today with my unit test and gpio sysfs in linux and my real world use-case with video4linux and both do work in principle. But I think we really should have a seperate worker for priority events:

1. The tcp stream use-case: The linux tcp stack supports sending things over a tcp connection with MSG_OOB set. On the receiving side this is then received over the same tcp connection, but is priorized and with POLLPRI set. This is an urgent message, that will arrive ahead of the normal buffered tcp stream actual in transit. By reusing the read waker for priority events in tokio, the priority mechanism will not work anymore. You will receive the message, but later than possible if we had another independent waker.
2. The video4linux use case: Linux does support POLLPRI events on certain video4linux devices. There are different events to receive, like frame sync, resolution changed or motion detected or something like that. That means POLLPRI is used like a second channel on the same video device. You receive your video pixel data by reading from the video device and you receive interesing events over POLLPRI. These should not be intermixed. This will not work very well with tokio when we reuse the read waker for priority events.

I think it is better to add a seperate priority waker, or am I missing something?

[poeschel]

re. pulling in more stuff: I'm open to it, but also think a separate later PR would be fine.

You are right! This is about AsyncFd here. The TcpStream can be another PR.

[Darksonn]

Adding a separate waker for priority events makes sense to me.

[folkertdev]

@poeschel I've been looking at this. One thing I don't see is how the current example would use this new priority direction/waker. You seem to be using it in a custom Stream implementation.

what are the consequences of adding the custom waker for the functionality in this PR so far? or is it really a primitive that is needed for e.g. TCP?

[carllerche]

1. The tcp stream use-case: The linux tcp stack supports sending things over a tcp connection with MSG_OOB set. On the receiving side this is then received over the same tcp connection, but is priorized and with POLLPRI set. This is an urgent message, that will arrive ahead of the normal buffered tcp stream actual in transit. By reusing the read waker for priority events in tokio, the priority mechanism will not work anymore. You will receive the message, but later than possible if we had another independent waker.

I'm not following this. Tokio has a Ready struct. If priority is exposed there, one could await readiness, check if priority is set, and process priority events first, if needed. I don't think you need a special waker.

It may be that AsyncFd is missing a method that returns the current readiness.

[folkertdev]

here's what I understand:

Because we don't have async traits, traits that deal with async often have functions that return Poll<T>. @poeschel 's branch has an example using futures::stream::Stream, but I think tokio's AsyncRead and AsyncWrite are similar.

impl Stream for AsyncV4l2EventHandle {
    type Item = io::Result<i32>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
        loop {
            let blah = self.asyncfd.poll_priority_ready_mut(cx);
            let mut guard = ready!(blah)?;
            let res = guard.try_io(|inner| inner.get_mut().get_event());
            guard.clear_ready();
            match res {
                Err(TryIoError { .. }) => {
                    continue;
                },
                Ok(Ok(event)) => {
                    return Poll::Ready(Some(Ok(event)));
                },
                Ok(Err(err)) => {
                    return Poll::Ready(Some(Err(err.into())));
                }
            }
        }
    }
}

The poll_priority_ready_mut function is analogous to poll_read_ready. I believe the Stream implementation above runs into exactly the reason that these poll_ functions exist, from their docs:

This method is intended for cases where creating and pinning a future
via [readable] is not feasible. Where possible, using [readable] is
preferred, as this supports polling from multiple tasks at once.

There is nowhere to pin a future in that impl (right?)

---

So assuming that we want poll_priority_ready_mut, it needs to be implemented. This is how poll_read_ready_mut works

impl AsyncFd<T> {
    pub fn poll_read_ready_mut<'a>(
        &'a mut self,
        cx: &mut Context<'_>,
    ) -> Poll<io::Result<AsyncFdReadyMutGuard<'a, T>>> {
        let event = ready!(self.registration.poll_read_ready(cx))?;

        Ok(AsyncFdReadyMutGuard {
            async_fd: self,
            event: Some(event),
        })
        .into()
    }
}

...

impl Registration {
    // Uses the poll path, requiring the caller to ensure mutual exclusion for
    // correctness. Only the last task to call this function is notified.
    pub(crate) fn poll_read_ready(&self, cx: &mut Context<'_>) -> Poll<io::Result<ReadyEvent>> {
        self.poll_ready(cx, Direction::Read)
    }
}

so working by analogy, we need some Direction::Priority to pass to poll_ready. and that ultimately needs an extra waker for priority events.

---

The above makes sense to me, and the conclusion seems to be that for correct behavior, an extra waker for priority events is needed. But maybe I'm missing something here?

[carllerche]

To avoid the waker issue, I would not expose priority via a "poll" method. The async fns do not need a waker per readiness type. You can always convert a future returned by an async fn to a nameable future by doing Box<dyn Future<...>>.

[carllerche]

It seems to me the key issue here is AsyncFd does not have an async fn ready like TcpStream does (here). Let's start with a PR that adds that.

Then, we can add "priority" to Interest, and the way to listen for priority events is to call fd.ready(Interest::PRIORITY).await (or whatever).

[folkertdev]

this PR already adds

pub async fn priority_ready<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {

which is follows the pattern of

pub async fn readable<'a>(&'a self) -> io::Result<AsyncFdReadyGuard<'a, T>> {

and then adds a PRIORITY variant to Interest. All of that works fine.

---

implementing ready looks like a nice addition, but I think you run into issues when an IO operation blocks. At that point you need to clear the readiness and with just a Ready value that is not possible.

The AsyncFdReadyGuard returned by readable and friends has only an Option<ReadyEvent>, and therefore cannot always give you a Ready value (given the current implementation)

---

I'm not sure how boxing helps in the example above. The problem is that there is no way to hold on to a future between calls to poll_next. The code below would compile

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
    loop {
        let fut = self.asyncfd.priority_ready();
        let mut guard = ready!(fut)?;

        ...
    }
}

but the call to priority_ready re-creates the future every time. I'm assuming that has performance downsides, but maybe it's fine.

---

all in all, I'd suggest that we declare polling out-of-scope for this PR, and focus just on the two functions from the title, priority_ready and priority_ready_mut.

[folkertdev]

wasm/wasi CI fails with

error[E0433]: failed to resolve: use of undeclared crate or module `log`
  --> /home/runner/.cargo/registry/src/index.crates.io-6f17d[22](https://github.com/tokio-rs/tokio/actions/runs/5114795867/jobs/9195428801?pr=5566#step:10:23)bba15001f/mio-0.8.7/src/sys/wasi/mod.rs:82:13
   |
82 |             log::warn!(
   |             ^^^ use of undeclared crate or module `log`

which is weird. In the Cargo.toml I specify we now use version 0.8.6 of mio, but the Cargo.lock file does actually show 0.8.7. Maybe something with feature flags changed in that version of mio?

[Darksonn]

The mio issue has been fixed upstream. I reran ci and it should pass now.

[carllerche]

LGTM, thanks

[Darksonn]

One nit, otherwise LGTM.

[carllerche]

Awesome, thanks for the work!

[folkertdev]

thanks for all the review/help, I really like where this ended up!

[poeschel]

Also, thank you very much from my side for your work!