net: add function for listening to socket readiness events

[carllerche]

TcpStream, TcpListener, UdpSocket, ... should provide async functions for receiving readiness events.

Something like:

async fn ready(interest: ???) -> io::Result<???>;

This would allow the user to hook into arbitrary readiness events and receive notifications such as HUP.

We also should provide poll_ready_ready and poll_write_ready for each socket type.

Open question

Should the ready function return a ReadyGuard similar to AsyncFd or should it clear readiness in the try_read / try_write function?

Refs: #2903

[Darksonn]

Progress on this issue:

• TcpStream net: add TcpStream::ready and non-blocking ops #3130
• UdpSocket net: add UdpSocket readiness and non-blocking ops #3138
• UnixDatagram net: expose poll_* methods on UnixDatagram #3223 (issue: Expose poll_* methods on UnixDatagram #2853)
• UnixStream net: add UnixStream readiness and non-blocking ops #3246

[benesch]

It's still not entirely clear to me whether it's possible to use this API to wait for just HUP. I managed to get something working here, but it's suboptimal: MaterializeInc/materialize#5235

The tl;dr is what you have to write currently is something like this, which polls Interest::READABLE and then inspects whether the resulting Ready indicates is_read_closed:

fn write_stream(conn: TcpStream, stream: &dyn Stream<Item = Vec<u8>>) -> Result<(), &'static str> {
    loop {
        match time::timeout(Duration::from_secs(1), stream.next()).await {
            Ok(None) => return Ok(()),
            Ok(Some(data)) => conn.write_all(data).await,
            Err(_) => {
                let ready = conn.ready(Interest::READABLE).await?;
                if ready.is_read_closed() {
                    return Err("conn broke");
                }
            }
        }
    }
}

But ideally you'd be able to block on the READ_CLOSED event directly, or something:

fn write_stream(conn: TcpStream, stream: &dyn Stream<Item = Vec<u8>>) -> Result<(), &'static str> {
    loop {
        select! {
            msg = stream.next() => match msg {
                None => return Ok(()),
                Some(data) => conn.write_all(data).await,
            }
            _ = conn.ready(Interest::READ_CLOSED) => return Err("conn broke"),
        }
    }
}

[benesch]

Oh, by the way, #3246 merged, so I think technically this issue is complete. I can file a new issue for the HUP thing, if you prefer.

[Darksonn]

Yeah, please file a new issue.