fsck ahead of Git: objects checked in parallel, graphs built with their lines in parallel, a few changes kept (refs #41) acadb5ed28a1

        use rayon::prelude::*;

        // A chunk at a time, read (and verified) and decoded on every core.

        for refs in self.store.list()?.chunks(8192) {

            let checked: Vec<std::result::Result<(), String>> = self

                .store

                .get_many(refs)

                .into_par_iter()

                .map(|got| match got {

                    Ok((ty, body)) => decodes(ty, &body)

                        .map_err(|e| format!("{} object does not decode: {e}", ty.name())),

                    Err(e) => Err(format!("unreadable object: {e}")),

                })

                .collect();

            for (r, c) in refs.iter().zip(checked) {

                rep.objects += 1;

                if let Err(e) = c {

                    rep.error(Some(*r), e);

                    bad.insert(*r);

/// The changes whose encodings are kept for building graphs: a batch of

/// files often takes inline text from a few changes in turn.

const CHANGE_CACHE_COUNT: usize = 4;

const CHANGE_CACHE_BYTES: usize = 512 << 20;

    /// The encodings of the changes last read for their inline inserts,

    /// oldest first.

    last_change: std::cell::RefCell<Vec<(ChangeHash, std::sync::Arc<Vec<u8>>)>>,

    /// last few read are kept: a checkout building graphs a batch at a

    /// time reads the same large changes for every batch, often in turn.

        let mut cache = self.last_change.borrow_mut();

        if let Some(i) = cache.iter().position(|(k, _)| k == h) {

            let hit = cache.remove(i);

            let b = hit.1.clone();

            cache.push(hit);

            return Ok(b);

        cache.push((*h, b.clone()));

        // The last few, within a memory budget; the newest always stays.

        while cache.len() > CHANGE_CACHE_COUNT

            || (cache.len() > 1

                && cache.iter().map(|(_, b)| b.len()).sum::<usize>() > CHANGE_CACHE_BYTES)

        {

            cache.remove(0);

        }

    let lines = crate::timing::phase("graphs: hand out blobs");

    // Hand each graph its own blobs, in order, then fill in its inserts'

    // lines and build it, all in parallel: graphs are independent.

    type Work = (NodeId, DeferredBody, Option<Vec<u8>>, Vec<Vec<u8>>);

    let mut work: Vec<Work> = Vec::with_capacity(taken.len());

    for (node, d) in taken.drain(..) {

        let wanted = d

            .texts

            .iter()

            .filter(|t| matches!(t, TextRef::Blob(..)))

            .count();

        let mine: Vec<Vec<u8>> = texts.by_ref().take(wanted).collect();

        if mine.len() != wanted {

            return Err(ObjectError::Schema("text blobs: wrong count".into()));

        work.push((node, d, bytes, mine));

    let _t = crate::timing::phase("graphs: lines and build");

    let changes = &changes;

        .map(|(node, mut d, bytes, mine)| {

            // Fill in the lines of inserts read without them, in order.

            let mut want = std::mem::take(&mut d.texts).into_iter();

            let mut mine = mine.into_iter();

            for (_, edits) in d.edits.iter_mut() {

                for e in edits.iter_mut() {

                    if let Edit::Insert { lines, .. } = e {

                        if lines.is_empty() {

                            *lines = match want.next() {

                                Some(TextRef::Blob(_, count, form)) => {

                                    let content = mine.next().ok_or_else(|| {

                                        ObjectError::Schema("text blobs: wrong count".into())

                                    })?;

                                    crate::objects::text_lines(&content, count, form)?

                                }

                                Some(TextRef::Inline { change, start, end }) => {

                                    let bytes = &changes[&change];

                                    let range = bytes

                                        .get(start as usize..end as usize)

                                        .ok_or_else(|| {

                                            ObjectError::Schema(

                                                "an inline insert lies outside its change".into(),

                                            )

                                        })?;

                                    crate::objects::inline_lines(range)?

                                }

                                None => {

                                    return Err(ObjectError::Schema(

                                        "an insert's text is missing".into(),

                                    ))

                                }

                            };

                        }

                    }

                }

            }