panic after few incoming messages

[tekjar]

thread 'main' panicked at 'assertion failed: end <= self.cap', /Users/raviteja/.cargo/registry/src/github.com-1ecc6299db9ec823/bytes-0.4.2/src/bytes.rs:1607
stack backtrace:
   1:        0x107c204fc - std::sys::imp::backtrace::tracing::imp::write::h21ca2762819c7ae8
   2:        0x107c225ae - std::panicking::default_hook::{{closure}}::h38f99a37d00bb19b
   3:        0x107c22250 - std::panicking::default_hook::ha2186ee24b50729c
   4:        0x107c22a67 - std::panicking::rust_panic_with_hook::h979db19ee91d2a53
   5:        0x107be73d3 - std::panicking::begin_panic::h55f38a455bc31745
   6:        0x107bea488 - bytes::bytes::Inner::set_end::h82befb8e78b27d8f
   7:        0x107bea0c7 - bytes::bytes::Inner::split_to::hdf30edda54ab41e6
   8:        0x107be9c3b - bytes::bytes::BytesMut::split_to::h58c00b1381d01291
   9:        0x107af9bf1 - <rumqttd::codec::MqttCodec as tokio_io::framed_read::Decoder>::decode::hc5c6fc4b82691933
  10:        0x107af45e2 - <tokio_io::framed::Fuse<T, U> as tokio_io::framed_read::Decoder>::decode::hf27d43a9a8b23015
  11:        0x107af6c8b - <tokio_io::framed_write::FramedWrite2<T> as tokio_io::framed_read::Decoder>::decode::h8e63524e3ad209a5
  12:        0x107af1543 - <tokio_io::framed_read::FramedRead2<T> as futures::stream::Stream>::poll::had597800fa87b314
  13:        0x107af055b - <tokio_io::framed::Framed<T, U> as futures::stream::Stream>::poll::h0add0986ba64e75f
  14:        0x107af1fb2 - <futures::stream::split::SplitStream<S> as futures::stream::Stream>::poll::h3fd48e3dce02280d
  15:        0x107a6b8f7 - <futures::stream::for_each::ForEach<S, F, U> as futures::future::Future>::poll::hfe201db90d0693d7
  16:        0x107ad61ed - <futures::future::chain::Chain<A, B, C>>::poll::hb70001641cce7c49
  17:        0x107af813c - <futures::future::then::Then<A, B, F> as futures::future::Future>::poll::h96d61bd1db456977
  18:        0x107b2d3c1 - <alloc::boxed::Box<F> as futures::future::Future>::poll::ha15f7165eb975205
  19:        0x107b17f8c - <futures::task_impl::Spawn<F>>::poll_future::{{closure}}::h7fa62e61d191c22c
  20:        0x107b1847e - <futures::task_impl::Spawn<T>>::enter::{{closure}}::he00427d7d77b3a1b
  21:        0x107b2f77a - futures::task_impl::set::{{closure}}::h9e1b772b2c99b467
  22:        0x107b1a211 - <std::thread::local::LocalKey<T>>::with::hf1203554760ba1fd
  23:        0x107b2f633 - futures::task_impl::set::h261628f58d78254f
  24:        0x107b18332 - <futures::task_impl::Spawn<T>>::enter::hf7dd52e83ef51739
  25:        0x107b17f0f - <futures::task_impl::Spawn<F>>::poll_future::h0f85f03fa28897d6
  26:        0x107b3cab6 - tokio_core::reactor::Core::dispatch_task::{{closure}}::h0acc6ca4c4f4174b
  27:        0x107b141b4 - <scoped_tls::ScopedKey<T>>::set::h7f15a903853429bc
  28:        0x107b3c4c5 - tokio_core::reactor::Core::dispatch_task::hc3fd8d6531a32c3c
  29:        0x107b3b765 - tokio_core::reactor::Core::dispatch::ha135ccae5690bfb4
  30:        0x107b3b2d9 - tokio_core::reactor::Core::poll::h76e14e964798dba8
  31:        0x107a6c69e - tokio_core::reactor::Core::run::hdb4568c67ea0558d
  32:        0x107aff980 - rumqttd::main::h380ac17aef49e4e9
  33:        0x107c239da - __rust_maybe_catch_panic
  34:        0x107c22e36 - std::rt::lang_start::hfc9882558f9403bf
  35:        0x107b02609 - main

here is my decode for reference

fn decode(&mut self, buf: &mut BytesMut) -> io::Result<Option<Packet>> {
        // NOTE: `decode` might be called with `buf.len == 0` when prevous
        // decode call read all the bytes in the stream. We should return
        // Ok(None) in those cases or else the `read` call will return
        // Ok(0) => translated to UnexpectedEOF by `byteorder` crate.
        // `read` call Ok(0) happens when buffer specified was 0 bytes in len
        // https://doc.rust-lang.org/std/io/trait.Read.html#tymethod.read
        if buf.len() == 0 {
            return Ok(None);
        }

        let (packet, len) = {
            let mut buf_ref = buf.as_ref();
            match buf_ref.read_packet_with_len() {
                Err(e) => {
                    if let mqtt3::Error::Io(e) = e {
                        match e.kind() {
                            ErrorKind::TimedOut | ErrorKind::WouldBlock => return Ok(None),
                            _ => return Err(io::Error::new(e.kind(), e.description())),
                        }
                    } else {
                        return Err(io::Error::new(ErrorKind::Other, e.description()));
                    }
                }
                Ok(v) => v,
            }
        };

        buf.split_to(len);

        // println!("{:?}, {:?}, {:?}", len, packet, buf.len());
        Ok(Some(packet))
}

[carllerche]

Thanks for the report. This is probably due to passing an invalid index to a function. ... possibly the split_to fn. I will close this for now. If you feel like there is a bug, please re-open... and a failing unit test would be helpful :)

[tekjar]

@carllerche Yeah this is at split_to fun. Each incoming packet is 24 bytes. buf.len() in 24 every time except before panic. Is it possible that buf doesn't have all 24 bytes yet but decode is invoked?

[carllerche]

Yes, decode is called when there is any number of bytes in the buffer.

[tekjar]

Ohh. Then I suppose I should return Ok(None) when buf.len() < len necessary to frame the packet

[carllerche]

Yep, If you return `Ok(None)`, decode will be called again when there is more data.

…

On Sun, Apr 16, 2017 at 1:05 PM RT ***@***.***> wrote: Ohh. Then I suppose I should return Ok(None) when buf.len() < len necessary to frame the packet — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#104 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAAYJOS50WLlSNvdGFSfTIU7ku4GG75Zks5rwnRygaJpZM4M-s8s> .

[tekjar]

Thanks for the help @carllerche :)