Wait for messages in the error queue

[folkertdev]

Is your feature request related to a problem? Please describe.

I work on ntpd-rs, an NTP implementation in rust. A big part of how it works is to configure a UDP socket to record timestamps when a message is sent or received, and then retrieving these timestamps. The timestamp is not available immediately (we've observed this for external hardware clocks, software timestamps do seem to be effectively instant).

These timestaps end up in the error queue (libc::MSG_ERRQUEUE) of the socket. Tokio does not have a way to await or read these messages currently. The reading of these messages is probably too low-level for tokio, but being able to await the messages appearing would be extremely useful.

Describe the solution you'd like

With #5566 heading in a promising direction, I think from a user perspective, the best API would be

let socket: AsyncFd<std::net::UdpSocket> = ...;

loop { 
    let mut guard = socket.ready(Interest::ERROR).await?;

    if guard.ready().is_error() {
        match read_from_error_queue(socket.get_ref()) {
            Ok(msg) => {
                println!("read {msg}");
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                guard.clear_ready_exact(Ready::ERROR);
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }
}

Adding Ready::ERROR is straightforward. But adding an error interest is really not. Currently, a tokio interest is just a wrapper around the mio::Interest:

pub struct Interest(mio::Interest);

impl Interest {
    /// Interest in all readable events.
    ///
    /// Readable interest includes read-closed events.
    pub const READABLE: Interest = Interest(mio::Interest::READABLE);

    ...

But mio::Interest::ERROR does not exist, with a good reason: the error "interest" is always registered by the OS (see tokio-rs/mio#1672 (comment)). Therefore (not) using mio::Interest::ERROR would have no effect either way, and would be confusing.

But for public tokio APIs, the interest is specifically what you listen for, and hence what you'll be notified for. That would mean that tokio would have to implement Interest itself (sort of inlining the mio implementation). I don't see any technical issues with that approach, and I don't think there are performance downsides really: all of the logic to convert to mio::Interest could use const fn.

But it's kind of inelegant because it duplicates a bunch of logic from mio, and increases maintenance burden and the change of inconsistencies. I'd say it's worth it, but there is a cost.

Describe alternatives you've considered

Another option would be to always include the error readiness, like mio:

impl Ready { 
    pub(crate) fn from_interest(interest: Interest) -> Ready {
        let mut ready = Ready::EMPTY;

        if interest.is_readable() {
            ready |= Ready::READABLE;
            ready |= Ready::READ_CLOSED;
        }

        // ...

        // just always include the error readiness
        ready |= Ready::ERROR

        ready
    }
}

this would sort of work, but it's extremely inelegant. The example from earlier would have to pick a different, arbitrary Interest, and then just ignore this interest and just see if the readiness includes the error flag.

loop { 
    let mut guard = socket.ready(Interest::READABLE).await?;

    if guard.ready().is_error() {
        // ...
    }
}

I also think this will lead to problems with e.g. readable firing but a read stil blocking. Code should be robust against that, but it's still inefficient to read when we know it would block.

Additional context

The error queue is used for many things besides timestamps. Here is an example that puts a explicitly puts a message on the error queue. It may come in handy for testing. The examples that I know of are all from unix systems, but I don't see any harm in exposing error readiness across platforms (mio already supports it anyway).

I'm happy to work on this, but would like some confirmation on this path not being totally unreasonable before starting.

[Darksonn]

Do you have any resources that explain what the error queue is?

[folkertdev]

Fair question, this stuff is kind of scattered around the internet. Also I have only really worked with this one use case, but here goes:

On unix systems, we have

ssize_t recvmsg(int sockfd, struct msghdr *msg, int flags);

One of the flag bits is MSG_ERRQUEUE:

MSG_ERRQUEUE (since Linux 2.2)
This flag specifies that queued errors should be received
from the socket error queue. The error is passed in an
ancillary message with a type dependent on the protocol
(for IPv4 IP_RECVERR). The user should supply a buffer of
sufficient size. See cmsg(3) and ip(7) for more
information. The payload of the original packet that
caused the error is passed as normal data via msg_iovec.
The original destination address of the datagram that
caused the error is supplied via msg_name.

the docs page provides some further information on the actual content of the data: https://man7.org/linux/man-pages/man2/recvmsg.2.html. The received values are called "control messages", which have a bunch of different uses.

Sadly, there is not some nice exhaustive list of when something actually ends up in the error queue. Generally speaking it allows you to get some extra info on what actually went wrong with a socket operation, e.g.:

    // Iterate through control messages
    for (cmsg = CMSG_FIRSTHDR(msg); cmsg != NULL; cmsg = CMSG_NXTHDR(msg, cmsg)) {
        // Check if it's a IPPROTO_IP control message with type IP_RECVERR
        if (cmsg->cmsg_level == IPPROTO_IP && cmsg->cmsg_type == IP_RECVERR) {
            ee = (struct sock_extended_err*)CMSG_DATA(cmsg);

            // Check if it's a destination unreachable error
            if (ee->ee_origin == SO_EE_ORIGIN_ICMP && ee->ee_type == ICMP_DEST_UNREACH) {
                struct sockaddr_in* addr = (struct sockaddr_in*)ee->ee_info;
                char ipstr[INET_ADDRSTRLEN];
                inet_ntop(AF_INET, &(addr->sin_addr), ipstr, INET_ADDRSTRLEN);

                printf("Destination Unreachable: Host %s is unreachable.\n", ipstr);
                // Handle the unreachable condition appropriately
                // ...
            }
        }
    }

Besides errors, this mechanism is also (ab)used to send ancillary data. One (wild) use case is to send file descriptors between processes. this blog post has some details for the curious reader. (https://man7.org/linux/man-pages/man7/unix.7.html, search for "ancillary")

Timestamps also fall into this ancillary data category. The socket is configured to generate timestamps, and the timetamp control messages appear on the error queue after the IO operation is done.

Is that helpful? there isn't really a good central place that explains what the error queue is, as far as I can find. so I tried to give some extra context here.

[malaire]

Not sure if this applies here, but Rust currently has experimental support for socket ancillary data. See experimental types here: https://doc.rust-lang.org/std/os/unix/net/index.html and also rust-lang/rust#76915

[Darksonn]

So, this is the same as socket ancillary data?

[folkertdev]

socket ancillary data is a part of it, although the current https://doc.rust-lang.org/std/os/unix/net/enum.AncillaryData.html seems to implement only a very limited set of control messages that can be received. In practice the pattern match at https://doc.rust-lang.org/src/std/os/unix/net/ancillary.rs.html#394 has way more branches. But I don't think all that really belongs in std (and I'm not entirely sure if/why the parts they picked should be).

In general the error queue is an additional mechanism for receiving data on a socket. It was initially designed for receiving extra error info without meddling with further "normal" data bytes that the socket was also receiving, but has also been (ab)used for other kinds of information.

I hope this is useful background information, but I'd like to clarify again that I don't think tokio should concern itself with receiving messages on the error queue. it's very low-level and niche, and hard to capture in a nice and efficient api in general. What I'm asking for specifically is the ability to await something entering the error queue, which is equivalent to waiting for a mio event where is_error is true.

[folkertdev]

So, the solution I propose does not actually work. When an interest is just the hypothetical Intersest::ERROR on the tokio side, there is no equivalent of that interest in mio. That means it is impossible to add such a source to the reactor:

self.registry
    .register(source, mio::Token(token), interest.to_mio())?; // <- no interest to provide there

and so that means we're kind of stuck here. As I mentioned in the OP an arbitrary Interest can be used, in combination with adding Ready::ERROR, because the OS (and hence mio) will always report error events, regardless of the interest.

That approach is recommended on this mio issue.

I think you should wait for read readiness, that should trigger error as well

I still think that is inelegant, and at least epoll can be configured to wait for just error events (by just not registering interest in anything else).

But the Ready::ERROR variant is useful regardless, fixes my problem (if inelegantly), exposes existing mio functionality, and may have usage way outside of the error queue (e.g. on other operating systems).

I also made a mio version of the earlier example: gist.