Introducing the BytesLines iterator

$ /bin/time --format "Run time: %e sec, Max RSS: %M KB" grep anomaly < test.txt
Run time: 0.05 sec, Max RSS: 2380 KB

$ valgrind grep "anomaly" < test.txt |& grep "heap usage"
total heap usage: 305 allocs, 265 frees, 146,413 bytes allocated

def logfile_iterator(reader):
    line_number = 0
    while True:
        line = reader.readline()
        if not line:
            break
        line = line.rstrip()
        line_number += 1
        for subline in line.split("\\n"):
            yield (subline, line_number)

/// A struct to hold the state of the iterator.
pub struct BufLines<R: Read> {
    reader: BufReader<R>,
    buffer: String,
    line_number: usize,
}

impl<R: Read> Iterator for BufLines<R> {
    type Item = Result<(String, usize)>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.buffer.is_empty() {
            self.read_line()
        } else {
            Some(Ok(self.get_line()))
        }
    }
}

// Read a new line and call get_line
fn read_line(&mut self) -> Option<Result<(String, usize)>> {
    match self.reader.read_line(&mut self.buffer) {
        Ok(n) if n > 0 => {
            // The read succeeded
            self.buffer = self.buffer.trim_end().to_owned();
            Some(Ok(self.get_line()))
        }
        Ok(_) => None,
        Err(e) => Some(Err(e)),
    }
}

// Return the first sub line found in the buffer.
fn get_line(&mut self) -> (String, usize) {
    let line = if let Some((sub_line, rest)) = self.buffer.split_once("\\n") {
        let sub_line = sub_line.clone();
        self.buffer = rest.to_owned();
        sub_line
    } else {
        self.line_number += 1;
        let line = self.buffer.clone();
        self.buffer.clear();
        line
    };
    (line, self.line_number)
}

impl<R: Read> Iterator for BufLines<R> {
   type Item = Result<&str>;
}

error[E0106]: missing lifetime specifier
  --> readline.rs:17:24
   |
17 |     type Item = Result<&str>;
   |                        ^ expected named lifetime parameter
   |
help: consider introducing a named lifetime parameter
   |
17 |     type Item<'a> = Result<&'a str>;
   |              ^^^^          ^^^

impl<R: Read> Iterator for BufLines<R> {
    type Item<'a> = Result<&'a str>;
}

error[E0658]: generic associated types are unstable
  --> readline.rs:17:5
   |
17 |     type Item<'a> = Result<&'a str>;
   |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   |
   = note: see issue #44265 <https://github.com/rust-lang/rust/issues/44265> for more information

/// The BytesLines struct holds a single buffer
pub struct BytesLines<R: Read> {
    reader: R,
    buf: BytesMut,
    line_count: usize,
}

impl<R: Read> Iterator for BytesLines<R> {
    type Item = Result<(Bytes, usize)>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.buf.is_empty() {
            self.read_slice()
        } else {
            self.get_slice()
        }
    }
}

// Read a new chunk and call get_slice
fn read_slice(&mut self) -> Option<Result<(Bytes, usize)>> {
    let pos = self.buf.len();
    self.buf.resize(pos + CHUNK_SIZE, 0);
    match self.reader.read(&mut self.buf[pos..]) {
        // We read some data.
        Ok(n) if n > 0 => {
            self.buf.truncate(pos + n);
            self.get_slice()
        }

        // We reached the end of the reader, this is the end.
        Ok(_) => None,

        // There was a reading error, we return it.
        Err(e) => Some(Err(e)),
    }
}

// Find the next line in the buffer
fn get_slice(&mut self) -> Option<Result<(Bytes, usize)>> {
    match self.find_next_line() {
        // We haven't found the end of the line, we need more data.
        None => {
            // reserve() will attempt to reclaim space in the buffer.
            self.buf.reserve(CHUNK_SIZE);
            self.read_slice()
        }

        // We found the end of the line, we can return it now.
        Some((pos, sep)) => {
            // split_to() creates a new zero copy reference to the buffer.
            let res = self.buf.split_to(pos).freeze();
            self.buf.advance(sep.len());
            Some(Ok((res, self.line_count)))
        }
    }
}

 ⭩- the buffer starts here.
[                          ]          < the buffer is empty, we read a chunk.
[aaaaaaaaaaaa\nbbbbb\nccccc]          < there is a line separator.
 ╰-----------⮡ next slice
              ⭨
[              bbbbb\nccccc]
               ╰----⮡ next slice
                     ⭨
[                     ccccc]          < the line is incomplete.
      ⭩ we reserve more space and move the left-overs at the begining of the buffer.
[ccccc                           ]    < we read another chunk after the left-overs.
[ccccccc\ndddddddddddddd\neeeeeee]
 ╰------⮡ next slice
         ⭨
[         dddddddddddddd\neeeeeee]
          ╰-------------⮡ next slice
                         ⭨
[                         eeeeeee]    < the line is incomplete.
        ⭩ we reserve more space and move the left-overs at the begining of the buffer.
[eeeeeee                            ] < we read another chunk after the left-overs.
[eeeeeeeee\n                        ] < we reach the end of file.
 ╰--------⮡ the last slice