gitoxide_core/pack/
create.rs

1use std::{ffi::OsStr, io, path::Path, str::FromStr, time::Instant};
2
3use anyhow::anyhow;
4use gix::{
5    Count, NestedProgress, Progress, hash, hash::ObjectId, interrupt, objs::bstr::ByteVec, odb::pack,
6    parallel::InOrderIter, prelude::Finalize, progress, traverse,
7};
8
9use crate::OutputFormat;
10
11pub const PROGRESS_RANGE: std::ops::RangeInclusive<u8> = 1..=2;
12
13#[derive(Default, Eq, PartialEq, Debug, Clone)]
14pub enum ObjectExpansion {
15    #[default]
16    None,
17    TreeTraversal,
18    TreeDiff,
19}
20
21impl ObjectExpansion {
22    pub fn variants() -> &'static [&'static str] {
23        &["none", "tree-traversal", "tree-diff"]
24    }
25}
26
27impl FromStr for ObjectExpansion {
28    type Err = String;
29
30    fn from_str(s: &str) -> Result<Self, Self::Err> {
31        use ObjectExpansion::*;
32        let slc = s.to_ascii_lowercase();
33        Ok(match slc.as_str() {
34            "none" => None,
35            "tree-traversal" => TreeTraversal,
36            "tree-diff" => TreeDiff,
37            _ => return Err("invalid value".into()),
38        })
39    }
40}
41
42impl From<ObjectExpansion> for pack::data::output::count::objects::ObjectExpansion {
43    fn from(v: ObjectExpansion) -> Self {
44        use pack::data::output::count::objects::ObjectExpansion::*;
45        match v {
46            ObjectExpansion::None => AsIs,
47            ObjectExpansion::TreeTraversal => TreeContents,
48            ObjectExpansion::TreeDiff => TreeAdditionsComparedToAncestor,
49        }
50    }
51}
52
53/// A general purpose context for many operations provided here
54pub struct Context<W> {
55    /// The way input objects should be handled
56    pub expansion: ObjectExpansion,
57    /// If `Some(threads)`, use this amount of `threads` to accelerate the counting phase at the cost of losing
58    /// determinism as the order of objects during expansion changes with multiple threads unless no expansion is performed.
59    /// In the latter case, this flag has no effect.
60    /// If `None`, counting will only use one thread and thus yield the same sequence of objects in any case.
61    pub nondeterministic_thread_count: Option<usize>,
62    /// If true, delta objects may refer to their base as reference, allowing it not to be included in the created back.
63    /// Otherwise these have to be recompressed in order to make the pack self-contained.
64    pub thin: bool,
65    /// If set, don't use more than this amount of threads.
66    /// Otherwise, usually use as many threads as there are logical cores.
67    /// A value of 0 is interpreted as no-limit
68    pub thread_limit: Option<usize>,
69    /// If set, statistics about the operation will be written to the output stream.
70    pub statistics: Option<OutputFormat>,
71    /// The size of the cache storing fully decoded delta objects. This can greatly speed up pack decoding by reducing the length of delta
72    /// chains. Note that caches also incur a cost and poorly used caches may reduce overall performance.
73    /// This is a total, shared among all threads if `thread_limit` permits.
74    ///
75    /// If 0, the cache is disabled entirely.
76    pub pack_cache_size_in_bytes: usize,
77    /// The size of the cache to store full objects by their ID, bypassing any lookup in the object database.
78    /// Note that caches also incur a cost and poorly used caches may reduce overall performance.
79    ///
80    /// This is a total, shared among all threads if `thread_limit` permits.
81    /// Only used when known to be effective, namely when `expansion == ObjectExpansion::TreeDiff`.
82    pub object_cache_size_in_bytes: usize,
83    /// The output stream for use of additional information
84    pub out: W,
85}
86
87pub fn create<W, P>(
88    repository_path: impl AsRef<Path>,
89    tips: impl IntoIterator<Item = impl AsRef<OsStr>>,
90    input: Option<impl io::BufRead + Send + 'static>,
91    output_directory: Option<impl AsRef<Path>>,
92    mut progress: P,
93    Context {
94        expansion,
95        nondeterministic_thread_count,
96        thin,
97        thread_limit,
98        statistics,
99        pack_cache_size_in_bytes,
100        object_cache_size_in_bytes,
101        mut out,
102    }: Context<W>,
103) -> anyhow::Result<()>
104where
105    W: std::io::Write,
106    P: NestedProgress,
107    P::SubProgress: 'static,
108{
109    type ObjectIdIter = dyn Iterator<Item = Result<ObjectId, Box<dyn std::error::Error + Send + Sync>>> + Send;
110
111    let repo = gix::discover(repository_path)?;
112    let pack_compression = repo.pack_compression()?;
113    let repo = repo.into_sync();
114    progress.init(Some(2), progress::steps());
115    let tips = tips.into_iter();
116    let make_cancellation_err = || anyhow!("Cancelled by user");
117    let (mut handle, mut input): (_, Box<ObjectIdIter>) = match input {
118        None => {
119            let mut progress = progress.add_child("traversing");
120            progress.init(None, progress::count("commits"));
121            let tips = tips
122                .map({
123                    let easy = repo.to_thread_local();
124                    move |tip| {
125                        ObjectId::from_hex(&Vec::from_os_str_lossy(tip.as_ref())).or_else(|_| {
126                            easy.find_reference(tip.as_ref())
127                                .map_err(anyhow::Error::from)
128                                .and_then(|r| r.into_fully_peeled_id().map(gix::Id::detach).map_err(Into::into))
129                        })
130                    }
131                })
132                .collect::<Result<Vec<_>, _>>()?;
133            let handle = repo.objects.into_shared_arc().to_cache_arc();
134            let iter = Box::new(
135                traverse::commit::Simple::new(tips, handle.clone())
136                    .map(|res| res.map_err(|err| Box::new(err) as Box<_>).map(|c| c.id))
137                    .inspect(move |_| progress.inc()),
138            );
139            (handle, iter)
140        }
141        Some(input) => {
142            let mut progress = progress.add_child("iterating");
143            progress.init(None, progress::count("objects"));
144            let handle = repo.objects.into_shared_arc().to_cache_arc();
145            (
146                handle,
147                Box::new(
148                    input
149                        .lines()
150                        .map(|hex_id| {
151                            hex_id
152                                .map_err(|err| Box::new(err) as Box<_>)
153                                .and_then(|hex_id| ObjectId::from_hex(hex_id.as_bytes()).map_err(Into::into))
154                        })
155                        .inspect(move |_| progress.inc()),
156                ),
157            )
158        }
159    };
160
161    let mut stats = Statistics::default();
162    let chunk_size = 1000; // What's a good value for this?
163    let counts = {
164        let mut progress = progress.add_child("counting");
165        progress.init(None, progress::count("objects"));
166        let may_use_multiple_threads =
167            nondeterministic_thread_count.is_some() || matches!(expansion, ObjectExpansion::None);
168        let thread_limit = if may_use_multiple_threads {
169            nondeterministic_thread_count.or(thread_limit)
170        } else {
171            Some(1)
172        };
173        if nondeterministic_thread_count.is_some() && !may_use_multiple_threads {
174            progress.fail("Cannot use multi-threaded counting in tree-diff object expansion mode as it may yield way too many objects.".into());
175        }
176        let (_, _, thread_count) = gix::parallel::optimize_chunk_size_and_thread_limit(50, None, thread_limit, None);
177        let progress = progress::ThroughputOnDrop::new(progress);
178
179        {
180            let per_thread_object_pack_size = pack_cache_size_in_bytes / thread_count;
181            if per_thread_object_pack_size >= 10_000 {
182                handle.set_pack_cache(move || {
183                    Box::new(pack::cache::lru::MemoryCappedHashmap::new(per_thread_object_pack_size))
184                });
185            }
186            if matches!(expansion, ObjectExpansion::TreeDiff) {
187                handle.set_object_cache(move || {
188                    let per_thread_object_cache_size = object_cache_size_in_bytes / thread_count;
189                    Box::new(pack::cache::object::MemoryCappedHashmap::new(
190                        per_thread_object_cache_size,
191                    ))
192                });
193            }
194        }
195        let input_object_expansion = expansion.into();
196        handle.prevent_pack_unload();
197        handle.ignore_replacements = true;
198        let (mut counts, count_stats) = if may_use_multiple_threads {
199            pack::data::output::count::objects(
200                handle.clone(),
201                input,
202                &progress,
203                &interrupt::IS_INTERRUPTED,
204                pack::data::output::count::objects::Options {
205                    thread_limit,
206                    chunk_size,
207                    input_object_expansion,
208                },
209            )?
210        } else {
211            pack::data::output::count::objects_unthreaded(
212                &handle,
213                &mut input,
214                &progress,
215                &interrupt::IS_INTERRUPTED,
216                input_object_expansion,
217            )?
218        };
219        stats.counts = count_stats;
220        counts.shrink_to_fit();
221        counts
222    };
223
224    progress.inc();
225    let num_objects = counts.len();
226    let mut in_order_entries = {
227        let progress = progress.add_child("creating entries");
228        InOrderIter::from(pack::data::output::entry::iter_from_counts(
229            counts,
230            handle,
231            Box::new(progress),
232            pack::data::output::entry::iter_from_counts::Options {
233                thread_limit,
234                mode: pack::data::output::entry::iter_from_counts::Mode::PackCopyAndBaseObjects,
235                allow_thin_pack: thin,
236                chunk_size,
237                version: Default::default(),
238                compression: pack_compression,
239            },
240        ))
241    };
242
243    let mut entries_progress = progress.add_child("consuming");
244    entries_progress.init(Some(num_objects), progress::count("entries"));
245    let mut write_progress = progress.add_child("writing");
246    write_progress.init(None, progress::bytes());
247    let start = Instant::now();
248
249    let mut named_tempfile_store: Option<tempfile::NamedTempFile> = None;
250    let mut sink_store: std::io::Sink;
251    let (mut pack_file, output_directory): (&mut dyn std::io::Write, Option<_>) = match output_directory {
252        Some(dir) => {
253            named_tempfile_store = Some(tempfile::NamedTempFile::new_in(dir.as_ref())?);
254            (named_tempfile_store.as_mut().expect("packfile just set"), Some(dir))
255        }
256        None => {
257            sink_store = std::io::sink();
258            (&mut sink_store, None)
259        }
260    };
261    let mut interruptible_output_iter = interrupt::Iter::new(
262        pack::data::output::bytes::FromEntriesIter::new(
263            in_order_entries.by_ref().inspect(|e| {
264                if let Ok(entries) = e {
265                    entries_progress.inc_by(entries.len());
266                }
267            }),
268            &mut pack_file,
269            num_objects as u32,
270            pack::data::Version::default(),
271            hash::Kind::default(),
272        ),
273        make_cancellation_err,
274    );
275    for io_res in interruptible_output_iter.by_ref() {
276        let written = io_res??;
277        write_progress.inc_by(written as usize);
278    }
279
280    let hash = interruptible_output_iter
281        .into_inner()
282        .digest()
283        .expect("iteration is done");
284    let pack_name = format!("{hash}.pack");
285    if let (Some(pack_file), Some(dir)) = (named_tempfile_store.take(), output_directory) {
286        pack_file.persist(dir.as_ref().join(pack_name))?;
287    } else {
288        writeln!(out, "{pack_name}")?;
289    }
290    stats.entries = in_order_entries.inner.finalize()?;
291
292    write_progress.show_throughput(start);
293    entries_progress.show_throughput(start);
294
295    if let Some(format) = statistics {
296        print(stats, format, out)?;
297    }
298    progress.inc();
299    Ok(())
300}
301
302fn print(stats: Statistics, format: OutputFormat, out: impl std::io::Write) -> anyhow::Result<()> {
303    match format {
304        OutputFormat::Human => human_output(stats, out).map_err(Into::into),
305        #[cfg(feature = "serde")]
306        OutputFormat::Json => serde_json::to_writer_pretty(out, &stats).map_err(Into::into),
307    }
308}
309
310fn human_output(
311    Statistics {
312        counts:
313            pack::data::output::count::objects::Outcome {
314                input_objects,
315                expanded_objects,
316                decoded_objects,
317                total_objects,
318            },
319        entries:
320            pack::data::output::entry::iter_from_counts::Outcome {
321                decoded_and_recompressed_objects,
322                missing_objects,
323                objects_copied_from_pack,
324                ref_delta_objects,
325            },
326    }: Statistics,
327    mut out: impl std::io::Write,
328) -> std::io::Result<()> {
329    let width = 30;
330    writeln!(out, "counting phase")?;
331    #[rustfmt::skip]
332    writeln!(
333        out,
334        "\t{:<width$} {}\n\t{:<width$} {}\n\t{:<width$} {}\n\t{:<width$} {}",
335        "input objects", input_objects,
336        "expanded objects", expanded_objects,
337        "decoded objects", decoded_objects,
338        "total objects", total_objects,
339        width = width
340    )?;
341    writeln!(out, "generation phase")?;
342    #[rustfmt::skip]
343    writeln!(
344        out,
345        "\t{:<width$} {}\n\t{:<width$} {}\n\t{:<width$} {}\n\t{:<width$} {}",
346        "decoded and recompressed", decoded_and_recompressed_objects,
347        "pack-to-pack copies", objects_copied_from_pack,
348        "ref-delta-objects", ref_delta_objects,
349        "missing objects", missing_objects,
350        width = width
351    )?;
352    Ok(())
353}
354
355#[derive(Default)]
356#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
357struct Statistics {
358    counts: pack::data::output::count::objects::Outcome,
359    entries: pack::data::output::entry::iter_from_counts::Outcome,
360}
361
362pub mod input_iteration {
363    use gix::{hash, traverse};
364    #[derive(Debug, thiserror::Error)]
365    pub enum Error {
366        #[error("input objects couldn't be iterated completely")]
367        Iteration(#[from] traverse::commit::simple::Error),
368        #[error("An error occurred while reading hashes from standard input")]
369        InputLinesIo(#[from] std::io::Error),
370        #[error("Could not decode hex hash provided on standard input")]
371        HashDecode(#[from] hash::decode::Error),
372    }
373}
gitoxide_core/pack/create.rs

gitoxide_core/pack/
create.rs
