Read / Write / BufRead Traits

Level 4 — Error Handling & Generics The std::io traits behind all blocking byte-oriented I/O — files, sockets, stdin/stdout.

1. Prerequisites

• Result<T, E> — Every I/O operation can fail, so every method here returns io::Result<T>.
• ? Operator — The idiomatic way to propagate I/O errors.
• Trait Objects (dyn Trait) — Box<dyn Read>/Box<dyn Write> are common ways to abstract over I/O sources.

2. Term Category

Standard Library Traits (the universal I/O interface): Read, Write, and BufRead are the trio of traits that make byte-oriented input/output generic in Rust. A function written against impl Read works identically whether the actual source is a file, a TCP socket, an in-memory Vec<u8>, or stdin — the caller decides what concrete type to plug in.

3. Explanation

(1) Design Motivation — "Why did we design this?"

Every I/O source — files, sockets, in-memory buffers, standard input — fundamentally does the same two things: you can pull bytes out of it, or push bytes into it. Rather than writing separate file-reading code, socket-reading code, and buffer-reading code, Rust abstracts this into two core traits: Read (has a .read(&mut buf) -> io::Result<usize> method) and Write (has .write(&buf) -> io::Result<usize>). Any function written generically over R: Read or W: Write (or the trait-object forms &mut dyn Read) automatically works with any current or future type that implements them — you write the logic once. BufRead extends Read with line-oriented and buffered convenience methods (.read_line(), .lines()), which require an internal buffer that plain Read doesn't guarantee.

(2) Reality Metaphor

Imagine a universal electrical outlet adapter that works in any country.

• Read is a plug shape that says "I can pull power out of any socket that speaks this protocol" — whether the socket is in a wall (a file), a power bank (an in-memory buffer), or a generator (a network stream), the appliance (your code) doesn't need to know or care.
• Write is the same idea, reversed: "I can push power into any receptacle that accepts this plug shape."
• BufRead is an upgraded adapter with a built-in surge protector and a readout screen (internal buffering) that lets you ask higher-level questions like "give me the next full line," instead of managing raw voltage (bytes) yourself.

(3) Rust Code Examples

Short Snippet (Generic Over Any Write)

use std::io::{self, Write};

// This function works with a File, a TcpStream, stdout, or a Vec<u8> — unchanged.
fn log_message(destination: &mut impl Write, msg: &str) -> io::Result<()> {
    writeln!(destination, "[LOG] {msg}")
}

fn main() -> io::Result<()> {
    let mut buffer: Vec<u8> = Vec::new();
    log_message(&mut buffer, "hello from an in-memory buffer")?;

    log_message(&mut io::stdout(), "hello from real stdout")?;

    println!("captured: {}", String::from_utf8_lossy(&buffer));
    Ok(())
}

Fuller Example (Reading Lines with BufRead)

use std::io::{self, BufRead};

fn count_lines(source: impl BufRead) -> io::Result<usize> {
    let mut count = 0;
    for line in source.lines() {
        let _line = line?; // Each line is its own io::Result<String>.
        count += 1;
    }
    Ok(count)
}

fn main() -> io::Result<()> {
    let text = "line one\nline two\nline three";
    let lines = count_lines(text.as_bytes())?; // &[u8] implements Read; wrap for BufRead.
    println!("{lines}"); // 3
    Ok(())
}

4. Common Mistakes & Pitfalls

Mistake 1: Misunderstanding Read Write Bufread Scoping and Lifecycle Rules

The mistake: Assuming Read Write Bufread instances remain valid beyond their declaring scope block or across asynchronous boundaries without explicit lifetime tracking.

Why it's wrong: Rust strictly enforces lexical scope boundaries and non-lexical lifetimes (NLL) at compile time. Accessing dropped values or failing to handle variable drop order results in compiler errors such as E0597 or E0382.

Incorrect:

fn get_ref() -> &str {
    let s = String::from("read_write_bufread_data");
    &s // ❌ Error E0106/E0515: returns a reference to data owned by the current function
}

Fix:

fn get_string() -> String {
    let s = String::from("read_write_bufread_data");
    s // Ownership of the String is transferred directly to the caller
}

Mistake 2: Mutating Read Write Bufread State Without Exclusive Ownership or mut Borrowing

The mistake: Attempting to mutate data associated with Read Write Bufread through an immutable reference &T or without specifying mut in variable declarations.

Why it's wrong: Rust's aliasing XOR mutability rule (&T for shared immutable access, &mut T for exclusive mutable access) prohibits mutating state through shared references unless interior mutability patterns (e.g. RefCell, Mutex) are explicitly used.

