git_internal/internal/pack/encode/
parallel.rs

1//! Parallel (non-delta) pack encoding path.
2//!
3//! When `window_size == 0`, entries are independently zlib-compressed in parallel via Rayon.
4//! Input is read in bounded batches to balance memory usage and parallelism.
5//!
6//! This path is not compatible with delta compression — use the delta path in
7//! [`super::inner_encode`] for windowed encoding.
8
9use rayon::prelude::*;
10use tokio::sync::mpsc;
11
12use super::header::{encode_header, encode_one_object};
13use crate::{
14    errors::GitError,
15    hash::ObjectHash,
16    internal::{
17        metadata::{EntryMeta, MetaAttached},
18        pack::{entry::Entry, index_entry::IndexEntry},
19    },
20    time_it,
21};
22
23impl super::PackEncoder {
24    /// Encode independent objects in parallel without delta compression.
25    ///
26    /// Input is read in bounded batches, and Rayon performs each object's header construction and
27    /// zlib compression concurrently. Rayon collection preserves input order, after which chunks
28    /// are written serially so pack offsets and the running checksum remain correct.
29    ///
30    /// This path is valid only when `window_size == 0`.
31    pub async fn parallel_encode(
32        &mut self,
33        mut entry_rx: mpsc::Receiver<MetaAttached<Entry, EntryMeta>>,
34    ) -> Result<(), GitError> {
35        if self.window_size != 0 {
36            return Err(GitError::PackEncodeError(
37                "parallel encode only works when window_size == 0".to_string(),
38            ));
39        }
40
41        // As in the delta path, the trailer checksum covers the header and all entries.
42        let head = encode_header(self.object_number);
43        self.inner_hash.update(&head);
44        self.send_data(head).await;
45
46        // Reusing the same encoder would corrupt its running offset and checksum state.
47        if self.start_encoding {
48            return Err(GitError::PackEncodeError(
49                "encoding operation is already in progress".to_string(),
50            ));
51        }
52
53        let mut idx_entries = Vec::with_capacity(self.object_number);
54        // Batching bounds temporary memory while giving Rayon enough work to distribute.
55        let batch_size = usize::max(1000, entry_rx.max_capacity() / 10); // Temporary heuristic.
56        tracing::info!("encode with batch size: {}", batch_size);
57        loop {
58            let mut batch_entries = Vec::with_capacity(batch_size);
59            time_it!("parallel encode: receive batch", {
60                for _ in 0..batch_size {
61                    match entry_rx.recv().await {
62                        Some(entry) => {
63                            if entry.inner.obj_type.is_ai_object() {
64                                return Err(GitError::PackEncodeError(format!(
65                                    "AI object type `{}` cannot be encoded in a pack file",
66                                    entry.inner.obj_type
67                                )));
68                            }
69                            batch_entries.push(entry.inner);
70                            self.process_index += 1;
71                        }
72                        None => break,
73                    }
74                }
75            });
76
77            if batch_entries.is_empty() {
78                break;
79            }
80
81            // Indexed parallel collection retains batch order even though compression finishes on
82            // different Rayon workers.
83            let batch_result: Vec<Result<(Vec<u8>, IndexEntry), GitError>> =
84                time_it!("parallel encode: encode batch", {
85                    batch_entries
86                        .par_iter()
87                        .map(|entry| {
88                            encode_one_object(entry, None)
89                                .map(|encoded| (encoded, IndexEntry::new(entry, 0)))
90                        })
91                        .collect()
92                });
93
94            time_it!("parallel encode: write batch", {
95                for obj_data in batch_result {
96                    let (encoded_bytes, mut idx_entry) = obj_data?;
97                    idx_entry.offset = self.inner_offset as u64;
98                    self.write_owned_and_update(encoded_bytes).await;
99                    idx_entries.push(idx_entry);
100                }
101            });
102        }
103
104        tracing::debug!("parallel encode idx entries: {:?}", idx_entries.len());
105        if self.process_index != self.object_number {
106            panic!(
107                "not all objects are encoded, process:{}, total:{}",
108                self.process_index, self.object_number
109            );
110        }
111
112        // Append the checksum trailer only after every encoded entry has updated the running hash.
113        // Infer the kind from the checksum length: this task may run on an async worker
114        // thread whose thread-local `HashKind` was never set, so `ObjectHash::from_bytes`
115        // could disagree with the hasher chosen at encoder construction.
116        let hash_result = self.inner_hash.clone().finalize();
117        self.final_hash = Some(
118            ObjectHash::from_bytes_infer_kind(&hash_result).map_err(GitError::PackEncodeError)?,
119        );
120        self.send_data(hash_result).await;
121        self.drop_sender();
122
123        self.idx_entries = Some(idx_entries);
124        Ok(())
125    }
126}
git_internal/internal/pack/encode/parallel.rs

git_internal/internal/pack/encode/
parallel.rs
