feat: prototype async API, with demonstrable perf improvements via benchmark

Cargo.toml

@@ -23,23 +23,45 @@ time = { version = "0.3.36", default-features = false }
 
 [dependencies]
 aes = { version = "0.8.4", optional = true }
+async-stream = { version = "0.3.5", optional = true }
 byteorder = "1.5.0"
 bzip2 = { version = "0.4.4", optional = true }
 chrono = { version = "0.4.38", optional = true }
+cfg-if = "1"
 constant_time_eq = { version = "0.3.0", optional = true }
 crc32fast = "1.4.0"
+displaydoc = "0.2.4"
+thiserror = "1.0.48"
 flate2 = { version = "1.0.28", default-features = false, optional = true }
+futures-core = { version = "0.3", optional = true }
+futures-util = { version = "0.3", optional = true }
 hmac = { version = "0.12.1", optional = true, features = ["reset"] }
+indexmap = { version =  "2", features = ["rayon"], optional = true }
+libc = { version = "0.2.148", optional = true }
+num_enum = "0.6.1"
+once_cell = { version = "1.18.0", optional = true }
+parking_lot = { version = "0.12.1", features = ["arc_lock"], optional = true }
 pbkdf2 = { version = "0.12.2", optional = true }
+rayon = { version = "1.8.0", optional = true }
 sha1 = { version = "0.10.6", optional = true }
+static_assertions = { version = "1.1.0", optional = true }
+tempfile = { version = "3.8.0", optional = true }
 time = { workspace = true, optional = true, features = [
     "std",
 ] }
 zstd = { version = "0.13.1", optional = true, default-features = false }
+tokio = { version = "1", features = ["rt", "io-util", "sync", "fs", "macros"], optional = true }
+tokio-pipe = { git = "https://github.com/cosmicexplorer/tokio-pipe", rev = "c44321ae17b4324a8ccaa4f687a8f68259fdca30", optional = true }
+tokio-stream = { version = "0.1.14", optional = true }
 zopfli = { version = "0.8.0", optional = true }
 deflate64 = { version = "0.1.8", optional = true }
 lzma-rs = { version = "0.3.0", default-features = false, optional = true }
 
+[dependencies.memchr2]
+version = "2.6.4"
+optional = true
+package = "memchr"
+
 [target.'cfg(any(all(target_arch = "arm", target_pointer_width = "32"), target_arch = "mips", target_arch = "powerpc"))'.dependencies]
 crossbeam-utils = "0.8.19"
 
@@ -48,10 +70,15 @@ arbitrary = { version = "1.3.2", features = ["derive"] }
 
 [dev-dependencies]
 bencher = "0.1.5"
-getrandom = { version = "0.2.14", features = ["js"] }
-walkdir = "2.5.0"
+criterion = { version = "0.5", features = ["async_tokio"] }
+getrandom = "0.2.14"
+tempfile = "3.8.0"
 time = { workspace = true, features = ["formatting", "macros"] }
-anyhow = "1"
+tokio = { version = "1", features = ["rt", "rt-multi-thread"] }
+tokio-test = "0.4.3"
+uuid = { version = "1.4.1", features = ["v4"] }
+walkdir = "2.5.0"
+
 [features]
 aes-crypto = ["aes", "constant_time_eq", "hmac", "pbkdf2", "sha1"]
 chrono = ["chrono/default"]
@@ -65,6 +92,11 @@ deflate-zlib = ["flate2/zlib", "_deflate-any"]
 deflate-zlib-ng = ["flate2/zlib-ng", "_deflate-any"]
 deflate-zopfli = ["zopfli", "_deflate-any"]
 lzma = ["lzma-rs/stream"]