Incorrect:

fn update_val(data: &i32) {
    // *data += 1; // ❌ Error E0594: cannot assign to `*data`, which is behind a `&` reference
}

Fix:

fn update_val(data: &mut i32) {
    *data += 1; // Correct: exclusive mutable reference permits mutation
}

Mistake 3: Concurrent Access to Read Write Bufread Across Threads Without Send / Sync Guards

The mistake: Sharing non-thread-safe Read Write Bufread instances across OS threads via std::thread::spawn.

Why it's wrong: Types that do not implement Send or Sync marker traits cannot safely cross thread boundaries. The compiler prevents data races by raising compile errors E0277 (trait Send is not implemented).

Incorrect:

use std::rc::Rc;
use std::thread;

let rc = Rc::new(42);
// thread::spawn(move || { println!("{}", rc); }); // ❌ Error E0277: `Rc` cannot be sent between threads safely

Fix:

use std::sync::Arc;
use std::thread;

let arc = Arc::new(42);
thread::spawn(move || {
    println!("{}", arc); // Correct: `Arc` implements `Send` and `Sync`
});

5. Practice Exercises

Exercise 1: High-Throughput Zero-Copy Stream Log Analyzer (BufRead::fill_buf & BufRead::consume)

Scenario: In high-throughput logging systems, log indexers, or streaming data ingestors, allocating a new String or Vec<u8> for every line (as done by BufRead::lines() or BufRead::read_line()) introduces severe heap allocation overhead and memory fragmentation. Production I/O pipelines bypass heap allocation by directly inspecting the internal ring-buffer of a BufRead implementation via fill_buf() and advancing the buffer cursor via consume().

Task: Implement a generic stream processor ChunkedLogAnalyzer<R> where R: BufRead.

1. Define a LogStats struct containing fields: total_lines: usize, non_empty_lines: usize, error_count: usize, warn_count: usize, info_count: usize, and total_bytes: usize.
2. Implement scan_logs(&mut self) -> io::Result<LogStats> on ChunkedLogAnalyzer<R> using fill_buf() and consume():
   • Search the buffer slice directly for \n line delimiters.
   • For complete lines residing fully inside the buffer slice, inspect bytes zero-copy without allocating heap memory.
   • If a line is split across fill_buf() buffer chunks, accumulate bytes into a transient fallback vector only for that split line, process it upon finding \n, and clear the vector.
   • Handle both \n and \r\n line endings.
   • Categorize lines starting with ERROR or [ERROR], WARN or [WARN], INFO or [INFO].
   • Correctly process any final line at EOF that lacks a trailing \n.
3. Include unit tests demonstrating zero-copy execution with Cursor<&[u8]> and BufReader with small capacity, asserting line metrics, severity counts, and byte counters using explicit test assertions (assert_eq!, assert!, assert_ne!, matches!).

use std::io::{self, BufRead, Cursor};

#[derive(Debug, Default, PartialEq, Eq)]
pub struct LogStats {
    pub total_lines: usize,
    pub non_empty_lines: usize,
    pub error_count: usize,
    pub warn_count: usize,
    pub info_count: usize,
    pub total_bytes: usize,
}

pub struct ChunkedLogAnalyzer<R> {
    reader: R,
}

impl<R: BufRead> ChunkedLogAnalyzer<R> {
    pub fn new(reader: R) -> Self {
        Self { reader }
    }

    pub fn scan_logs(&mut self) -> io::Result<LogStats> {
        let mut stats = LogStats::default();
        let mut partial_line = Vec::new();

        loop {
            let buffer = self.reader.fill_buf()?;
            if buffer.is_empty() {
                // Process final line at EOF if stream ended without trailing '\n'
                if !partial_line.is_empty() {
                    Self::process_line(&partial_line, &mut stats);
                    partial_line.clear();
                }
                break;
            }

            let mut consumed = 0;
            let buf_len = buffer.len();

            while consumed < buf_len {
                let remaining = &buffer[consumed..];
                if let Some(newline_pos) = remaining.iter().position(|&b| b == b'\n') {
                    let line_slice = &remaining[..newline_pos];
                    if partial_line.is_empty() {
                        // Zero-copy path: entire line resides within current buffer slice
                        Self::process_line(line_slice, &mut stats);
                    } else {
                        // Stitched line path: append chunk remainder and process
                        partial_line.extend_from_slice(line_slice);
                        Self::process_line(&partial_line, &mut stats);
                        partial_line.clear();
                    }
                    consumed += newline_pos + 1; // Advance cursor past '\n'
                } else {
                    // No newline found in remaining slice; accumulate into transient buffer
                    partial_line.extend_from_slice(remaining);
                    consumed += remaining.len();
                }
            }

            self.reader.consume(consumed);
        }

        Ok(stats)
    }

