parallel loading of entries right before reducing them (#293)

Author: Byron

git-index/src/decode/entries.rs

@@ -1,7 +1,7 @@
 use std::ops::Range;
 
 use crate::{
-    decode::{header, Error},
+    decode::{self, header},
     entry,
     util::{read_u32, split_at_byte_exclusive, split_at_pos},
     Entry, Version,
@@ -11,8 +11,6 @@ use crate::{
 pub const AVERAGE_V4_DELTA_PATH_LEN_IN_BYTES: usize = 80;
 
 pub struct Outcome {
-    pub entries: Vec<Entry>,
-    pub path_backing: Vec<u8>,
     pub is_sparse: bool,
 }
 
@@ -42,15 +40,14 @@ pub fn estimate_path_storage_requirements_in_bytes(
 }
 
 /// Note that `data` must point to the beginning of the entries, right past the header.
-pub fn load_all(
-    mut data: &[u8],
+pub fn load_chunk<'a>(
+    mut data: &'a [u8],
+    entries: &mut Vec<Entry>,
+    path_backing: &mut Vec<u8>,
     num_entries: u32,
-    path_backing_capacity: usize,
     object_hash: git_hash::Kind,
     version: Version,
-) -> Result<(Outcome, &[u8]), Error> {
-    let mut path_backing = Vec::<u8>::with_capacity(path_backing_capacity);
-    let mut entries = Vec::<Entry>::with_capacity(num_entries as usize);
+) -> Result<(Outcome, &'a [u8]), decode::Error> {
     let mut is_sparse = false;
     let has_delta_paths = version == Version::V4;
     let mut prev_path = None;
@@ -59,12 +56,12 @@ pub fn load_all(
     for idx in 0..num_entries {
         let (entry, remaining) = load_one(
             data,
-            &mut path_backing,
+            path_backing,
             object_hash.len_in_bytes(),
             has_delta_paths,
             prev_path,
         )
-        .ok_or(Error::Entry(idx))?;
+        .ok_or(decode::Error::Entry(idx))?;
 
         data = remaining;
         if entry::mode::is_sparse(entry.stat.mode) {
@@ -77,14 +74,7 @@ pub fn load_all(
         prev_path = entries.last().map(|e| (e.path.clone(), &mut delta_buf));
     }
 
-    Ok((
-        Outcome {
-            entries,
-            path_backing,
-            is_sparse,
-        },
-        data,
-    ))
+    Ok((Outcome { is_sparse }, data))
 }
 
 /// Note that `prev_path` is only useful if the version is V4

git-index/src/decode/mod.rs

@@ -1,6 +1,6 @@
 use filetime::FileTime;
 
-use crate::{extension, State};
+use crate::{extension, Entry, State, Version};
 
 mod entries;
 pub mod header;
@@ -49,13 +49,13 @@ impl State {
         Options {
             object_hash,
             thread_limit,
-            min_extension_block_in_bytes_for_threading: _,
+            min_extension_block_in_bytes_for_threading,
         }: Options,
     ) -> Result<(Self, git_hash::ObjectId), Error> {
         let (version, num_entries, post_header_data) = header::decode(data, object_hash)?;
         let start_of_extensions = extension::end_of_index_entry::decode(data, object_hash);
 
-        let num_threads = git_features::parallel::num_threads(thread_limit);
+        let mut num_threads = git_features::parallel::num_threads(thread_limit);
         let path_backing_buffer_size = entries::estimate_path_storage_requirements_in_bytes(
             num_entries,
             data.len(),
@@ -66,37 +66,82 @@ impl State {
 
         let (entries, ext, data) = match start_of_extensions {
             Some(offset) if num_threads > 1 => {
-                let start_of_extensions = &data[offset..];
-                let index_offsets_table = extension::index_entry_offset_table::find(start_of_extensions, object_hash);
-                let (entries_res, (ext, data)) = git_features::parallel::threads(|_scope| {
-                    match index_offsets_table {
+                let extensions_data = &data[offset..];
+                let index_offsets_table = extension::index_entry_offset_table::find(extensions_data, object_hash);
+                let (entries_res, (ext, data)) = git_features::parallel::threads(|scope| {
+                    let extension_loading =
+                        (extensions_data.len() > min_extension_block_in_bytes_for_threading).then({
+                            num_threads -= 1;
+                            || scope.spawn(|_| extension::decode::all(extensions_data, object_hash))
+                        });
+                    let entries_res = match index_offsets_table {
                         Some(entry_offsets) => {
-                            dbg!(entry_offsets);
-                            todo!("threaded entry loading if its worth it")
+                            let chunk_size = (entry_offsets.len() as f32 / num_threads as f32).ceil() as usize;
+                            let num_chunks = entry_offsets.chunks(chunk_size).count();
+                            let mut threads = Vec::with_capacity(num_chunks);
+                            for (id, chunks) in entry_offsets.chunks(chunk_size).enumerate() {
+                                let chunks = chunks.to_vec();
+                                threads.push(scope.spawn(move |_| {
+                                    let num_entries = chunks.iter().map(|c| c.num_entries).sum::<u32>() as usize;
+                                    let mut entries = Vec::with_capacity(num_entries);
+                                    let path_backing_buffer_size = entries::estimate_path_storage_requirements_in_bytes(
+                                        num_entries as u32,
+                                        data.len() / num_chunks,
+                                        start_of_extensions.map(|ofs| ofs / num_chunks),
+                                        object_hash,
+                                        version,
+                                    );
+                                    let mut path_backing = Vec::with_capacity(path_backing_buffer_size);
+                                    let mut is_sparse = false;
+                                    for offset in chunks {
+                                        let (
+                                            entries::Outcome {
+                                                is_sparse: chunk_is_sparse,
+                                            },
+                                            _data,
+                                        ) = entries::load_chunk(
+                                            &data[offset.from_beginning_of_file as usize..],
+                                            &mut entries,
+                                            &mut path_backing,
+                                            offset.num_entries,
+                                            object_hash,
+                                            version,
+                                        )?;
+                                        is_sparse |= chunk_is_sparse;
+                                    }
+                                    Ok::<_, Error>((
+                                        id,
+                                        EntriesOutcome {
+                                            entries,
+                                            path_backing,
+                                            is_sparse,
+                                        },
+                                    ))
+                                }));
+                            }
+                            todo!("combined thread results in order ")
                         }
-                        None => {
-                            // TODO load all extensions in scoped, then get IEOT, then possibly multi-threaded entry parsing
-                            (
-                                entries::load_all(
-                                    post_header_data,
-                                    num_entries,
-                                    path_backing_buffer_size,
-                                    object_hash,
-                                    version,
-                                ),
-                                extension::decode::all(start_of_extensions, object_hash),
-                            )
-                        }
-                    }
+                        None => load_entries(
+                            post_header_data,
+                            path_backing_buffer_size,
+                            num_entries,
+                            object_hash,
+                            version,
+                        ),
+                    };
+                    let ext_res = extension_loading
+                        .map(|thread| thread.join().unwrap())
+                        .unwrap_or_else(|| extension::decode::all(extensions_data, object_hash));
+                    (entries_res, ext_res)
                 })
                 .unwrap(); // this unwrap is for panics - if these happened we are done anyway.
                 (entries_res?.0, ext, data)
             }
             None | Some(_) => {
-                let (entries, data) = entries::load_all(
+                let (entries, data) = load_entries(
                     post_header_data,
-                    num_entries,
                     path_backing_buffer_size,
+                    num_entries,
                     object_hash,
                     version,
                 )?;
@@ -113,7 +158,7 @@ impl State {
         }
 
         let checksum = git_hash::ObjectId::from(data);
-        let entries::Outcome {
+        let EntriesOutcome {
             entries,
             path_backing,
             is_sparse,
@@ -133,3 +178,38 @@ impl State {
         ))
     }
 }
+
+struct EntriesOutcome {
+    pub entries: Vec<Entry>,
+    pub path_backing: Vec<u8>,
+    pub is_sparse: bool,
+}
+
+fn load_entries(
+    post_header_data: &[u8],
+    path_backing_buffer_size: usize,
+    num_entries: u32,
+    object_hash: git_hash::Kind,
+    version: Version,
+) -> Result<(EntriesOutcome, &[u8]), Error> {
+    let mut entries = Vec::with_capacity(num_entries as usize);
+    let mut path_backing = Vec::with_capacity(path_backing_buffer_size);
+    entries::load_chunk(
+        post_header_data,
+        &mut entries,
+        &mut path_backing,
+        num_entries,
+        object_hash,
+        version,
+    )
+    .map(|(entries::Outcome { is_sparse }, data): (entries::Outcome, &[u8])| {
+        (
+            EntriesOutcome {
+                entries,
+                path_backing,
+                is_sparse,
+            },
+            data,
+        )
+    })
+}

git-index/src/extension/mod.rs

@@ -31,7 +31,7 @@ pub(crate) mod end_of_index_entry;
 pub(crate) mod index_entry_offset_table {
     use crate::{extension, extension::Signature, util::read_u32};
 
-    #[derive(Debug)]
+    #[derive(Debug, Clone, Copy)]
     pub struct Offset {
         pub from_beginning_of_file: u32,
         pub num_entries: u32,

git-index/tests/index/file/mod.rs

@@ -5,17 +5,28 @@ mod init {
     fn file(name: &str) -> git_index::File {
         git_index::File::at(crate::index::fixture_path(name), git_index::decode::Options::default()).unwrap()
     }
+    fn file_opt(name: &str, opts: git_index::decode::Options) -> git_index::File {
+        git_index::File::at(crate::index::fixture_path(name), opts).unwrap()
+    }
 
     #[test]
     fn read_v2_with_single_entry_tree_and_eoie_ext() {
-        let file = file("v2");
-        assert_eq!(file.version(), Version::V2);
+        let file_disallow_threaded_loading = file_opt(
+            "v2",
+            git_index::decode::Options {
+                min_extension_block_in_bytes_for_threading: 100000,
+                ..Default::default()
+            },
+        );
+        for file in [file("v2"), file_disallow_threaded_loading] {
+            assert_eq!(file.version(), Version::V2);
 
-        assert_eq!(file.entries().len(), 1);
+            assert_eq!(file.entries().len(), 1);
 
-        let entry = &file.entries()[0];
-        assert_eq!(entry.id, hex_to_id("e69de29bb2d1d6434b8b29ae775ad8c2e48c5391"));
-        assert_eq!(entry.path(&file.state), "a");
+            let entry = &file.entries()[0];
+            assert_eq!(entry.id, hex_to_id("e69de29bb2d1d6434b8b29ae775ad8c2e48c5391"));
+            assert_eq!(entry.path(&file.state), "a");
+        }
     }
 
     #[test]