pipelined extraction

- initial sketch of lexicographic trie for pipelining
- move path splitting into a submodule
- lex trie can now propagate entry data
- outline handle allocation
- mostly handle files
- mostly handle dirs
- clarify symlink FIXMEs
- do symlink validation
- extract writable dir setting to helper method
- modify args to handle allocation method
- handle allocation test passes
- simplify perms a lot
- outline evaluation
- handle symlinks
- BIGGER CHANGE! add EntryReader/etc
- make initial pipelined extract work
- fix file perms by writing them after finishing the file write
- support directory entries by unix mode as well
- impl split extraction
- remove dependency on reader refactoring
- add dead_code to methods we don't use yet
- bzip2 support needed for benchmark test
- correctly handle backslashes in entry names (i.e. don't)
- make PathSplitError avoid consing a String until necessary
- add repro_old423 test for pipelining
- silence dead code warnings for windows
- avoid erroring for top-level directory entries
- use num_cpus by default for parallelism
- we spawn three threads per chunk
- add dynamically-generated test archive
- initialize the test archives exactly once in statics
- add benchmarks for dynamic and static test data
- use lazy_static
- add FIXME for follow-up work to support absolute paths
- impl From<DirEntry<...>> for FSEntry
- move handle_creation module to a separate file
- downgrade HandleCreationError to io::Error
- use ByAddress over ZipDataHandle
- replace unsafe transmutes with Pod methods
- add note about shared future dependency task DAG

- box each level of the b-tree together with its values

this may technically reduce heap fragmentation, but since this data structure only exists
temporarily, that's probably not too important. instead, this change just reduces the amount of
coercion and unboxing we need to do

Author: cosmicexplorer

Cargo.toml

@@ -28,20 +28,24 @@ time = { version = "0.3.37", default-features = false }
 
 [dependencies]
 aes = { version = "0.8", optional = true }
+by_address = { version = "1.2.1", optional = true }
 bzip2 = { version = "0.6.0", optional = true }
 chrono = { version = "^0.4.27", optional = true }
 constant_time_eq = { version = "0.3.1", optional = true }
 crc32fast = "1.4"
+displaydoc = "0.2.5"
 flate2 = { version = "1.1.1", default-features = false, optional = true }
 getrandom = { version = "0.3.1", features = ["std"], optional = true }
 hmac = { version = "0.12", optional = true, features = ["reset"] }
 indexmap = "2"
 jiff = { version = "0.2.4", optional = true }
 memchr = "2.7"
 nt-time = { version = "0.10.6", default-features = false, optional = true }
+num_cpus = { version = "1.16", optional = true }
 ppmd-rust = { version = "1.2", optional = true }
 pbkdf2 = { version = "0.12", optional = true }
 sha1 = { version = "0.10", optional = true }
+thiserror = "2"
 time = { workspace = true, optional = true, features = [
     "std",
 ] }
@@ -69,8 +73,7 @@ anyhow = "1.0.95"
 clap = { version = "=4.4.18", features = ["derive"] }
 tempdir = "0.3.7"
 tempfile = "3.15"
-lazy_static = "1.5"
-num_cpus = "1.16"
+num_cpus = "1"
 
 [features]
 aes-crypto = ["dep:aes", "dep:constant_time_eq", "hmac", "pbkdf2", "sha1", "getrandom", "zeroize"]
@@ -95,6 +98,7 @@ parallelism = ["libc", "num_cpus", "by_address"]
 xz = ["dep:lzma-rust2"]
 xz-static = ["lzma"]
 legacy-zip = ["bitstream-io"]
+parallelism = ["libc", "num_cpus", "by_address"]
 default = [
     "aes-crypto",
     "bzip2",

benches/extract.rs

@@ -1,13 +1,12 @@
 use bencher::{benchmark_group, benchmark_main};
 
 use bencher::Bencher;
-use lazy_static::lazy_static;
 use tempdir::TempDir;
 use tempfile::tempfile;
 
 use std::fs;
 use std::path::Path;
-use std::sync::{Arc, Mutex};
+use std::sync::{LazyLock, Mutex};
 
 use zip::result::ZipResult;
 use zip::write::ZipWriter;
@@ -34,12 +33,10 @@ fn static_test_archive() -> ZipResult<ZipArchive<fs::File>> {
     ZipArchive::new(file)
 }
 
-lazy_static! {
-    static ref STATIC_TEST_ARCHIVE: Arc<Mutex<ZipArchive<fs::File>>> = {
-        let archive = static_test_archive().unwrap();
-        Arc::new(Mutex::new(archive))
-    };
-}
+static STATIC_TEST_ARCHIVE: LazyLock<Mutex<ZipArchive<fs::File>>> = LazyLock::new(|| {
+    let archive = static_test_archive().unwrap();
+    Mutex::new(archive)
+});
 
 /* This archive is generated dynamically, in order to scale with the number of reported CPUs.
  * - We want at least 768 files (4 per VCPU on EC2 *.48xlarge instances) to run in CI.
@@ -66,13 +63,11 @@ fn dynamic_test_archive(src_archive: &mut ZipArchive<fs::File>) -> ZipResult<Zip
     output_archive.finish_into_readable()
 }
 
-lazy_static! {
-    static ref DYNAMIC_TEST_ARCHIVE: Arc<Mutex<ZipArchive<fs::File>>> = {
-        let mut src = STATIC_TEST_ARCHIVE.lock().unwrap();
-        let archive = dynamic_test_archive(&mut src).unwrap();
-        Arc::new(Mutex::new(archive))
-    };
-}
+static DYNAMIC_TEST_ARCHIVE: LazyLock<Mutex<ZipArchive<fs::File>>> = LazyLock::new(|| {
+    let mut src = STATIC_TEST_ARCHIVE.lock().unwrap();
+    let archive = dynamic_test_archive(&mut src).unwrap();
+    Mutex::new(archive)
+});
 
 fn do_extract_basic(bench: &mut Bencher, archive: &mut ZipArchive<fs::File>) {
     let total_size: u64 = archive.decompressed_size().unwrap().try_into().unwrap();

src/read/pipelining.rs

@@ -662,15 +662,16 @@ pub mod split_extraction {
 
                                 for (entry, output_file) in compressed_receiver.iter() {
                                     /* Construct the decompressing reader. */
-                                    let limited_reader = ((&mut compressed_read_end)
-                                        as &mut dyn Read)
-                                        .take(entry.compressed_size);
+                                    let limited_reader =
+                                        (&mut compressed_read_end).take(entry.compressed_size);
                                     let crypto_reader =
                                         make_crypto_reader(entry, limited_reader, None, None)?;
                                     let mut decompressing_reader = make_reader(
                                         entry.compression_method,
+                                        entry.uncompressed_size,
                                         entry.crc32,
                                         crypto_reader,
+                                        entry.flags,
                                     )?;
                                     let mut limited_writer = TakeWrite::take(
                                         uncompressed_write_end.by_ref(),