Merge rust-bitcoin/rust-bitcoin#680: Deprecate `StreamReader`

e860333bf3 Fix typos (Riccardo Casatta)
9189539715 Use BufReader internally in StreamReader to avoid performance regression on existing callers (Riccardo Casatta)
5dfb93df71 Deprecate StreamReader (Riccardo Casatta)
9ca6c75b18 Bench StreamReader (Riccardo Casatta)

Pull request description:

  `StreamReader` performance is extremely poor in case the object decoded is "big enough" for example a full Block.

  In the common case, the buffer is 64k, so to successfully parse a 1MB block 16 decode attempts are made.
  Even if a user increases the buffer size, `read` is not going to necessarily fill the buffer, as stated in the doc https://doc.rust-lang.org/stable/std/io/trait.Read.html#tymethod.read. In my tests, the reads are 64kB even with a 1MB buffer.

  I think this is the root issue of the performance issue found in electrs in https://github.com/romanz/electrs/issues/547 and they now have decided to decode the TCP stream with their own code in cd0531b8b7 and 05e0221b8e.

  Using directly `consensus_encode` seems to make more sense (taking care of using `BufRead` if necessary) so the `StreamReader` is deprecated

ACKs for top commit:
  Kixunil:
    ACK e860333bf3
  apoelstra:
    ACK e860333bf3

Tree-SHA512: a15a14f3f087be36271da5008d8dfb63866c9ddeb5ceb0e328b4a6d870131132a8b05103f7a3fed231f5bca099865efd07856b4766834d56ce2384b1bcdb889b

examples/handshake.rs

@@ -3,11 +3,10 @@ extern crate bitcoin;
 use std::net::{IpAddr, Ipv4Addr, Shutdown, SocketAddr, TcpStream};
 use std::time::{SystemTime, UNIX_EPOCH};
 use std::{env, process};
-use std::io::Write;
+use std::io::{Write, BufReader};
 
-use bitcoin::consensus::encode;
+use bitcoin::consensus::{encode, Decodable};
 use bitcoin::network::{address, constants, message, message_network};
-use bitcoin::network::stream_reader::StreamReader;
 use bitcoin::secp256k1;
 use bitcoin::secp256k1::rand::Rng;
 
