chore(performance): file-source testing and benchmarks (#6742)

* Begin rework of `file-source`

This commit is the start of a process to address #6730. The major changes
introduced in this commit so far are application of clippy suggestions and model
checks of the `read_until_with_max_size` test. I have a pretty good idea of how
that function works now and will be introducing benchmarks.

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Clippy fixes

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Introduce benchmark

In this commit I have introduce a benchmark for the read_until etc
function. To do this I've had to make it part of the public API of the crate,
but since the crate sits inside a larger project I'm less chuffed about this. I
have fiddled with the test layout some as well.

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* end Cargo.toml in newline

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Slim down the feature list slightly, relax bounds

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Reverse 'bytes' upgrade

This crate looks to be challenging to upgrade. Best to do once
tokio is updated in this project.

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* Use libc on Windows

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

* remove unused file

Signed-off-by: Brian L. Troutwine <brian@troutwine.us>

lib/file-source/Cargo.toml

@@ -6,24 +6,75 @@ edition = "2018"
 publish = false
 license = "MIT"
 
+[target.'cfg(windows)'.dependencies]
+libc = "0.2"
+winapi = { version = "0.3", features = ["winioctl"] }
+
 [dependencies]
-bstr = "0.2"
-bytes = "0.5"
-chrono = { version = "0.4.19", features = ["serde"] }
 crc = "1.8.1"
-dashmap = "4.0.2"
-flate2 = "1.0.19"
-futures = { version = "0.3", default-features = false, features = ["executor"] }
 glob = "0.3.0"
-indexmap = {version = "1.6.2", features = ["serde"]}
-libc =  "0.2"
 scan_fmt = "0.2.6"
-serde = { version = "1.0.117", features = ["derive"] }
-serde_json = "1.0.33"
-tokio = { version = "0.2.13", features = ["rt-core", "blocking", "time"] }
-tracing = "0.1.15"
-winapi = { version = "0.3", features = ["winioctl"] }
+
+[dependencies.bstr]
+version = "0.2"
+default-features = false
+features = []
+
+[dependencies.bytes]
+version = "0.5"
+default-features = false
+features = []
+
+[dependencies.chrono]
+version = "0.4"
+default-features = false
+features = ["clock", "serde"]
+
+[dependencies.dashmap]
+version = "4.0"
+default-features = false
+features = []
+
+[dependencies.indexmap]
+version = "1.6"
+default-features = false
+features = ["serde"]
+
+[dependencies.flate2]
+version = "1.0"
+default-features = false
+features = ["rust_backend"]
+
+[dependencies.futures]
+version = "0.3"
+default-features = false
+features = ["executor"]
+
+[dependencies.serde]
+version = "1.0"
+default-features = false
+features = ["derive"]
+
+[dependencies.serde_json]
+version = "1.0"
+default-features = false
+features = []
+
+[dependencies.tracing]
+version = "0.1"
+default-features = false
+features = []
+
+[dependencies.tokio]
+version = "0.2"
+default-features = false
+features = ["rt-core", "blocking", "time"]
 
 [dev-dependencies]
+criterion = "0.3"
 quickcheck = "1"
 tempfile = "3.1.0"
+
+[[bench]]
+name = "buffer"
+harness = false

lib/file-source/benches/buffer.rs

@@ -0,0 +1,76 @@
+use bytes::BytesMut;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+use file_source::buffer::read_until_with_max_size;
+use std::fmt;
+use std::io::Cursor;
+
+struct Parameters {
+    bytes: Vec<u8>,
+    delim_offsets: Vec<usize>,
+    delim: u8,
+    bytes_before_first_delim: usize,
+    max_size: u8,
+}
+
+impl fmt::Display for Parameters {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "bytes_before_first_delim: {}",
+            self.bytes_before_first_delim,
+        )
+    }
+}
+
+fn read_until_bench(c: &mut Criterion) {
+    let mut group = c.benchmark_group("read_until");
+
+    let mut parameters = vec![
+        Parameters {
+            bytes: vec![0; 1024],
+            delim_offsets: vec![100, 500, 502],
+            delim: 1,
+            bytes_before_first_delim: 501,
+            max_size: 1,
+        },
+        Parameters {
+            bytes: vec![0; 1024],
+            delim_offsets: vec![900, 999, 1004, 1021, 1023],
+            delim: 1,
+            bytes_before_first_delim: 1022,
+            max_size: 1,
+        },
+    ];
+
+    for param in &mut parameters {
+        for offset in &param.delim_offsets {
+            param.bytes[*offset] = param.delim;
+        }
+    }
+
+    for param in &parameters {
+        group.throughput(Throughput::Bytes(param.bytes_before_first_delim as u64));
+
+        let mut position = 0;
+        let mut buffer = BytesMut::with_capacity(param.max_size as usize);
+        let mut reader = Cursor::new(&param.bytes);
+        let delimiter: [u8; 1] = [param.delim];
+        group.bench_with_input(BenchmarkId::from_parameter(&param), &param, |b, _| {
+            b.iter(|| {
+                let _ = read_until_with_max_size(
+                    &mut reader,
+                    &mut position,
+                    &delimiter,
+                    &mut buffer,
+                    param.max_size as usize,
+                );
+                reader.set_position(0);
+            })
+        });
+    }
+}
+
+criterion_group!(name = benches;
+                 config = Criterion::default();
+                 targets = read_until_bench);
+criterion_main!(benches);