OXYGENHUB

Store a large change's texts a batch at a time 17f94b13ae54

Encoding a change made every stored insert's text blob before any of it
was stored, so a change adding a whole tree held all its text a second
time (the first commit of the Linux tree peaked at 7.4 GB). The encoder
now hands each batch of 256 texts to a sink as it makes them
(Change::encode_streaming); the store writes their chunks and blobs at
once (Store::put_texts) and the batch is dropped. put_change and the
snapshot use it. The first commit of the Linux tip now peaks at 4.8 GB
(5.8 GB RSS, from 8.7 GB) in 41.6 s.

refs #35
Author
Andy McCall @andymccall <[email protected]>
When
· 2026-09-29 23:30 UTC
Signature
✔ signed by Andy McCall
Hash
17f94b13ae54264c322cce42947a313d81e11df9a85601320ff3ebb8bd152737
Depends on
192fad1e46a2 1d360c91a289 6f8aeb996558 8b276e0106d1 982864068a59 a4fad5b085c7 ff0af1ee299d
Depended on by
5b4c51307b14 cd4eb55e3b3d 0f93f49ce059 01a47cd7f0c8
In bookmarks
agents-in-readme bare-hashes carried-not-removed carry-identical doctor-trust identical-sides keyless-tags main merge-remote-bookmark only-help-103 paths-for-resolutions-and-amend push-notes show-stat subrepo-pin-state switch-says
In tags
v0.1.0 v0.10.0 v0.11.0 v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.4.1 v0.4.2 v0.4.3 v0.4.4 v0.4.5 v0.4.6 v0.5.0 v0.5.1 v0.6.0 v0.7.0 v0.8.0 v0.9.0
Edits
20 atoms, 0 tree operations, 4 files

Browse the tree at this change

modified crates/oxy-core/src/objects.rs +41 −9
…
636 unchanged lines
637 637
    }
638 638
639 639
    pub fn to_value(&self) -> Value {
640 -
        self.value_with(&mut None)
640 +
        self.value_with(&mut |_| true)
641 641
    }
642 642
643 643
    /// The text blob of each edit stored as one, `None` for the others;
644 644
    /// the stored texts go to `texts`, in order, when it is given. Worked
645 645
    /// out on every core, a bounded batch at a time: chunking and hashing
646 646
    /// a whole tree's text is most of encoding a change that adds it.
647 -
    fn stored_texts(&self, texts: &mut Option<&mut Vec<StoredText>>) -> Vec<Option<ObjectRef>> {
647 +
    fn stored_texts(&self, sink: &mut TextSink) -> Vec<Option<ObjectRef>> {
648 648
        use rayon::prelude::*;
649 649
        let mut refs = vec![None; self.edits.len()];
650 650
        let stored: Vec<usize> = (0..self.edits.len())
…
13 unchanged lines
664 664
                    }
665 665
                })
666 666
                .collect();
667 -
            for (i, t) in batch.iter().zip(made) {
667 +
            for (i, t) in batch.iter().zip(&made) {
668 668
                refs[*i] = Some(t.blob.hash());
669 -
                if let Some(sink) = texts {
670 -
                    sink.push(t);
671 -
                }
669 +
            }
670 +
            if !sink(made) {
671 +
                break;
672 672
            }
673 673
        }
674 674
        refs
675 675
    }
