feat(codec): Introduce Decoder/Encoder traits

[alce]

In preparation for experimenting with a different decoding strategy, this PR swaps tokio's Decoder and Encoder traits with our own.

[LucioFranco]

@alce friendly ping, anything I can help here?

[alce]

Hi, I’ll resume working on this early next week.

…

On 4 Jan 2020, at 13:24, Lucio Franco ***@***.***> wrote:  @alce friendly ping, anything I can help here? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub, or unsubscribe.

[LucioFranco]

So I brought this branch inline with latest master and ran benchmarks:

 name              old ns/iter        new ns/iter        diff ns/iter  diff %  speedup
 chunk_size_100    967 (1039 MB/s)    1,169 (859 MB/s)            202  20.89%   x 0.83
 chunk_size_1005   368 (2730 MB/s)    394 (2550 MB/s)              26   7.07%   x 0.93
 chunk_size_500    481 (2089 MB/s)    532 (1889 MB/s)              51  10.60%   x 0.90
 message_count_1   364 (1387 MB/s)    393 (1284 MB/s)              29   7.97%   x 0.93
 message_count_10  1,551 (3255 MB/s)  1,860 (2715 MB/s)           309  19.92%   x 0.83
 message_count_20  2,882 (3504 MB/s)  3,436 (2939 MB/s)           554  19.22%   x 0.84
 message_size_10k  2,140 (9350 MB/s)  2,129 (9398 MB/s)           -11  -0.51%   x 1.01
 message_size_1k   512 (3925 MB/s)    543 (3701 MB/s)              31   6.05%   x 0.94
 message_size_5k   1,140 (8780 MB/s)  1,206 (8300 MB/s)            66   5.79%   x 0.95

Old is #6673f9 and new is this branch with a merge of that commit.

It looks like it may not be worth it or we should spend some more time on this.

[LucioFranco]

Perfect! Thanks!

[magnet]

I haven't dug into this much, but while updating to tonic 0.1 this has bitten ne as it seems this is preventing me from doing zero-copy handling of Bytes like I used to do by hiding the Bytes/BytesMut type. I previously used tokio's BytesCodec and passed buffers (containing a flatbuffers payload) to tonic in a zero-copy manner. Has this ability been lost?

[magnet]

Here's the raw byte codec I was using on tonic beta. I was mostly delegating to BytesCodec. The encoder still allocated with reserve() and then copied the buffer I want to send. That's something I can do with the new design.

However I had the decoder do zero-copy splitting of BytesMut -- e.g the byte buffer returned by tonic was reused in my application code, and that's something that is not possible with the new DecodeBuf trait because the internal &mut BytesMut reference is hidden, and the split_to method hidden (and not available on bytes::Buf).

Previously:

#[derive(Clone, Copy, Default, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub(crate) struct FlatbuffersCodec(());

impl FlatbuffersCodec {
    pub(crate) const fn new() -> Self {
        FlatbuffersCodec(())
    }
}

impl Codec for FlatbuffersCodec {
    type Encode = Bytes;
    type Decode = BytesMut;

    type Encoder = Self;
    type Decoder = Self;

    fn encoder(&mut self) -> Self::Encoder {
        FlatbuffersCodec::new()
    }

    fn decoder(&mut self) -> Self::Decoder {
        FlatbuffersCodec::new()
    }
}

impl Encoder for FlatbuffersCodec {
    type Item = Bytes;
    type Error = Status;

    fn encode(&mut self, item: Self::Item, buf: &mut BytesMut) -> Result<(), Self::Error> {
        BytesCodec::new().encode(item, buf).map_err(from_io_error)
    }
}

impl Decoder for FlatbuffersCodec {
    type Item = BytesMut;
    type Error = Status;

    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
        BytesCodec::new().decode(buf).map_err(from_io_error)
    }
}

fn from_io_error(error: std::io::Error) -> Status {
    // Map Protobuf parse errors to an INTERNAL status code, as per
    // https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
    Status::new(Code::Internal, error.to_string())
}

With the new trait design:

#[derive(Clone, Copy, Default, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
pub(crate) struct FlatbuffersCodec(());

impl FlatbuffersCodec {
    pub(crate) const fn new() -> Self {
        FlatbuffersCodec(())
    }
}

impl Codec for FlatbuffersCodec {
    type Encode = Bytes;
    type Decode = BytesMut;

    type Encoder = Self;
    type Decoder = Self;

    fn encoder(&mut self) -> Self::Encoder {
        FlatbuffersCodec::new()
    }

    fn decoder(&mut self) -> Self::Decoder {
        FlatbuffersCodec::new()
    }
}

impl Encoder for FlatbuffersCodec {
    type Item = Bytes;
    type Error = Status;

    fn encode(&mut self, item: Self::Item, buf: &mut EncodeBuf<'_>) -> Result<(), Self::Error> {
        // this is the same as before
        buf.reserve(item.len());
        buf.put(item);
        Ok(())
    }
}

impl Decoder for FlatbuffersCodec {
    type Item = BytesMut;
    type Error = Status;

    fn decode(&mut self, buf: &mut DecodeBuf<'_>) -> Result<Option<Self::Item>, Self::Error> {
        if buf.remaining() > 0 {
            let mut res = BytesMut::with_capacity(buf.remaining()); // This is now an extra alloc
            res.put_slice(buf.bytes());
            Ok(Some(res))
        } else {
            Ok(None)
        }
    }
}

[davedbase]

@magnet did you find a solution to this?

[magnet]

I haven't, but I have been busy with other stuff since.

[davedbase]

Any chance you are going to continue with it @magnet? This is a pretty awesome addition. I'm assuming that the new trait design totally prohibits you though. Would Tonic need changes to make this possible?

[magnet]

Not in the short term, no, maybe I'll pick this up sometimes next year.