Add log statements to libp2p-mplex. (#179)

* Add log statements to libp2p-mplex.

Addresses issue #126.

* Run `cargo fmt`.

* Wording.

* Adjust log levels.

mplex/Cargo.toml

@@ -4,19 +4,20 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
-bytes = "0.4.5"
-num-traits = "0.1.40"
-num-bigint = { version = "0.1.40", default-features = false }
-tokio-io = "0.1"
-futures = "0.1"
-parking_lot = "0.4.8"
 arrayvec = "0.4.6"
-rand = "0.3.17"
+bytes = "0.4.5"
 circular-buffer = { path = "../circular-buffer" }
-libp2p-swarm = { path = "../swarm" }
-varint = { path = "../varint-rs" }
 error-chain = "0.11.0"
+futures = "0.1"
 futures-mutex = { git = "https://github.com/paritytech/futures-mutex" }
+libp2p-swarm = { path = "../swarm" }
+log = "0.4"
+num-bigint = { version = "0.1.40", default-features = false }
+num-traits = "0.1.40"
+parking_lot = "0.4.8"
+rand = "0.3.17"
+tokio-io = "0.1"
+varint = { path = "../varint-rs" }
 
 [dev-dependencies]
 libp2p-tcp-transport = { path = "../tcp-transport" }

mplex/src/lib.rs

@@ -26,6 +26,8 @@ extern crate error_chain;
 extern crate futures;
 extern crate futures_mutex;
 extern crate libp2p_swarm as swarm;
+#[macro_use]
+extern crate log;
 extern crate num_bigint;
 extern crate num_traits;
 extern crate parking_lot;

mplex/src/read.rs

@@ -30,7 +30,17 @@ use circular_buffer::Array;
 pub enum NextMultiplexState {
     NewStream(u32),
     ParsingMessageBody(u32),
-    Ignore,
+    Ignore(u32),
+}
+
+impl NextMultiplexState {
+    pub fn substream_id(&self) -> u32 {
+        match *self {
+            NextMultiplexState::NewStream(id)
+            | NextMultiplexState::ParsingMessageBody(id)
+            | NextMultiplexState::Ignore(id) => id,
+        }
+    }
 }
 
 pub enum MultiplexReadState {
@@ -51,6 +61,7 @@ pub enum MultiplexReadState {
         remaining_bytes: usize,
     },
     Ignore {
+        substream_id: u32,
         remaining_bytes: usize,
     },
 }
@@ -231,6 +242,7 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                             substream_id,
                             packet_type,
                         } = MultiplexHeader::parse(header).map_err(|err| {
+                            debug!(target: "libp2p-mplex", "failed to parse header: {}", err);
                             io::Error::new(
                                 io::ErrorKind::Other,
                                 format!("Error parsing header: {:?}", err),
@@ -254,7 +266,7 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                             PacketType::Close(_) | PacketType::Reset(_) => {
                                 lock.read_state = Some(BodyLength {
                                     state: Default::default(),
-                                    next: NextMultiplexState::Ignore,
+                                    next: NextMultiplexState::Ignore(substream_id),
                                 });
 
                                 lock.close_stream(substream_id);
@@ -269,8 +281,10 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                     }
                     Err(error) => {
                         return if let varint::Error(varint::ErrorKind::Io(inner), ..) = error {
+                            debug!(target: "libp2p-mplex", "failed to read header: {}", inner);
                             Err(inner)
                         } else {
+                            debug!(target: "libp2p-mplex", "failed to read header: {}", error);
                             Err(io::Error::new(io::ErrorKind::Other, error.description()))
                         };
                     }
@@ -282,10 +296,15 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
             } => {
                 use self::NextMultiplexState::*;
 
-                match varint_state
+                let body_len = varint_state.read(&mut lock.stream).map_err(|e| {
-                    .read(&mut lock.stream)
+                    debug!(target: "libp2p-mplex",
-                    .map_err(|_| io::Error::new(io::ErrorKind::Other, "Error reading varint"))?
+                           "substream {}: failed to read body length: {}",
-                {
+                           next.substream_id(),
+                           e);
+                    io::Error::new(io::ErrorKind::Other, "Error reading varint")
+                })?;
+
+                match body_len {
                     Async::Ready(length) => {
                         // TODO: Limit `length` to prevent resource-exhaustion DOS
                         let length = if let Some(length) = length {
@@ -295,7 +314,8 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                         };
 
                         lock.read_state = match next {
-                            Ignore => Some(MultiplexReadState::Ignore {
+                            Ignore(substream_id) => Some(MultiplexReadState::Ignore {
+                                substream_id,
                                 remaining_bytes: length,
                             }),
                             NewStream(substream_id) => {
@@ -325,6 +345,7 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                                     })
                                 } else {
                                     Some(MultiplexReadState::Ignore {
+                                        substream_id,
                                         remaining_bytes: length,
                                     })
                                 }
@@ -374,12 +395,15 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                             return on_block;
                         }
                         Err(other) => {
+                            debug!(target: "libp2p-mplex",
+                                   "substream {}: failed to read new stream: {}",
+                                   substream_id,
+                                   other);
                             lock.read_state = Some(NewStream {
                                 substream_id,
                                 name,
                                 remaining_bytes,
                             });
-
                             return Err(other);
                         }
                     }
@@ -435,6 +459,10 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                                 return on_block;
                             }
                             Err(other) => {
+                                debug!(target: "libp2p-mplex",
+                                       "substream {}: failed to read message body: {}",
+                                       substream_id,
+                                       other);
                                 return Err(other);
                             }
                         }
@@ -452,6 +480,7 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                 }
             }
             Ignore {
+                substream_id,
                 mut remaining_bytes,
             } => {
                 let mut ignore_buf: [u8; 256] = [0; 256];
@@ -466,17 +495,26 @@ fn read_stream_internal<T: AsyncRead, Buf: Array<Item = u8>>(
                             Ok(consumed) => {
                                 remaining_bytes -= consumed;
                                 lock.read_state = Some(Ignore {
+                                    substream_id,
                                     remaining_bytes: remaining_bytes,
                                 });
                             }
                             Err(ref err) if err.kind() == io::ErrorKind::WouldBlock => {
-                                lock.read_state = Some(Ignore { remaining_bytes });
+                                lock.read_state = Some(Ignore {
-
+                                    substream_id,
+                                    remaining_bytes,
+                                });
                                 return on_block;
                             }
                             Err(other) => {
-                                lock.read_state = Some(Ignore { remaining_bytes });
+                                debug!(target: "libp2p-mplex",
-
+                                       "substream {}: failed to read ignore bytes: {}",
+                                       substream_id,
+                                       other);
+                                lock.read_state = Some(Ignore {
+                                    substream_id,
+                                    remaining_bytes,
+                                });
                                 return Err(other);
                             }
                         }

mplex/src/shared.rs

@@ -97,6 +97,7 @@ impl<T, Buf: Array> MultiplexShared<T, Buf> {
     }
 
     pub fn open_stream(&mut self, id: u32) -> bool {
+        trace!(target: "libp2p-mplex", "open stream {}", id);
         self.open_streams
             .entry(id)
             .or_insert(SubstreamMetadata::new_open())
@@ -104,6 +105,7 @@ impl<T, Buf: Array> MultiplexShared<T, Buf> {
     }
 
     pub fn close_stream(&mut self, id: u32) {
+        trace!(target: "libp2p-mplex", "close stream {}", id);
         self.open_streams.insert(id, SubstreamMetadata::Closed);
     }
 }

mplex/src/write.rs

@@ -207,6 +207,10 @@ pub fn write_stream<Buf: circular_buffer::Array, T: AsyncWrite>(
                             Err(Some(Body { size }))
                         }
                         Err(other) => {
+                            debug!(target: "libp2p-mplex",
+                                   "substream {}: failed to write body: {}",
+                                   id,
+                                   other);
                             return Err(other);
                         }
                     }