use std::{cmp::Ordering, sync::Arc}; use git_features::{parallel, progress::Progress}; use crate::data::{output, output::ChunkId}; /// Given a known list of object `counts`, calculate entries ready to be put into a data pack.////// This allows objects to be written quite soon without having to wait for the entire pack to be built in memory./// A chunk of objects is held in memory and compressed using DEFLATE, and serve the output of this iterator./// That way slow writers will naturally apply back pressure, and communicate to the implementation that more time can be/// spent compressing objects.////// * `counts`///   * A list of previously counted objects to add to the pack. Duplication checks are not performed, no object is expected to be duplicated./// * `progress`///   * a way to obtain progress information/// * `options`///   * more configuration////// _Returns_ the checksum of the pack////// ## Discussion////// ### Caches////// `make_cache` is only suitable for speeding up cache access of blobs as these are not looked up during counting anymore - it only/// sees trees as only these are needed for traversal/object counting.////// ### Advantages////// * Begins writing immediately and supports back-pressure./// * Abstract over object databases and how input is provided.////// ### Disadvantages////// * **does not yet support thin packs** as we don't have a way to determine which objects are supposed to be thin./// * ~~currently there is no way to easily write the pack index, even though the state here is uniquely positioned to do///   so with minimal overhead (especially compared to `gixp index-from-pack`)~~ Probably works now by chaining Iterators///  or keeping enough state to write a pack and then generate an index with recorded data.///pub fn iter_from_counts<Find, Cache>(    mut counts: Vec<output::Count>,    db: Find,    make_cache: impl Fn() -> Cache + Send + Clone + Sync + 'static,    mut progress: impl Progress,    Options {        version,        mode,        allow_thin_pack,        thread_limit,        chunk_size,    }: Options,) -> impl Iterator<Item = Result<(ChunkId, Vec<output::Entry>), Error<Find::Error>>>       + parallel::reduce::Finalize<Reduce = reduce::Statistics<Error<Find::Error>>>where    Find: crate::Find + Clone + Send + Sync + 'static,    <Find as crate::Find>::Error: Send,    Cache: crate::cache::DecodeEntry,{    assert!(        matches!(version, crate::data::Version::V2),        "currently we can only write version 2"    );    let (chunk_size, thread_limit, _) =        parallel::optimize_chunk_size_and_thread_limit(chunk_size, Some(counts.len()), thread_limit, None);    let chunks = util::ChunkRanges::new(chunk_size, counts.len());    {        let progress = Arc::new(parking_lot::Mutex::new(progress.add_child("resolving")));        progress.lock().init(None, git_features::progress::count("counts"));        let enough_counts_present = counts.len() > 4_000;        let start = std::time::Instant::now();        parallel::in_parallel_if(            || enough_counts_present,            chunks.clone(),            thread_limit,            |_n| Vec::<u8>::new(),            {                let progress = Arc::clone(&progress);                let counts = &counts;                let db = &db;                move |chunk_range, buf| {                    let chunk = {                        let c = &counts[chunk_range];                        let mut_ptr = c.as_ptr() as *mut output::Count;                        // SAFETY: We know that 'chunks' is only non-overlapping slices, and this function owns `counts`.                        #[allow(unsafe_code)]                        unsafe {                            std::slice::from_raw_parts_mut(mut_ptr, c.len())                        }                    };                    let chunk_size = chunk.len();                    for count in chunk {                        use crate::data::output::count::PackLocation::*;                        match count.entry_pack_location {                            LookedUp(_) => continue,                            NotLookedUp => count.entry_pack_location = LookedUp(db.location_by_oid(count.id, buf)),                        }                    }                    progress.lock().inc_by(chunk_size);                    Ok::<_, ()>(())                }            },            parallel::reduce::IdentityWithResult::<(), ()>::default(),        )        .expect("infallible - we ignore none-existing objects");        progress.lock().show_throughput(start);    }    let counts_range_by_pack_id = match mode {        Mode::PackCopyAndBaseObjects => {            let mut progress = progress.add_child("sorting");            progress.init(Some(counts.len()), git_features::progress::count("counts"));            let start = std::time::Instant::now();             use crate::data::output::count::PackLocation::*;            counts.sort_by(|lhs, rhs| match (&lhs.entry_pack_location, &rhs.entry_pack_location) {                (LookedUp(None), LookedUp(None)) => Ordering::Equal,                (LookedUp(Some(_)), LookedUp(None)) => Ordering::Greater,                (LookedUp(None), LookedUp(Some(_))) => Ordering::Less,                (LookedUp(Some(lhs)), LookedUp(Some(rhs))) => lhs                    .pack_id                    .cmp(&rhs.pack_id)                    .then(lhs.pack_offset.cmp(&rhs.pack_offset)),                (_, _) => unreachable!("counts were resolved beforehand"),            });             let mut index: Vec<(u32, std::ops::Range<usize>)> = Vec::new();            let mut chunks_pack_start = counts.partition_point(|e| e.entry_pack_location.is_none());            let mut slice = &counts[chunks_pack_start..];            while !slice.is_empty() {                let current_pack_id = slice[0].entry_pack_location.as_ref().expect("packed object").pack_id;                let pack_end = slice.partition_point(|e| {                    e.entry_pack_location.as_ref().expect("packed object").pack_id == current_pack_id                });                index.push((current_pack_id, chunks_pack_start..chunks_pack_start + pack_end));                slice = &slice[pack_end..];                chunks_pack_start += pack_end;            }             progress.set(counts.len());            progress.show_throughput(start);             index        }    };    let counts = Arc::new(counts);    let progress = Arc::new(parking_lot::Mutex::new(progress));     parallel::reduce::Stepwise::new(        chunks.enumerate(),        thread_limit,        {            let progress = Arc::clone(&progress);            move |n| {                (                    Vec::new(),   // object data buffer                    make_cache(), // cache to speed up pack operations                    progress.lock().add_child(format!("thread {}", n)),                )            }        },        {            let counts = Arc::clone(&counts);            move |(chunk_id, chunk_range): (ChunkId, std::ops::Range<usize>), (buf, cache, progress)| {                let mut out = Vec::new();                let chunk = &counts[chunk_range];                let mut stats = Outcome::default();                let mut pack_offsets_to_id = None;                progress.init(Some(chunk.len()), git_features::progress::count("objects"));                 for count in chunk.iter() {                    out.push(match count                        .entry_pack_location                        .as_ref()                        .and_then(|l| db.entry_by_location(l).map(|pe| (l, pe)))                    {                        Some((location, pack_entry)) => {                            if let Some((cached_pack_id, _)) = &pack_offsets_to_id {                                if *cached_pack_id != location.pack_id {                                    pack_offsets_to_id = None;                                }                            }                            let pack_range = counts_range_by_pack_id[counts_range_by_pack_id                                .binary_search_by_key(&location.pack_id, |e| e.0)                                .expect("pack-id always present")]                            .1                            .clone();                            let base_index_offset = pack_range.start;                            let counts_in_pack = &counts[pack_range];                            match output::Entry::from_pack_entry(                                pack_entry,                                count,                                counts_in_pack,                                base_index_offset,                                allow_thin_pack.then(|| {                                    |pack_id, base_offset| {                                        let (cached_pack_id, cache) = pack_offsets_to_id.get_or_insert_with(|| {                                            db.bundle_by_pack_id(pack_id)                                                .map(|b| {                                                    let mut v = b                                                        .index                                                        .iter()                                                        .map(|e| (e.pack_offset, e.oid))                                                        .collect::<Vec<_>>();                                                    v.sort_by_key(|e| e.0);                                                    (pack_id, v)                                                })                                                .expect("pack used for counts is still available")                                        });                                        debug_assert_eq!(*cached_pack_id, pack_id);                                        stats.ref_delta_objects += 1;                                        cache                                            .binary_search_by_key(&base_offset, |e| e.0)                                            .ok()                                            .map(|idx| cache[idx].1)                                    }                                }),                                version,                            ) {                                Some(entry) => {                                    stats.objects_copied_from_pack += 1;                                    entry                                }                                None => match db.try_find(count.id, buf, cache).map_err(Error::FindExisting)? {                                    Some(obj) => {                                        stats.decoded_and_recompressed_objects += 1;                                        output::Entry::from_data(count, &obj)                                    }                                    None => {                                        stats.missing_objects += 1;                                        Ok(output::Entry::invalid())                                    }                                },                            }                        }                        None => match db.try_find(count.id, buf, cache).map_err(Error::FindExisting)? {                            Some(obj) => {                                stats.decoded_and_recompressed_objects += 1;                                output::Entry::from_data(count, &obj)                            }                            None => {                                stats.missing_objects += 1;                                Ok(output::Entry::invalid())                            }                        },                    }?);                    progress.inc();                }                Ok((chunk_id, out, stats))            }        },        reduce::Statistics::default(),    )} mod util {    #[derive(Clone)]    pub struct ChunkRanges {        cursor: usize,        size: usize,        len: usize,    }     impl ChunkRanges {        pub fn new(size: usize, total: usize) -> Self {            ChunkRanges {                cursor: 0,                size,                len: total,            }        }    }     impl Iterator for ChunkRanges {        type Item = std::ops::Range<usize>;         fn next(&mut self) -> Option<Self::Item> {            if self.cursor >= self.len {                None            } else {                let upper = (self.cursor + self.size).min(self.len);                let range = self.cursor..upper;                self.cursor = upper;                Some(range)            }        }    }} mod reduce {    use std::marker::PhantomData;     use git_features::parallel;     use super::{ChunkId, Outcome};    use crate::data::output;     pub struct Statistics<E> {        total: Outcome,        _err: PhantomData<E>,    }     impl<E> Default for Statistics<E> {        fn default() -> Self {            Statistics {                total: Default::default(),                _err: PhantomData::default(),            }        }    }     impl<Error> parallel::Reduce for Statistics<Error> {        type Input = Result<(ChunkId, Vec<output::Entry>, Outcome), Error>;        type FeedProduce = (ChunkId, Vec<output::Entry>);        type Output = Outcome;        type Error = Error;         fn feed(&mut self, item: Self::Input) -> Result<Self::FeedProduce, Self::Error> {            item.map(|(cid, entries, stats)| {                self.total.aggregate(stats);                (cid, entries)            })        }         fn finalize(self) -> Result<Self::Output, Self::Error> {            Ok(self.total)        }    }} mod types {    use crate::data::output::entry;     /// Information gathered during the run of [`iter_from_counts()`][super::iter_from_counts()].    #[derive(Default, PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone, Copy)]    #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]    pub struct Outcome {        /// The amount of fully decoded objects. These are the most expensive as they are fully decoded.        pub decoded_and_recompressed_objects: usize,        /// The amount of objects that could not be located despite them being mentioned during iteration        pub missing_objects: usize,        /// The amount of base or delta objects that could be copied directly from the pack. These are cheapest as they        /// only cost a memory copy for the most part.        pub objects_copied_from_pack: usize,        /// The amount of objects that ref to their base as ref-delta, an indication for a thin back being created.        pub ref_delta_objects: usize,    }     impl Outcome {        pub(in crate::data::output::entry) fn aggregate(            &mut self,            Outcome {                decoded_and_recompressed_objects: decoded_objects,                missing_objects,                objects_copied_from_pack,                ref_delta_objects,            }: Self,        ) {            self.decoded_and_recompressed_objects += decoded_objects;            self.missing_objects += missing_objects;            self.objects_copied_from_pack += objects_copied_from_pack;            self.ref_delta_objects += ref_delta_objects;        }    }     /// The way the iterator operates.    #[derive(PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone, Copy)]    #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]    pub enum Mode {        /// Copy base objects and deltas from packs, while non-packed objects will be treated as base objects        /// (i.e. without trying to delta compress them). This is a fast way of obtaining a back while benefitting        /// from existing pack compression and spending the smallest possible time on compressing unpacked objects at        /// the cost of bandwidth.        PackCopyAndBaseObjects,    }     /// Configuration options for the pack generation functions provied in [this module][crate::data::output].    #[derive(PartialEq, Eq, Debug, Hash, Ord, PartialOrd, Clone, Copy)]    #[cfg_attr(feature = "serde1", derive(serde::Serialize, serde::Deserialize))]    pub struct Options {        /// The amount of threads to use at most when resolving the pack. If `None`, all logical cores are used.        pub thread_limit: Option<usize>,        /// The algorithm to produce a pack        pub mode: Mode,        /// If set, the resulting back can have deltas that refer to an object which is not in the pack. This can happen        /// if the initial counted objects do not contain an object that an existing packed delta refers to, for example, because        /// it wasn't part of the iteration, for instance when the iteration was performed on tree deltas or only a part of the        /// commit graph. Please note that thin packs are not valid packs at rest, thus they are only valid for packs in transit.        ///        /// If set to false, delta objects will be decompressed and recompressed as base objects.        pub allow_thin_pack: bool,        /// The amount of objects per chunk or unit of work to be sent to threads for processing        /// TODO: could this become the window size?        pub chunk_size: usize,        /// The pack data version to produce for each entry        pub version: crate::data::Version,    }     impl Default for Options {        fn default() -> Self {            Options {                thread_limit: None,                mode: Mode::PackCopyAndBaseObjects,                allow_thin_pack: false,                chunk_size: 10,                version: Default::default(),            }        }    }     /// The error returned by the pack generation function [`to_entry_iter()`][crate::data::output::entry::iter_from_counts()].    #[derive(Debug, thiserror::Error)]    #[allow(missing_docs)]    pub enum Error<FindErr>    where        FindErr: std::error::Error + 'static,    {        #[error(transparent)]        FindExisting(FindErr),        #[error(transparent)]        NewEntry(#[from] entry::Error),    }}pub use types::{Error, Mode, Options, Outcome};