    fn process_line(raw_line: &[u8], stats: &mut LogStats) {
        // Strip Windows carriage return '\r' if present
        let line = if raw_line.ends_with(b"\r") {
            &raw_line[..raw_line.len() - 1]
        } else {
            raw_line
        };

        stats.total_lines += 1;
        stats.total_bytes += line.len();

        if line.is_empty() {
            return;
        }

        stats.non_empty_lines += 1;

        if line.starts_with(b"ERROR") || line.starts_with(b"[ERROR]") {
            stats.error_count += 1;
        } else if line.starts_with(b"WARN") || line.starts_with(b"[WARN]") {
            stats.warn_count += 1;
        } else if line.starts_with(b"INFO") || line.starts_with(b"[INFO]") {
            stats.info_count += 1;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::BufReader;

    #[test]
    fn test_zero_copy_log_scanner_basic() {
        let log_data = "INFO: System started\nWARN: High memory usage\nERROR: Out of disk space\n";
        let cursor = Cursor::new(log_data);
        let mut analyzer = ChunkedLogAnalyzer::new(cursor);
        let stats = analyzer.scan_logs().unwrap();

        assert_eq!(stats.total_lines, 3);
        assert_eq!(stats.non_empty_lines, 3);
        assert_eq!(stats.info_count, 1);
        assert_eq!(stats.warn_count, 1);
        assert_eq!(stats.error_count, 1);
        assert_ne!(stats.total_bytes, 0);
    }

    #[test]
    fn test_log_scanner_chunked_and_no_trailing_newline() {
        let log_data = "[INFO] boot\n[WARN] temp high\n[ERROR] crash";
        let cursor = Cursor::new(log_data);
        // Intentionally small 8-byte buffer capacity forces multi-chunk split lines
        let buffered = BufReader::with_capacity(8, cursor);
        let mut analyzer = ChunkedLogAnalyzer::new(buffered);
        let stats = analyzer.scan_logs().unwrap();

        assert_eq!(stats.total_lines, 3);
        assert_eq!(stats.non_empty_lines, 3);
        assert_eq!(stats.info_count, 1);
        assert_eq!(stats.warn_count, 1);
        assert_eq!(stats.error_count, 1);
        assert!(matches!(stats.error_count, 1));
    }

    #[test]
    fn test_log_scanner_empty_and_whitespace() {
        let log_data = "\n\r\nINFO ok\n";
        let cursor = Cursor::new(log_data);
        let mut analyzer = ChunkedLogAnalyzer::new(cursor);
        let stats = analyzer.scan_logs().unwrap();

        assert_eq!(stats.total_lines, 3);
        assert_eq!(stats.non_empty_lines, 1);
        assert_eq!(stats.info_count, 1);
        assert_eq!(stats.warn_count, 0);
    }
}

Exercise 2: Production Multi-Writer Tee & Inline Hashing Engine (std::io::Write)

Scenario: In storage proxies, cloud object storage splitters, or streaming replication engines, data written to a primary destination (e.g. disk file or TCP socket) must simultaneously be mirrored to a secondary stream (e.g. audit log or replica node) while computing an inline integrity digest (such as FNV-1a or CRC32) without storing the entire payload in memory.

Task: Implement a generic dual-writer struct HashingTeeWriter<W1, W2> where W1: Write and W2: Write.

1. Fields: primary: W1, secondary: W2, checksum: u64 (initialized using FNV offset basis 0xcbf29ce484222325).
2. Implement std::io::Write for HashingTeeWriter<W1, W2>:
   • write(&mut self, buf: &[u8]) -> io::Result<usize>: Writes the complete slice to primary via write_all(buf) and then to secondary via write_all(buf). If either write returns an io::Error, return that error immediately. Update the cumulative FNV-1a hash (hash = (hash ^ (byte as u64)).wrapping_mul(0x100000001b3)). Return Ok(buf.len()).
   • flush(&mut self) -> io::Result<()>: Flush both sinks sequentially, propagating errors if either fails.
3. Provide helper methods: new(primary: W1, secondary: W2) -> Self, checksum(&self) -> u64, and into_inner(self) -> (W1, W2).
4. Include unit tests demonstrating tee operations into dual Vec<u8> buffers, verifying byte output equality, hash accumulation across multiple writes, error propagation when a writer fails, using explicit test assertions (assert_eq!, assert!, assert_ne!, matches!).

use std::io::{self, Write};

const FNV_OFFSET_BASIS: u64 = 0xcbf29ce484222325;
const FNV_PRIME: u64 = 0x100000001b3;

pub struct HashingTeeWriter<W1, W2> {
    primary: W1,
    secondary: W2,
    checksum: u64,
}

impl<W1: Write, W2: Write> HashingTeeWriter<W1, W2> {
    pub fn new(primary: W1, secondary: W2) -> Self {
        Self {
            primary,
            secondary,
            checksum: FNV_OFFSET_BASIS,
        }
    }

    pub fn checksum(&self) -> u64 {
        self.checksum
    }

    pub fn into_inner(self) -> (W1, W2) {
        (self.primary, self.secondary)
    }

    fn update_checksum(&mut self, bytes: &[u8]) {
        for &byte in bytes {
            self.checksum ^= byte as u64;
            self.checksum = self.checksum.wrapping_mul(FNV_PRIME);
        }
    }
}

impl<W1: Write, W2: Write> Write for HashingTeeWriter<W1, W2> {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        // Guarantee full delivery to both tee sinks before updating state
        self.primary.write_all(buf)?;
        self.secondary.write_all(buf)?;

        self.update_checksum(buf);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        self.primary.flush()?;
        self.secondary.flush()?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    struct FailingWriter;
    impl Write for FailingWriter {
        fn write(&mut self, _buf: &[u8]) -> io::Result<usize> {
            Err(io::Error::new(io::ErrorKind::WriteZero, "disk full failure"))
        }
        fn flush(&mut self) -> io::Result<()> {
            Err(io::Error::new(io::ErrorKind::Other, "flush failed"))
        }
    }

    #[test]
    fn test_tee_writer_dual_output_and_checksum() {
        let primary_buf = Vec::new();
        let secondary_buf = Vec::new();

        let mut tee = HashingTeeWriter::new(primary_buf, secondary_buf);
        let written = tee.write(b"Hello, Rust I/O!").unwrap();
        tee.flush().unwrap();

        assert_eq!(written, 16);
        assert_ne!(tee.checksum(), FNV_OFFSET_BASIS);

        let initial_checksum = tee.checksum();
        tee.write_all(b" Streaming chunk.").unwrap();
        assert_ne!(tee.checksum(), initial_checksum);

        let (primary, secondary) = tee.into_inner();
        assert_eq!(primary, b"Hello, Rust I/O! Streaming chunk.");
        assert_eq!(secondary, b"Hello, Rust I/O! Streaming chunk.");
    }

    #[test]
    fn test_tee_writer_failure_propagation() {
        let primary_buf = Vec::new();
        let failing = FailingWriter;

        let mut tee = HashingTeeWriter::new(primary_buf, failing);
        let res = tee.write(b"test data");
        assert!(res.is_err());
        assert!(matches!(
            res.unwrap_err().kind(),
            io::ErrorKind::WriteZero
        ));
    }
}

Exercise 3: Stream Transformation Copy Pipeline over Trait Objects (Box<dyn BufRead> & Box<dyn Write>)

Scenario: In cloud streaming microservices and dynamic plugin architectures, input streams and target output sinks are often not known at compile time. Data pipelines receive heterogeneous sources wrapped as dynamic trait objects (Box<dyn BufRead> and Box<dyn Write>), transform payload chunks via stream processors (e.g. byte encryption, format transcoding, or header insertion), and stream output efficiently using fixed stack/heap transfer buffers.

Task: Implement StreamTransformPipeline handling dynamic trait object streams.

1. Define StreamTransformPipeline holding:
   • reader: Box<dyn BufRead>
   • writer: Box<dyn Write>
   • chunk_size: usize
2. Implement methods:
   • pub fn new(reader: Box<dyn BufRead>, writer: Box<dyn Write>, chunk_size: usize) -> Self
   • pub fn process<F>(&mut self, mut transform: F) -> io::Result<usize> where F: FnMut(&[u8]) -> Vec<u8>:
     • Read chunks of up to chunk_size bytes into a intermediate buffer using self.reader.read(&mut buffer).
     • When read returns 0, break (EOF reached).
     • Apply transform to the read slice &buffer[..bytes_read].
     • Write transformed bytes to self.writer.write_all(&transformed).
     • Track total output bytes written and call self.writer.flush() before returning Ok(total_output_bytes).
3. Write unit tests utilizing dynamic dispatch (Box<dyn BufRead> wrapping BufReader<Cursor<&[u8]>> and Box<dyn Write>), verifying chunked processing, byte transformations, custom thread-safe shared sinks, using explicit test assertions (assert_eq!, assert!, assert_ne!, matches!).

use std::io::{self, BufRead, Write};

pub struct StreamTransformPipeline {
    reader: Box<dyn BufRead>,
    writer: Box<dyn Write>,
    chunk_size: usize,
}

impl StreamTransformPipeline {
    pub fn new(reader: Box<dyn BufRead>, writer: Box<dyn Write>, chunk_size: usize) -> Self {
        let chunk_size = if chunk_size == 0 { 8192 } else { chunk_size };
        Self {
            reader,
            writer,
            chunk_size,
        }
    }

    pub fn process<F>(&mut self, mut transform: F) -> io::Result<usize>
    where
        F: FnMut(&[u8]) -> Vec<u8>,
    {
        let mut buffer = vec![0u8; self.chunk_size];
        let mut total_output_bytes = 0;

        loop {
            let bytes_read = self.reader.read(&mut buffer)?;
            if bytes_read == 0 {
                break;
            }

            let transformed = transform(&buffer[..bytes_read]);
            self.writer.write_all(&transformed)?;
            total_output_bytes += transformed.len();
        }

        self.writer.flush()?;
        Ok(total_output_bytes)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::{BufReader, Cursor};
    use std::sync::{Arc, Mutex};

    #[test]
    fn test_pipeline_trait_objects_and_transformation() {
        let input_data = b"abcdefghijklmnopqrstuvwxyz";
        let reader: Box<dyn BufRead> = Box::new(BufReader::new(Cursor::new(input_data)));
        let output_vec: Vec<u8> = Vec::new();
        let writer: Box<dyn Write> = Box::new(output_vec);

        // 5-byte chunk size tests multi-iteration execution
        let mut pipeline = StreamTransformPipeline::new(reader, writer, 5);

        let total_written = pipeline
            .process(|chunk| chunk.iter().map(|b| b.to_ascii_uppercase()).collect())
            .unwrap();

        assert_eq!(total_written, 26);
        assert_ne!(total_written, 0);
    }

    #[test]
    fn test_pipeline_custom_sink_and_assertions() {
        #[derive(Clone)]
        struct SharedSink(Arc<Mutex<Vec<u8>>>);
        impl Write for SharedSink {
            fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
                self.0.lock().unwrap().extend_from_slice(buf);
                Ok(buf.len())
            }
            fn flush(&mut self) -> io::Result<()> {
                Ok(())
            }
        }

        let shared_output = Arc::new(Mutex::new(Vec::new()));
        let sink = SharedSink(shared_output.clone());

        let input_data = b"Hello World";
        let reader: Box<dyn BufRead> = Box::new(BufReader::new(Cursor::new(input_data)));
        let writer: Box<dyn Write> = Box::new(sink);

        let mut pipeline = StreamTransformPipeline::new(reader, writer, 4);
        let res = pipeline.process(|chunk| {
            let mut v = Vec::new();
            for &b in chunk {
                v.push(b);
                v.push(b); // Duplicate every byte
            }
            v
        });

        assert!(res.is_ok());
        assert_eq!(res.unwrap(), 22);

        let output_bytes = shared_output.lock().unwrap().clone();
        assert_eq!(output_bytes, b"HHeelllloo  WWoorrlldd");
        assert!(matches!(output_bytes.len(), 22));
    }
}

6. Related Terms

• ? Operator — The idiomatic propagation tool for the io::Result<T> every method here returns.
• The Rust Standard Library (std) — Read/Write/BufRead are specifically part of the OS-integration layer that only std (not core/alloc) provides.
• Trait Objects (dyn Trait) — Box<dyn Read>/&mut dyn Write are common ways to store a heterogeneous I/O source/sink.
• tokio — The async counterpart (AsyncRead/AsyncWrite) for non-blocking I/O.

7. Key Takeaways

• Read and Write are the universal, generic byte-I/O traits — a single function written against them works with files, sockets, and in-memory buffers alike.
• BufRead extends Read with buffered, line-oriented convenience methods like .lines() and .read_line().
• .read() may perform a short read — use .read_exact() or .read_to_end()/.read_to_string() when you need guaranteed-complete reads.
• Every method returns io::Result<T>, making the ? operator the natural way to write I/O code.