net: Expose UdpSocket::try_clone

[Ralith]

Motivation

Applications sending many UDP packets (e.g. QUIC servers) benefit from parallelism, since at least Linux UDP sends scale almost linearly with thread count. However, Tokio's UdpSocket only stores one waker per direction, so if backpressure from the kernel send queue suspends multiple senders, only one gets woken.

Solution

Cloning the UDP socket should allow concurrent I/O from multiple tasks without wakers getting clobbered.

[Darksonn]

I know that registering clones of the same socket with the same epoll instance can cause problems under some circumstances, but I do not remember the details.

I cannot accept a try_clone method without an understanding of whether it introduces that kind of incompatibility.

[Ralith]

From the manpage:

What happens if you register the same file descriptor on an epoll instance twice?

You will probably get EEXIST. However, it is possible to add a duplicate (dup(2), dup2(2), fcntl(2) F_DUPFD) file descriptor to the same epoll instance. This can be a useful technique for filtering events, if the duplicate file descriptors are registered with different events masks.

So far so good.

A file descriptor is removed from an interest list only after [it is removed with using epoll_ctl(2) EPOLL_CTL_DEL or] all the file descriptors referring to the underlying open file description have been closed.

Each clone gets its own PollEvented, which registers on construction and unregisters on drop, so this shouldn't cause any surprises.

I don't see any other mention of dup'd FDs having notable interactions with epoll. Does that address your concern?

[Noah-Kennedy]

epoll tracks the underlying file description, not the descriptor itself. If you dup the socket and register the new fd with tokio, you just get redundant events on both descriptors. This isn't like SO_REUSEPORT where you basically shard the socket in some ways. You are better off putting it in an Arc.

There are valid uses of dup here for things like watching from several epoll instances, like another runtime's, but cloning and reregistering in the same runtime as the original socket doesn't really have much of a point.

[Ralith]

Added a test that exercises concurrent reads, just for insurance.

In the course of validating the test, I discovered that unlike the poll_* functions, the async fn interface of UdpSocket supports manipulation from concurrent tasks natively. Is there a way I can access that capability from within a hand-written future without tons of gratuitous boxing? That would obviate my need for try_clone.

[Darksonn]

No, but you could use ReusableBoxFuture from tokio-util?

[Ralith]

you just get redundant events on both descriptors

That's exactly the goal. I've only got concurrent tasks for outgoing data (so far), so typically operations will complete immediately, and all waiters should be woken when send space becomes newly available.

You are better off putting it in an Arc.

That makes it quite difficult to use concurrently, because the poll_* interface only supports one waiter per direction.

[Noah-Kennedy]

you just get redundant events on both descriptors

That's exactly the goal. I've only got concurrent tasks for outgoing data (so far), so typically operations will complete immediately, and all waiters should be woken when send space becomes newly available.

You are better off putting it in an Arc.

That makes it quite difficult to use concurrently, because the poll_* interface only supports one waiter per direction.

right, i forgot about that limitation

[Ralith]

I'm exploring options to use the multi-waiter-capable async fn APIs instead, but doing so through a runtime abstraction layer (previously a trait full of poll functions) is proving difficult.

[Ralith]

Managed to solve my problem using an Arc'd UDP socket. I replaced my runtime-abstract UDP socket trait's poll_send method with try_send and create_io_poller(Arc<Self>) -> Pin<Box<dyn UdpPoller>>, with

trait UdpPoller {
    fn poll_writable(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>>;
}

pin_project! {
    struct UdpPollHelper<F, T> {
        f: F,
        #[pin]
        fut: Option<T>,
    }
}

impl<F, T> UdpPollHelper<F, T> {
    fn new(f: F) -> Self {
        Self { f, fut: None }
    }
}

impl<F, T> UdpPoller for UdpPollHelper<F, T>
where
    F: Fn() -> T,
    T: Future<Output = io::Result<()>>,
{
    fn poll_writable(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
        let mut this = self.project();
        if this.fut.is_none() {
            this.fut.set(Some((this.f)()));
        }
        let result = this.fut.as_mut().as_pin_mut().unwrap().poll(cx);
        if result.is_ready() {
            this.fut.set(None);
        }
        result
    }
}

The tokio implementation then creates a poller with:

Box::pin(UdpPollHelper::new(move || {
    let socket = self.clone();
    async move { socket.io.writable().await }
}))

This makes the overall runtime abstraction a bit weirder, but it's simple enough to use and implement, so I'm satisfied. Thanks for the guidance!