How to wait for a socket to be writable? udp / Interest::READABLE | Interest::WRITABLE

[feschber]

How to wait for a socket to be writable? udp / Interest::READABLE | Interest::WRITABLE

[feschber]

Well thanks a lot, you definitely helped me! I will close this, as it has nothing to do with mio :)

[Thomasdezeeuw]

In general it's ok to try an I/O operations even if you don't get an event for it. However you have to handle the WouldBlock error returned at that point, but this can be returned even after you do get an event, so really it doesn't change the error handling.

As for the not getting a writable event, can you share some code that demonstrates that? Because we have tests for it:

mio/tests/udp_socket.rs

Lines 123 to 145 in 9f21ce1

	poll.registry()
	.register(
	&mut socket1,
	ID1,
	Interest::READABLE.add(Interest::WRITABLE),
	)
	.expect("unable to register UDP socket");
	poll.registry()
	.register(
	&mut socket2,
	ID2,
	Interest::READABLE.add(Interest::WRITABLE),
	)
	.expect("unable to register UDP socket");
	expect_events(
	&mut poll,
	&mut events,
	vec![
	ExpectEvent::new(ID1, Interest::WRITABLE),
	ExpectEvent::new(ID2, Interest::WRITABLE),
	],
	);

[feschber]

Thank you!

That makes sense.

As for the TX events:
The following is what my usecase boils down to:

# Cargo.toml
[package]
name = "udp_test"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
libc = "0.2.148"
mio = { version = "0.8", features = ["os-ext", "os-poll", "net"] }

use std::{io::{self, ErrorKind, Write}, net::SocketAddr, thread, os::fd::AsRawFd, time::Duration};

use mio::{Token, Poll, unix::pipe::Receiver, Events, Interest, net::UdpSocket};

const UDP: Token = Token(0);
const PRODUCER: Token = Token(1);

fn main() -> io::Result<()> {

    // register event sources
    let port = 4242u16;
    let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
    let mut socket = UdpSocket::bind(listen_addr)?;
    let (pipe_tx, mut pipe_rx) = mio::unix::pipe::new()?;
    pipe_rx.set_nonblocking(true)?;
    let mut poll = Poll::new()?;

    poll.registry().register(&mut socket, UDP, Interest::READABLE | Interest::WRITABLE)?;
    poll.registry().register(&mut pipe_rx, PRODUCER, Interest::READABLE)?;

    // producer thread writing stuff onto the pipe in intervals
    thread::spawn(move || {
        loop {
            let buf = b"test";
            if let Err(e) = pipe_tx.try_io(|| {
                let buf_ptr = &buf as *const _ as *const _;
                let res = unsafe { libc::write(pipe_tx.as_raw_fd(), buf_ptr, buf.len()) };
                if res != -1 {
                    Ok(res as usize)
                } else {
                    Err(io::Error::last_os_error())
                }
            }) {
                eprintln!("{e}");
            }
            thread::sleep(Duration::from_millis(10));
        }
    });

    let mut events = Events::with_capacity(10);

    loop {
        poll.poll(&mut events, None)?;
        for event in &events {
            match event.token() {
                UDP => {
                    if event.is_readable() {
                        handle_udp_rx(&socket);
                        println!("udp RX {:?}", event);
                    }
                    if event.is_writable() {
                        println!("udp TX {:?}", event);
                    }
                }
                PRODUCER => {
                    handle_producer(&pipe_rx, &socket);
                }
                _ => panic!()
            }
        }
        events.clear();
    }
}

fn handle_udp_rx(socket: &UdpSocket) {
    loop {
        let mut buf = vec![0u8; 4];
        let (count, addr) = match socket.recv_from(&mut buf) {
            Ok((count, addr)) => (count, addr),
            Err(e) if e.kind() == ErrorKind::WouldBlock => return,
            Err(e) => {
                eprintln!("{}", e);
                continue
            }
        };
        println!("{addr} {:?} ({count})", buf);
    }
}