+tokio-async = [
+  "dep:tokio", "dep:memchr2", "dep:tokio-stream", "dep:tokio-pipe", "dep:parking_lot", "dep:libc", "dep:futures-core",
+  "dep:futures-util", "dep:async-stream", "dep:indexmap", "dep:once_cell", "dep:static_assertions", "dep:rayon",
+  "dep:tempfile",
+]
 unreserved = []
 default = [
     "aes-crypto",
@@ -76,6 +108,7 @@ default = [
     "lzma",
     "time",
     "zstd",
+    "tokio-async"
 ]
 
 [[bench]]
@@ -85,3 +118,20 @@ harness = false
 [[bench]]
 name = "read_metadata"
 harness = false
+
+[[bench]]
+name = "extract"
+harness = false
+
+# [[bench]]
+# name = "merge_archive"
+# harness = false
+
+[profile.release]
+strip                   = false
+debug                   = true
+# lto                     = true
+
+[package.metadata.docs.rs]
+all-features = true
+rustdoc-args = ["--cfg", "docsrs"]

benches/extract.rs

@@ -0,0 +1,221 @@
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
+
+use std::io::Cursor;
+use std::path::{Path, PathBuf};
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+
+use getrandom::getrandom;
+use once_cell::sync::Lazy;
+use tempfile::tempdir;
+use tokio::{fs, io, runtime::Runtime};
+use uuid::Uuid;
+
+use zip::{result::ZipResult, write::FileOptions, ZipWriter};
+
+fn generate_random_archive(
+    num_entries: usize,
+    entry_size: usize,
+    options: FileOptions,
+) -> ZipResult<Cursor<Box<[u8]>>> {
+    use std::io::Write;
+
+    let buf = Cursor::new(Vec::new());
+    let mut zip = ZipWriter::new(buf);
+
+    let mut bytes = vec![0u8; entry_size];
+    for i in 0..num_entries {
+        let name = format!("random{}.dat", i);
+        zip.start_file(name, options)?;
+        getrandom(&mut bytes).unwrap();
+        zip.write_all(&bytes)?;
+    }
+
+    let buf = zip.finish()?.into_inner();
+
+    Ok(Cursor::new(buf.into_boxed_slice()))
+}
+
+static BIG_ARCHIVE_PATH: Lazy<PathBuf> =
+    Lazy::new(|| Path::new(env!("CARGO_MANIFEST_DIR")).join("benches/target.zip"));
+
+static SMALL_ARCHIVE_PATH: Lazy<PathBuf> =
+    Lazy::new(|| Path::new(env!("CARGO_MANIFEST_DIR")).join("benches/small-target.zip"));
+
+const NUM_ENTRIES: usize = 1_000;
+const ENTRY_SIZE: usize = 10_000;
+
+pub fn bench_io(c: &mut Criterion) {
+    let rt = Runtime::new().unwrap();
+
+    let mut group = c.benchmark_group("io");
+
+    let options = FileOptions::default().compression_method(zip::CompressionMethod::Deflated);
+
+    let gen_td = tempdir().unwrap();
+    let random_path = gen_td.path().join("random.zip");
+    std::io::copy(
+        &mut generate_random_archive(NUM_ENTRIES, ENTRY_SIZE, options).unwrap(),
+        &mut std::fs::File::create(&random_path).unwrap(),
+    )
+    .unwrap();
+
+    for (path, desc, n) in [
+        (&*BIG_ARCHIVE_PATH, "big archive", Some(30)),
+        (&*SMALL_ARCHIVE_PATH, "small archive", None),
+        (&random_path, "random archive", None),
+    ] {
+        let len = std::fs::metadata(&path).unwrap().len() as usize;
+        let id = format!("{}({} bytes)", desc, len);
+
+        group.throughput(Throughput::Bytes(len as u64));
+
+        if let Some(n) = n {
+            group.sample_size(n);
+        }
+
+        group.bench_function(BenchmarkId::new(&id, "<copy_file_range>"), |b| {
+            use zip::tokio::os::copy_file_range::*;
+
+            let td = tempdir().unwrap();
+            b.to_async(&rt).iter(|| async {
+                let sync_handle = std::fs::File::open(path).unwrap();
+                let mut src = MutateInnerOffset::new(sync_handle, Role::Readable).unwrap();
+                let mut src = Pin::new(&mut src);
+
+                let cur_name = format!("{}.zip", Uuid::new_v4());
+                let tf = td.path().join(cur_name);
+                let out = std::fs::File::create(tf).unwrap();
+                let mut dst = MutateInnerOffset::new(out, Role::Writable).unwrap();
+                let mut dst = Pin::new(&mut dst);
+
+                let written = copy_file_range(src, dst, len).await.unwrap();
+                assert_eq!(written, len);
+            });
+        });
+
+        group.bench_function(BenchmarkId::new(&id, "<async copy_buf>"), |b| {
+            let td = tempdir().unwrap();
+            b.to_async(&rt).iter(|| async {
+                let handle = fs::File::open(path).await.unwrap();
+                let mut buf_handle = io::BufReader::with_capacity(len, handle);
+
+                let cur_name = format!("{}.zip", Uuid::new_v4());
+                let tf = td.path().join(cur_name);
+
+                let mut out = fs::File::create(tf).await.unwrap();
+                assert_eq!(
+                    len as u64,
+                    io::copy_buf(&mut buf_handle, &mut out).await.unwrap()
+                );
+            });
+        });
+
+        group.bench_function(BenchmarkId::new(&id, "<async copy (no buf)>"), |b| {
+            let td = tempdir().unwrap();
+            b.to_async(&rt).iter(|| async {
+                let mut handle = fs::File::open(path).await.unwrap();
+
+                let cur_name = format!("{}.zip", Uuid::new_v4());
+                let tf = td.path().join(cur_name);
+
+                let mut out = fs::File::create(tf).await.unwrap();
+                assert_eq!(len as u64, io::copy(&mut handle, &mut out).await.unwrap());
+            });
+        });
+
+        group.bench_function(BenchmarkId::new(&id, "<sync copy (no buf)>"), |b| {
+            let td = tempdir().unwrap();
+            b.iter(|| {
+                let mut sync_handle = std::fs::File::open(path).unwrap();
+
+                let cur_name = format!("{}.zip", Uuid::new_v4());
+                let tf = td.path().join(cur_name);
+
+                let mut out = std::fs::File::create(tf).unwrap();
+                assert_eq!(
+                    len as u64,
+                    /* NB: this doesn't use copy_buf like the async case, because std::io has no
+                     * corresponding function, and using an std::io::BufReader wrapper actually
+                     * hurts perf by a lot!! */
+                    std::io::copy(&mut sync_handle, &mut out).unwrap()
+                );
+            });
+        });
+    }
+}
+
+pub fn bench_extract(c: &mut Criterion) {
+    let rt = Runtime::new().unwrap();
+    let mut group = c.benchmark_group("extract");
+
+    let options = FileOptions::default().compression_method(zip::CompressionMethod::Deflated);
+
+    let gen_td = tempdir().unwrap();
+    let random_path = gen_td.path().join("random.zip");
+    std::io::copy(
+        &mut generate_random_archive(NUM_ENTRIES, ENTRY_SIZE, options).unwrap(),
+        &mut std::fs::File::create(&random_path).unwrap(),
+    )
+    .unwrap();
+
+    for (path, desc, n, t) in [
+        (
+            &*BIG_ARCHIVE_PATH,
+            "big archive",
+            Some(10),
+            Some(Duration::from_secs(10)),
+        ),
+        (&*SMALL_ARCHIVE_PATH, "small archive", None, None),
+        (&random_path, "random archive", None, None),
+    ] {
+        let len = std::fs::metadata(&path).unwrap().len() as usize;
+        let id = format!("{}({} bytes)", desc, len);
+
+        group.throughput(Throughput::Bytes(len as u64));
+
+        if let Some(n) = n {
+            group.sample_size(n);
+        }
+        if let Some(t) = t {
+            group.measurement_time(t);
+        }
+
+        group.bench_function(BenchmarkId::new(&id, "<async extract()>"), |b| {
+            let td = tempdir().unwrap();
+            b.to_async(&rt).iter(|| async {
+                let out_dir = Arc::new(td.path().to_path_buf());
+                let handle = fs::File::open(path).await.unwrap();
+                let mut zip = zip::tokio::read::ZipArchive::new(Box::pin(handle))
+                    .await
+                    .unwrap();
+                Pin::new(&mut zip).extract(out_dir).await.unwrap();
+            })
+        });
+
+        group.bench_function(BenchmarkId::new(&id, "<async extract_simple()>"), |b| {
+            let td = tempdir().unwrap();
+            b.to_async(&rt).iter(|| async {
+                let out_dir = Arc::new(td.path().to_path_buf());
+                let handle = fs::File::open(path).await.unwrap();
+                let mut zip = zip::tokio::read::ZipArchive::new(Box::pin(handle))
+                    .await
+                    .unwrap();
+                Pin::new(&mut zip).extract_simple(out_dir).await.unwrap();
+            })
+        });
+
+        group.bench_function(BenchmarkId::new(&id, "<sync extract()>"), |b| {
+            let td = tempdir().unwrap();
+            b.iter(|| {
+                let sync_handle = std::fs::File::open(path).unwrap();
+                let mut zip = zip::read::ZipArchive::new(sync_handle).unwrap();
+                zip.extract(td.path()).unwrap();
+            })
+        });
+    }
+}
+
+criterion_group!(benches, bench_io, bench_extract);
+criterion_main!(benches);