How to use tokio's UdpSocket to handle messages in a 1 server: N clients setup?

Question

What I want to do:

... write a (1) server/ (N) clients (network-game-)architecture that uses UDP sockets as underlying base for communication.

• Messages are sent as Vec<u8>, encoded via bincode (crate)

• I also want to be able to occasionally send datagrams that can exceed the typical max MTU of ~1500 bytes and be correctly assembled on receiver end, including sending of ack-messages etc. (I assume I'll have to implement that myself, right?)

For the UdpSocket I thought about using tokio's implementation and maybe framed. I am not sure whether this is a good choice though, as it seems that this would introduce an unnecessary step of mapping Vec<u8> (serialized by bincode) to Vec<u8> (needed by UdpCodec of tokio) (?)

Consider this minimal code-example:

Cargo.toml (server)

bincode = "1.0"
futures = "0.1"
tokio-core = "^0.1"

(Serde and serde-derive are used in shared crate where the protocol is defined!)

(I want to replace tokio-core with tokio asap)

fn main() -> () {
    let addr = format!("127.0.0.1:{port}", port = 8080);
    let addr = addr.parse::<SocketAddr>().expect(&format!("Couldn't create valid SocketAddress out of {}", addr));

    let mut core = Core::new().unwrap();
    let handle = core.handle();
    let socket = UdpSocket::bind(&addr, &handle).expect(&format!("Couldn't bind socket to address {}", addr));


    let udp_future = socket.framed(MyCodec {}).for_each(|(addr, data)| {
        socket.send_to(&data, &addr); // Just echo back the data
        Ok(())
    });

    core.run(udp_future).unwrap();
}

struct MyCodec;

impl UdpCodec for MyCodec {
    type In = (SocketAddr, Vec<u8>);
    type Out = (SocketAddr, Vec<u8>);

    fn decode(&mut self, src: &SocketAddr, buf: &[u8]) -> io::Result<Self::In> {
        Ok((*src, buf.to_vec()))
    }

    fn encode(&mut self, msg: Self::Out, buf: &mut Vec<u8>) -> SocketAddr {
        let (addr, mut data) = msg;
        buf.append(&mut data);
        addr
    }
}

The problem here is:

let udp_future = socket.framed(MyCodec {}).for_each(|(addr, data)| { | ------ value moved here ^^^^^^^^^^^^^^ value captured here after move | = note: move occurs because socket has type tokio_core::net::UdpSocket, which does not implement the Copy trait

The error makes total sense, yet I am not sure how I would create such a simple echo-service. In reality, the handling of a message involves a bit more logic ofc, but for the sake of a minimal example, this should be enough to give a rough idea.

My workaround is an ugly hack: creating a second socket.

Answer 1

Here's the signature of UdpSocket::framed from Tokio's documentation:

pub fn framed<C: UdpCodec>(self, codec: C) -> UdpFramed<C>

Note that it takes self, not &self; that is, calling this function consumes the socket. The UdpFramed wrapper owns the underlying socket when you call this. Your compilation error is telling you that you're moving socket when you call this method, but you're also trying to borrow socket inside your closure (to call send_to).

This probably isn't what you want for real code. The whole point of using framed() is to turn your socket into something higher-level, so you can send your codec's items directly instead of having to assemble datagrams. Using send or send_to directly on the socket will probably break the framing of your message protocol. In this code, where you're trying to implement a simple echo server, you don't need to use framed at all. But if you do want to have your cake and eat it and use both framed and send_to, luckily UdpFramed still allows you to borrow the underlying UdpSocket, using get_ref. You can fix your problem this way:

let framed = {
    let socket = UdpSocket::bind(&addr, &handle).expect(&format!("Couldn't bind socket to address {}", addr));
    socket.framed(MyCodec {})
}

let udp_future = framed.for_each(|(addr, data)| {
    info!(self.logger, "Udp packet received from {}: length: {}", addr, data.len());
    framed.get_ref().send_to(&data, &addr); // Just echo back the data

    Ok(())
});

I haven't checked this code, since (as Shepmaster rightly pointed out) your code snippet has other problems, but it should give you the idea anyway. I'll repeat my warning from earlier: if you do this in real code, it will break the network protocol you're using. get_ref's documentation puts it like this:

Note that care should be taken to not tamper with the underlying stream of data coming in as it may corrupt the stream of frames otherwise being worked with.

---

To answer the new part of your question: yes, you need to handle reassembly yourself, which means your codec does actually need to do some framing on the bytes you're sending. Typically this might involve a start sequence which cannot occur in the Vec<u8>. The start sequence lets you recognise the start of the next message after a packet was lost (which happens a lot with UDP). If there's no byte sequence that can't occur in the Vec<u8>, you need to escape it when it does occur. You might then send the length of the message, followed by the data itself; or just the data, followed by an end sequence and a checksum so you know none was lost. There are pros and cons to these designs, and it's a big topic in itself.

You also need your UdpCodec to contain data: a map from SocketAddr to the partially-reassembled message that's currently in progress. In decode, if you are given the start of a message, copy it into the map and return Ok. If you are given the middle of a message, and you already have the start of a message in the map (for that SocketAddr), append the buffer to the existing buffer and return Ok. When you get to the end of the message, return the whole thing and empty the buffer. The methods on UdpCodec take &mut self in order to enable this use case. (NB In theory, you should also deal with packets arriving out of order, but that's actually quite rare in the real world.)

encode is a lot simpler: you just need to add the same framing and copy the message into the buffer.

Let me reiterate here that you don't need to and shouldn't use the underlying socket after calling framed() on it. UdpFramed is both a source and a sink, so you use that one object to send the replies as well. You can even use split() to get separate Stream and Sink implementations out of it, if that makes the ownership easier in your application.

---

Overall, now I've seen how much of the problem you're struggling with, I'd recommend just using several TCP sockets instead of UDP. If you want a connection-oriented, reliable protocol, TCP already exists and does that for you. It's very easy to spend a lot of time making a "reliable" layer on top of UDP that is both slower and less reliable than TCP.