fn handle_producer(pipe_rx: &Receiver, socket: &UdpSocket) {
    let mut buf = [0u8; 4];
    if let Err(e) = pipe_rx.try_io(|| {
        let buf_ptr = &mut buf as *mut _ as *mut _;
        let res = unsafe { libc::read(pipe_rx.as_raw_fd(), buf_ptr, buf.len()) };
        if res != -1 {
            Ok(res as usize)
        } else {
            Err(io::Error::last_os_error())
        }
    }) {
        println!("{e}");
    }
    if let Err(e) = socket.send_to(&buf, "192.168.178.189:4242".parse::<SocketAddr>().unwrap()) {
    //     eprintln!("{e}");
        eprint!("e");
    } else {
        eprint!(".");
    }
    std::io::stderr().flush().unwrap();
    if let Err(e) = socket.send_to(&buf, "192.168.178.172:4242".parse::<SocketAddr>().unwrap()) {
    //     eprintln!("{e}");
            eprint!("e");
    } else {
        eprint!(".");
    }
    std::io::stderr().flush().unwrap();
    if let Err(e) = socket.send_to(&buf, "192.168.178.125:4242".parse::<SocketAddr>().unwrap()) {
        // eprintln!("{e}");
        eprint!("e");
    } else {
        eprint!(".");
    }
    std::io::stderr().flush().unwrap();
}

It results in an output that looks like this:

..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee

I kind of expected to receive a tx event everytime I could write to the socket (which should be after each .)

The printed es correspond to WouldBlock errors (errno 11).
So I assume that has to do with the level triggered event type?

Now my real problem is the fact that sometimes it takes ~3 seconds until a new sendto() is possible.
Do you happen to know why that might be?
I read somewhere it might have to do with arp requests blocking the send...

So this likely may not be related to mio at all.

[Thomasdezeeuw]

Thank you!

That makes sense.

As for the TX events: The following is what my usecase boils down to:

# Cargo.toml
[package]
name = "udp_test"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
libc = "0.2.148"
mio = { version = "0.8", features = ["os-ext", "os-poll", "net"] }

use std::{io::{self, ErrorKind, Write}, net::SocketAddr, thread, os::fd::AsRawFd, time::Duration};

use mio::{Token, Poll, unix::pipe::Receiver, Events, Interest, net::UdpSocket};

const UDP: Token = Token(0);
const PRODUCER: Token = Token(1);

fn main() -> io::Result<()> {

    // register event sources
    let port = 4242u16;
    let listen_addr = SocketAddr::new("0.0.0.0".parse().unwrap(), port);
    let mut socket = UdpSocket::bind(listen_addr)?;
    let (pipe_tx, mut pipe_rx) = mio::unix::pipe::new()?;
    pipe_rx.set_nonblocking(true)?;

Mio's unix pipes are already non-blocking, so this is not needed.

let mut poll = Poll::new()?;

poll.registry().register(&mut socket, UDP, Interest::READABLE | Interest::WRITABLE)?;
poll.registry().register(&mut pipe_rx, PRODUCER, Interest::READABLE)?;

// producer thread writing stuff onto the pipe in intervals
thread::spawn(move || {
    loop {
        let buf = b"test";
        if let Err(e) = pipe_tx.try_io(|| {
            let buf_ptr = &buf as *const _ as *const _;
            let res = unsafe { libc::write(pipe_tx.as_raw_fd(), buf_ptr, buf.len()) };
            if res != -1 {
                Ok(res as usize)
            } else {
                Err(io::Error::last_os_error())
            }

Why not pipe_tx.write (from the std::io::Write trait)?

        }) {
            eprintln!("{e}");
        }
        thread::sleep(Duration::from_millis(10));
    }
});

let mut events = Events::with_capacity(10);

loop {
    poll.poll(&mut events, None)?;
    for event in &events {
        match event.token() {
            UDP => {
                if event.is_readable() {
                    handle_udp_rx(&socket);
                    println!("udp RX {:?}", event);
                }
                if event.is_writable() {
                    println!("udp TX {:?}", event);
                }
            }
            PRODUCER => {
                handle_producer(&pipe_rx, &socket);
            }
            _ => panic!()
        }
    }
    events.clear();
}

}