676 676
677 -
    fn value_with(&self, texts: &mut Option<&mut Vec<StoredText>>) -> Value {
677 +
    fn value_with(&self, sink: &mut TextSink) -> Value {
678 678
        Value::map([
679 679
            (0, Value::Uint(self.version)),
680 680
            (1, Value::Bytes(self.change_id.0.to_vec())),
…
18 unchanged lines
699 699
                Value::Array(
700 700
                    self.edits
701 701
                        .iter()
702 -
                        .zip(self.stored_texts(texts))
702 +
                        .zip(self.stored_texts(sink))
703 703
                        .map(|(e, stored)| e.to_value(stored))
704 704
                        .collect(),
705 705
                ),
…
19 unchanged lines
725 725
            "deps must be normalized"
726 726
        );
727 727
        let mut texts = Vec::new();
728 -
        let bytes = cbor::encode(&self.value_with(&mut Some(&mut texts)));
728 +
        let bytes = cbor::encode(&self.value_with(&mut |batch| {
729 +
            texts.extend(batch);
730 +
            true
731 +
        }));
729 732
        (bytes, texts)
733 +
    }
734 +
735 +
    /// The encoding, handing the stored texts to `store` a batch at a time
736 +
    /// as they are made, so a change adding a whole tree never holds all
737 +
    /// its text twice. The first error from `store` is returned.
738 +
    pub fn encode_streaming<E>(
739 +
        &self,
740 +
        mut store: impl FnMut(Vec<StoredText>) -> Result<(), E>,
741 +
    ) -> Result<Vec<u8>, E> {
742 +
        debug_assert!(
743 +
            self.deps.windows(2).all(|w| w[0] < w[1]),
744 +
            "deps must be normalized"
745 +
        );
746 +
        let mut failed = None;
747 +
        let value = self.value_with(&mut |batch| match store(batch) {
748 +
            Ok(()) => true,
749 +
            Err(e) => {
750 +
                failed = Some(e);
751 +
                false
752 +
            }
753 +
        });
754 +
        match failed {
755 +
            Some(e) => Err(e),
756 +
            None => Ok(cbor::encode(&value)),
757 +
        }
730 758
    }
731 759
732 760
    /// Whether any insert is stored as a text blob.
…
82 unchanged lines
815 843
/// A resolver may also return no lines, leaving a placeholder for a reader
816 844
/// that fills it in later (history-only states, spec §11).
817 845
pub type Resolver<'a> = dyn FnMut(&ObjectRef, u32) -> Result<Vec<Vec<u8>>, ObjectError> + 'a;
846 +
847 +
/// Receives the stored texts of a change being encoded, a batch at a
848 +
/// time; returns false to stop making more.
849 +
type TextSink<'a> = dyn FnMut(Vec<StoredText>) -> bool + 'a;
818 850
819 851
/// Where a history-only load finds an insert's lines when it builds that
820 852
/// file's graph (spec §11): it keeps no line until then.
…
644 unchanged lines
modified crates/oxy-core/src/repo/snapshot.rs +5 −3
…
395 unchanged lines
396 396
        let statuses = built.statuses;
397 397
        // Empty snapshot: hash the candidate before storing; only record an op on change.
398 398
        let record = crate::timing::phase("snapshot: hash and record");
399 -
        // Encoded once: the hash and, when it changed, what is stored.
400 -
        let (encoded, texts) = wc.encode_parts();
399 +
        // Encoded once, its texts stored a batch at a time as they are
400 +
        // made (a stored blob is not stored again); the change itself only
401 +
        // when it changed.
402 +
        let encoded = wc.encode_streaming(|batch| self.store.put_texts(&batch))?;
401 403
        let new_hash = crate::hash::hash_object(crate::hash::ObjectType::Change, &encoded);
402 404
        let changed = new_hash != old_hash;
403 405
        if changed {
404 -
            self.store.put_change_encoded(&encoded, &texts)?;
406 +
            self.store.put(crate::hash::ObjectType::Change, &encoded)?;
405 407
            let mut view = op.view.clone();
406 408
            view.workspaces.insert(self.workspace, new_hash);
407 409
            self.record_op(&[op_ref], &format!("{command}{SNAPSHOT_SUFFIX}"), view)?;
…
1950 unchanged lines
modified crates/oxy-core/src/store.rs +6 −11
…
969 unchanged lines
970 970
    /// Store a change and the text blobs (and their chunks) its large
971 971
    /// inserts are stored as (spec §6.1).
972 972
    pub fn put_change(&self, c: &Change) -> Result<ObjectRef> {
973 -
        let (bytes, texts) = c.encode_parts();
974 -
        self.put_change_encoded(&bytes, &texts)
973 +
        let bytes = c.encode_streaming(|batch| self.put_texts(&batch))?;
974 +
        self.put(ObjectType::Change, &bytes)
975 975
    }
976 976
977 -
    /// Store a change already encoded by [`Change::encode_parts`].
978 -
    pub fn put_change_encoded(
979 -
        &self,
980 -
        bytes: &[u8],
981 -
        texts: &[crate::objects::StoredText],
982 -
    ) -> Result<ObjectRef> {
983 -
        // Every text's chunks at once (compressed on every core in a large
984 -
        // batch), then the blobs and the change.
977 +
    /// Store text blobs and their chunks: every chunk at once (compressed
978 +
    /// on every core in a large batch), then the blobs.
979 +
    pub fn put_texts(&self, texts: &[crate::objects::StoredText]) -> Result<()> {
985 980
        let mut chunks: Vec<(ObjectType, &[u8])> = Vec::new();
986 981
        for t in texts {
987 982
            if !self.has(&t.blob.hash()) {
…
8 unchanged lines
996 991
        for t in texts {
997 992
            self.put(ObjectType::Blob, &t.blob.encode())?;
998 993
        }
999 -
        self.put(ObjectType::Change, bytes)
994 +
        Ok(())
1000 995
    }
1001 996
1002 997
    pub fn get_change(&self, r: &ObjectRef) -> Result<Change> {
…
838 unchanged lines
modified docs/benchmarks.md +5 −2
…
382 unchanged lines
383 383
|---|---:|---:|---:|
384 384
| Before | 61 s | 7.1 GB | 10.1 GB |
385 385
| After | 40.7 s | 7.4 GB | 8.7 GB |
386 +
| Texts stored a batch at a time while encoding | 41.6 s | 4.8 GB | 5.8 GB |
386 387
| Git | 33.7 to 37.9 s | | 42 MB |
387 388
388 -
The peak is now in storing the change, which holds every text's content
389 -
at once; the lines themselves (a Vec each) are the rest.
389 +
Storing a change's texts a batch at a time as they are made (the
390 +
encoder hands each batch of 256 to the store) removed the storing peak.
391 +
The peak is now the working copy's content (1.6 GB) and its lines, a Vec
392 +
each (2.6 GB): M2.
390 393