@@ -41,10 +40,10 @@ fn main() {
 
         // Setup StreamReader
         let read_stream = stream.try_clone().unwrap();
-        let mut stream_reader = StreamReader::new(read_stream, None);
+        let mut stream_reader = BufReader::new(read_stream);
         loop {
             // Loop an retrieve new messages
-            let reply: message::RawNetworkMessage = stream_reader.read_next().unwrap();
+            let reply = message::RawNetworkMessage::consensus_decode(&mut stream_reader).unwrap();
             match reply.payload {
                 message::NetworkMessage::Version(_) => {
                     println!("Received version message: {:?}", reply.payload);

src/network/stream_reader.rs

@@ -14,78 +14,53 @@
 
 //! Stream reader.
 //!
-//! This module defines the `StreamReader` struct and its implementation which
+//! Deprecated
-//! is used for parsing an incoming stream into separate `RawNetworkMessage`s,
+//!
-//! handling assembling messages from multiple packets, or dealing with partial
+//! This module defines `StreamReader` struct and its implementation which is used
-//! or multiple messages in the stream (e.g. when reading from a TCP socket).
+//! for parsing incoming stream into separate `Decodable`s, handling assembling
+//! messages from multiple packets or dealing with partial or multiple messages in the stream
+//! (like can happen with reading from TCP socket)
 //!
 
-use prelude::*;
-
 use core::fmt;
-use io::{self, Read};
+use io::{Read, BufReader};
 
 use consensus::{encode, Decodable};
 
 /// Struct used to configure stream reader function
 pub struct StreamReader<R: Read> {
     /// Stream to read from
-    pub stream: R,
+    pub stream: BufReader<R>,
-    /// I/O buffer
-    data: Vec<u8>,
-    /// Buffer containing unparsed message part
-    unparsed: Vec<u8>
 }
 
 impl<R: Read> fmt::Debug for StreamReader<R> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "StreamReader with buffer_size={} and unparsed content {:?}",
+        write!(f, "StreamReader")
-               self.data.capacity(), self.unparsed)
     }
 }
 
 impl<R: Read> StreamReader<R> {
-    /// Constructs new stream reader for a given input stream `stream` with
+    /// Constructs new stream reader for a given input stream `stream`
-    /// optional parameter `buffer_size` determining reading buffer size
+    #[deprecated(since="0.28.0", note="wrap your stream into a buffered reader if necessary and use consensus_encode directly")]
-    pub fn new(stream: R, buffer_size: Option<usize>) -> StreamReader<R> {
+    pub fn new(stream: R, _buffer_size: Option<usize>) -> StreamReader<R> {
         StreamReader {
-            stream,
+            stream: BufReader::new(stream),
-            data: vec![0u8; buffer_size.unwrap_or(64 * 1024)],
-            unparsed: vec![]
         }
     }
 
-    /// Reads stream and parses next message from its current input,
+    /// Reads stream and parses next message from its current input
-    /// also taking into account previously unparsed partial message (if there was such).
+    #[deprecated(since="0.28.0", note="wrap your stream into a buffered reader if necessary and use consensus_encode directly")]
     pub fn read_next<D: Decodable>(&mut self) -> Result<D, encode::Error> {
-        loop {
+        Decodable::consensus_decode(&mut self.stream)
-            match encode::deserialize_partial::<D>(&self.unparsed) {
-                // In this case we just have an incomplete data, so we need to read more
-                Err(encode::Error::Io(ref err)) if err.kind () == io::ErrorKind::UnexpectedEof => {
-                    let count = self.stream.read(&mut self.data)?;
-                    if count > 0 {
-                        self.unparsed.extend(self.data[0..count].iter());
-                    }
-                    else {
-                        return Err(encode::Error::Io(io::Error::from(io::ErrorKind::UnexpectedEof)));
-                    }
-                },
-                Err(err) => return Err(err),
-                // We have successfully read from the buffer
-                Ok((message, index)) => {
-                    self.unparsed.drain(..index);
-                    return Ok(message)
-                },
-            }
-        }
     }
 }
 
+#[allow(deprecated)]
 #[cfg(test)]
 mod test {
     use std::thread;
     use std::time::Duration;
-    use io::{self, BufReader, Write};
+    use io::{BufReader, Write};
     use std::net::{TcpListener, TcpStream, Shutdown};
     use std::thread::JoinHandle;
     use network::constants::ServiceFlags;
@@ -193,22 +168,6 @@ mod test {
         }
     }
 
-    #[test]
-    fn parse_multipartmsg_test() {
-        let stream = io::empty();
-        let mut reader = StreamReader::new(stream, None);
-        reader.unparsed = MSG_ALERT[..24].to_vec();
-        let message: Result<RawNetworkMessage, _> = reader.read_next();
-        assert!(message.is_err());
-        assert_eq!(reader.unparsed.len(), 24);
-
-        reader.unparsed = MSG_ALERT.to_vec();
-        let message = reader.read_next().unwrap();
-        assert_eq!(reader.unparsed.len(), 0);
-
-        check_alert_msg(&message);
-    }
-
     #[test]
     fn read_singlemsg_test() {
         let stream = MSG_VERSION[..].to_vec();
@@ -346,4 +305,17 @@ mod test {
         // should be also ok for a non-witness block as commitment is optional in that case
         assert!(block.check_witness_commitment());
     }
+
+    #[test]
+    fn parse_multipartmsg_test() {
+        let mut multi = MSG_ALERT.to_vec();
+        multi.extend(&MSG_ALERT[..]);
+        let mut reader = StreamReader::new(&multi[..], None);
+        let message: Result<RawNetworkMessage, _> = reader.read_next();
+        assert!(message.is_ok());
+        check_alert_msg(&message.unwrap());
+        let message: Result<RawNetworkMessage, _> = reader.read_next();
+        assert!(message.is_ok());
+        check_alert_msg(&message.unwrap());
+    }
 }