fn handle_udp_rx(socket: &UdpSocket) {
loop {
let mut buf = vec![0u8; 4];
let (count, addr) = match socket.recv_from(&mut buf) {
Ok((count, addr)) => (count, addr),
Err(e) if e.kind() == ErrorKind::WouldBlock => return,
Err(e) => {
eprintln!("{}", e);
continue
}
};
println!("{addr} {:?} ({count})", buf);

This should be buf[n..] as the bytes after n aren't written.

}

}

fn handle_producer(pipe_rx: &Receiver, socket: &UdpSocket) {
let mut buf = [0u8; 4];
if let Err(e) = pipe_rx.try_io(|| {
let buf_ptr = &mut buf as *mut _ as *mut _;
let res = unsafe { libc::read(pipe_rx.as_raw_fd(), buf_ptr, buf.len()) };
if res != -1 {
Ok(res as usize)
} else {
Err(io::Error::last_os_error())
}

Can do pipe_rx.read(&mut buf) here (from the std::io::Read trait).

}) {
    println!("{e}");
}
if let Err(e) = socket.send_to(&buf, "192.168.178.189:4242".parse::<SocketAddr>().unwrap()) {

Are you sure this IP works? Why not use 127.0.0.1? That is localhost and should always work.

//     eprintln!("{e}");
    eprint!("e");
} else {
    eprint!(".");
}
std::io::stderr().flush().unwrap();
if let Err(e) = socket.send_to(&buf, "192.168.178.172:4242".parse::<SocketAddr>().unwrap()) {
//     eprintln!("{e}");
        eprint!("e");
} else {
    eprint!(".");
}
std::io::stderr().flush().unwrap();
if let Err(e) = socket.send_to(&buf, "192.168.178.125:4242".parse::<SocketAddr>().unwrap()) {
    // eprintln!("{e}");
    eprint!("e");
} else {
    eprint!(".");
}
std::io::stderr().flush().unwrap();

}

It results in an output that looks like this:

..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeudp TX Event { token: Token(0), readable: false, writable: true, error: false, read_closed: false, write_closed: false, priority: false, aio: false, lio: false }
..............................................................................................................................................................................................................................................eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee

I kind of expected to receive a tx event everytime I could write to the socket (which should be after each `.`)

The printed `e`s correspond to WouldBlock errors (errno 11). So I assume that has to do with the level triggered event type?

Now my real problem is the fact that sometimes it takes ~3 seconds until a new sendto() is possible. Do you happen to know why that might be? I read somewhere it might have to do with arp requests blocking the send...

That can be because of a range of reasons. Running anti-virus/malware software by any chance?

So this likely may not be related to mio at all.

It would recommend enabling debug logging for Mio, it will show all events that it processes.

[feschber]

Thank you for the suggestion, I just forgot about the Read and Write traits ... (kind of new to rust).

I'm on Linux and not using any AV software and also tried this with the firewall disabled.

The ips correspond to different interfaces on my laptop, with only one of them reachable.
I wanted to ping them and see which one responds.

What's strange is the fact, that all of this works, as long as I dont send to more than one of the ips.

(I can send packages to either of the ips indefinitely without getting WouldBlock)

however if I send to multiple interfaces these 3 second delays occur.

[Thomasdezeeuw]

What's strange is the fact, that all of this works, as long as I dont send to more than one of the ips.

(I can send packages to either of the ips indefinitely without getting WouldBlock)

however if I send to multiple interfaces these 3 second delays occur.

I don't have a good explanation for this either.