Move changes as stored in a fetch on one machine 2c75672fb2fe
A local fetch decoded every change's text on the sending side (1.8 GB of lines for the Linux tree's first change) and the import encoded it again: chunking, hashing and compressing it all. A fetch between repositories on one machine now carries changes as they are stored (Bundle::raw_changes, never on the wire; encoded as they are if it ever is), with their text blobs and chunks among the objects. The import stores the objects at once (compressed on every core), checks each raw change's dependencies from its skeleton and that its text blobs came, and stores its bytes as they are. A test covers the refusal of a change whose text did not come. A local clone of the Linux tip takes 48.2 s and 3.9 GB (from 56.8 s and 4.7 GB). refs #35
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-09-30 00:01 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
2c75672fb2fe0ff6bfe8be8f2230ba153204b7422615abccc6aefa24380bbe2e- Depends on
- 132b884b9a8d 1d360c91a289 519c7c061bcb 598009a75f6c 5b9c6ad45f6e 6f8aeb996558 750251b21a96 75167672ebcd 982864068a59 b3c5a59a875a c5ce0367eddf
- Depended on by
- 2f0b4b0651ea d36ba2f79f95 f0a517d28237 0f93f49ce059 a87c7b9d563a
- 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
- 19 atoms, 0 tree operations, 4 files
Browse the tree at this change
modified crates/oxy-core/src/protocol.rs +15 −2
| … | 89 unchanged lines |
||
| 90 | 90 | /// the bundle knows it: the store it came from, or the bytes it was |
|
| 91 | 91 | /// decoded from. Never read from the wire. Empty when unknown. |
|
| 92 | 92 | pub hashes: Vec<ChangeHash>, |
|
| 93 | + | /// Changes as stored, with their signatures, in dependency order, for a |
|
| 94 | + | /// fetch between repositories on one machine: their text blobs and |
|
| 95 | + | /// chunks are among `objects`. Never on the wire. |
|
| 96 | + | pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>, |
|
| 93 | 97 | } |
|
| 94 | 98 | ||
| 95 | 99 | impl Bundle { |
|
| … | 77 unchanged lines |
||
| 173 | 177 | self.repo.negotiate(have, want) |
|
| 174 | 178 | } |
|
| 175 | 179 | fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> { |
|
| 176 | - | self.repo.bundle_for_filtered(changes, max_blob) |
|
| 180 | + | self.repo.bundle_raw_for(changes, max_blob) |
|
| 177 | 181 | } |
|
| 178 | 182 | fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> { |
|
| 179 | 183 | self.repo.objects_for(refs) |
|
| … | 362 unchanged lines |
||
| 542 | 546 | /// §6.1) are added to its objects when they are not already there, so a |
|
| 543 | 547 | /// bundle can always be read on its own. |
|
| 544 | 548 | pub fn bundle_value(b: &Bundle) -> Value { |
|
| 545 | - | let mut changes = Vec::with_capacity(b.changes.len()); |
|
| 549 | + | let mut changes = Vec::with_capacity(b.raw_changes.len() + b.changes.len()); |
|
| 550 | + | // Changes as stored are already their encoding, their texts among |
|
| 551 | + | // the objects. |
|
| 552 | + | for (bytes, s) in &b.raw_changes { |
|
| 553 | + | changes.push(Value::Array(vec![ |
|
| 554 | + | Value::Bytes(bytes.clone()), |
|
| 555 | + | sigs_value(s), |
|
| 556 | + | ])); |
|
| 557 | + | } |
|
| 546 | 558 | let mut texts: Vec<(ObjectType, Vec<u8>)> = Vec::new(); |
|
| 547 | 559 | let mut listed: Option<std::collections::HashSet<ObjectRef>> = None; |
|
| 548 | 560 | for (c, s) in &b.changes { |
|
| … | 425 unchanged lines |
||
| 974 | 986 | sigs: vec![psig.clone()], |
|
| 975 | 987 | }], |
|
| 976 | 988 | hashes: vec![], |
|
| 989 | + | raw_changes: vec![], |
|
| 977 | 990 | }; |
|
| 978 | 991 | let d = wire::decode_bundle(&wire::encode_bundle(&bundle)).unwrap(); |
|
| 979 | 992 | assert_eq!( |
|
| … | 135 unchanged lines |
modified crates/oxy-core/src/repo/sync.rs +93 −13
| … | 270 unchanged lines |
||
| 271 | 271 | changes: &[ChangeHash], |
|
| 272 | 272 | max_blob: Option<u64>, |
|
| 273 | 273 | ) -> Result<Bundle> { |
|
| 274 | + | self.bundle_of(changes, max_blob, false) |
|
| 275 | + | } |
|
| 276 | + | ||
| 277 | + | /// Like [`Repo::bundle_for_filtered`] for a fetch between repositories |
|
| 278 | + | /// on one machine: changes as they are stored ([`Bundle::raw_changes`]), |
|
| 279 | + | /// with their text blobs and chunks among the objects, so neither side |
|
| 280 | + | /// decodes or encodes their text. |
|
| 281 | + | pub fn bundle_raw_for(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> { |
|
| 282 | + | self.bundle_of(changes, max_blob, true) |
|
| 283 | + | } |
|
| 284 | + | ||
| 285 | + | fn bundle_of( |
|
| 286 | + | &self, |
|
| 287 | + | changes: &[ChangeHash], |
|
| 288 | + | max_blob: Option<u64>, |
|
| 289 | + | raw: bool, |
|
| 290 | + | ) -> Result<Bundle> { |
|
| 274 | 291 | let mut bundle = Bundle::default(); |
|
| 275 | 292 | let mut objects: BTreeSet<ObjectRef> = BTreeSet::new(); |
|
| 276 | 293 | let mut seals: BTreeSet<ObjectRef> = BTreeSet::new(); |
|
| … | 3 unchanged lines |
||
| 280 | 297 | } |
|
| 281 | 298 | }; |
|
| 282 | 299 | for h in changes { |
|
| 283 | - | let c = self.store.get_change(h)?; |
|
| 300 | + | // Raw: the stored bytes, and a skeleton for the rest. |
|
| 301 | + | let (c, encoded) = if raw { |
|
| 302 | + | let bytes = self.store.get_typed(h, ObjectType::Change)?; |
|
| 303 | + | let (skel, texts) = Change::decode_skeleton(&bytes, *h)?; |
|
| 304 | + | for t in texts { |
|
| 305 | + | if let crate::objects::TextRef::Blob(r, _) = t { |
|
| 306 | + | let blob = Blob::decode(&self.store.get_typed(&r, ObjectType::Blob)?)?; |
|
| 307 | + | objects.insert(r); |
|
| 308 | + | objects.extend(blob.chunks); |
|
| 309 | + | } |
|
| 310 | + | } |
|
| 311 | + | (skel, Some(bytes)) |
|
| 312 | + | } else { |
|
| 313 | + | (self.store.get_change(h)?, None) |
|
| 314 | + | }; |
|
| 284 | 315 | let sigs = self.store.signatures(h)?; |
|
| 285 | 316 | add_object(&c.author, &mut objects); |
|
| 286 | 317 | for s in &sigs { |
|
| … | 18 unchanged lines |
||
| 305 | 336 | } |
|
| 306 | 337 | for s in self.store.seals_for(h)? { |
|
| 307 | 338 | seals.insert(s); |
|
| 339 | + | } |
|
| 340 | + | match encoded { |
|
| 341 | + | Some(bytes) => bundle.raw_changes.push((bytes, sigs)), |
|
| 342 | + | None => { |
|
| 343 | + | bundle.changes.push((c, sigs)); |
|
| 344 | + | bundle.hashes.push(*h); |
|
| 345 | + | } |
|
| 308 | 346 | } |
|
| 309 | - | bundle.changes.push((c, sigs)); |
|
| 310 | - | bundle.hashes.push(*h); |
|
| 311 | 347 | } |
|
| 312 | 348 | for r in seals { |
|
| 313 | 349 | let attest = self.store.get_attest(&r)?; |
|
| … | 40 unchanged lines |
||
| 354 | 390 | strict: bool, |
|
| 355 | 391 | shallow: bool, |
|
| 356 | 392 | ) -> Result<Vec<ChangeHash>> { |
|
| 357 | - | for (ty, body) in &bundle.objects { |
|
| 358 | - | match ty { |
|
| 393 | + | for (ty, _) in &bundle.objects { |
|
| 394 | + | if !matches!( |
|
| 395 | + | ty, |
|
| 359 | 396 | ObjectType::Chunk |
|
| 360 | - | | ObjectType::Blob |
|
| 361 | - | | ObjectType::Identity |
|
| 362 | - | | ObjectType::State |
|
| 363 | - | | ObjectType::Checkpoint => { |
|
| 364 | - | self.store.put(*ty, body)?; |
|
| 365 | - | } |
|
| 366 | - | other => return fail(format!("bundle may not carry {} objects", other.name())), |
|
| 397 | + | | ObjectType::Blob |
|
| 398 | + | | ObjectType::Identity |
|
| 399 | + | | ObjectType::State |
|
| 400 | + | | ObjectType::Checkpoint |
|
| 401 | + | ) { |
|
| 402 | + | return fail(format!("bundle may not carry {} objects", ty.name())); |
|
| 367 | 403 | } |
|
| 368 | 404 | } |
|
| 405 | + | // All at once: in a large batch they are compressed on every core. |
|
| 406 | + | let objects: Vec<(ObjectType, &[u8])> = bundle |
|
| 407 | + | .objects |
|
| 408 | + | .iter() |
|
| 409 | + | .map(|(ty, body)| (*ty, body.as_slice())) |
|
| 410 | + | .collect(); |
|
| 411 | + | self.store.put_many(&objects)?; |
|
| 369 | 412 | for sa in &bundle.attests { |
|
| 370 | 413 | let r = self.store.put_attest(&sa.attest)?; |
|
| 371 | 414 | for sig in &sa.sigs { |
|
| … | 13 unchanged lines |
||
| 385 | 428 | } else { |
|
| 386 | 429 | bundle.changes.iter().map(|(c, _)| c.hash()).collect() |
|
| 387 | 430 | }; |
|
| 388 | - | let incoming: BTreeSet<ChangeHash> = hashes.iter().copied().collect(); |
|
| 431 | + | let raw_hashes: Vec<ChangeHash> = bundle |
|
| 432 | + | .raw_changes |
|
| 433 | + | .iter() |
|
| 434 | + | .map(|(b, _)| hash_object(ObjectType::Change, b)) |
|
| 435 | + | .collect(); |
|
| 436 | + | let incoming: BTreeSet<ChangeHash> = hashes.iter().chain(&raw_hashes).copied().collect(); |
|
| 389 | 437 | let boundary = self.boundary(); |
|
| 390 | 438 | let mut imported = Vec::new(); |
|
| 439 | + | // Changes as stored: checked, then stored as they are, their text |
|
| 440 | + | // blobs having come with the objects. |
|
| 441 | + | for ((bytes, sigs), h) in bundle.raw_changes.iter().zip(raw_hashes) { |
|
| 442 | + | let (skel, texts) = Change::decode_skeleton(bytes, h)?; |
|
| 443 | + | for d in &skel.deps { |
|
| 444 | + | if !self.store.has(d) && !incoming.contains(d) && !shallow && !boundary.contains(d) |
|
| 445 | + | { |
|
| 446 | + | return fail(format!( |
|
| 447 | + | "change {} depends on {}, which is neither stored nor in the bundle", |
|
| 448 | + | h.short(12), |
|
| 449 | + | d.short(12) |
|
| 450 | + | )); |
|
| 451 | + | } |
|
| 452 | + | } |
|
| 453 | + | for t in texts { |
|
| 454 | + | if let crate::objects::TextRef::Blob(r, _) = t { |
|
| 455 | + | if !self.store.has(&r) { |
|
| 456 | + | return fail(format!( |
|
| 457 | + | "change {} came without its text blob {}", |
|
| 458 | + | h.short(12), |
|
| 459 | + | r.short(12) |
|
| 460 | + | )); |
|
| 461 | + | } |
|
| 462 | + | } |
|
| 463 | + | } |
|
| 464 | + | self.store.put(ObjectType::Change, bytes)?; |
|
| 465 | + | for sig in sigs { |
|
| 466 | + | self.store.put_signature(&h, sig)?; |
|
| 467 | + | } |
|
| 468 | + | imported.push(h); |
|
| 469 | + | } |
|
| 391 | 470 | for ((c, sigs), h) in bundle.changes.iter().zip(hashes) { |
|
| 392 | 471 | for d in &c.deps { |
|
| 393 | 472 | if !self.store.has(d) && !incoming.contains(d) && !shallow && !boundary.contains(d) |
|
| … | 1118 unchanged lines |
||
| 1512 | 1591 | if !extra.is_empty() { |
|
| 1513 | 1592 | let more = self.bundle_for(&extra)?; |
|
| 1514 | 1593 | bundle.changes.extend(more.changes); |
|
| 1594 | + | bundle.hashes.extend(more.hashes); |
|
| 1515 | 1595 | bundle.attests.extend(more.attests); |
|
| 1516 | 1596 | bundle.objects.extend(more.objects); |
|
| 1517 | 1597 | } |
|
| … | 142 unchanged lines |
modified crates/oxy-core/src/repo/tests.rs +24 −0
| … | 5914 unchanged lines |
||
| 5915 | 5915 | .statuses |
|
| 5916 | 5916 | .is_empty()); |
|
| 5917 | 5917 | } |
|
| 5918 | + | ||
| 5919 | + | /// A fetch on one machine moves changes as stored, text blobs among the |
|
| 5920 | + | /// objects; a change whose text blob did not come is refused, not stored. |
|
| 5921 | + | #[test] |
|
| 5922 | + | fn a_raw_bundle_without_a_changes_text_is_refused() { |
|
| 5923 | + | hermetic(); |
|
| 5924 | + | let dir = tempfile::tempdir().unwrap(); |
|
| 5925 | + | let root = dir.path(); |
|
| 5926 | + | let a = Repo::init(&root.join("a")).unwrap(); |
|
| 5927 | + | let big: String = (0..3000).map(|i| format!("line {i}\n")).collect(); |
|
| 5928 | + | write(&root.join("a"), "big.txt", big.as_bytes()); |
|
| 5929 | + | let h = a.commit("big", &[]).unwrap(); |
|
| 5930 | + | let bundle = a.bundle_raw_for(&[h], None).unwrap(); |
|
| 5931 | + | assert_eq!(bundle.raw_changes.len(), 1); |
|
| 5932 | + | assert!(bundle.changes.is_empty()); |
|
| 5933 | + | let b = Repo::init(&root.join("b")).unwrap(); |
|
| 5934 | + | let mut without = bundle.clone(); |
|
| 5935 | + | without.objects.retain(|(ty, _)| *ty != ObjectType::Blob); |
|
| 5936 | + | let e = b.import_bundle(&without, false).unwrap_err(); |
|
| 5937 | + | assert!(e.to_string().contains("came without its text blob"), "{e}"); |
|
| 5938 | + | assert!(!b.store().has(&h)); |
|
| 5939 | + | assert_eq!(b.import_bundle(&bundle, false).unwrap(), vec![h]); |
|
| 5940 | + | assert_eq!(b.store().get_change(&h).unwrap().message, "big"); |
|
| 5941 | + | } |
modified docs/benchmarks.md +7 −3
| … | 410 unchanged lines |
||
| 411 | 411 | | Before | 96 s (81.6 s in a quieter run) | 5.2 to 5.4 GB | |
|
| 412 | 412 | | After | 61.0 s | 4.1 GB | |
|
| 413 | 413 | | The change read once for all batches | 56.8 s | 4.7 GB | |
|
| 414 | + | | Changes moved as stored (load average 5.2) | 48.2 s | 3.9 GB | |
|
| 414 | 415 | | Git | 9.6 s | 469 MB | |
|
| 415 | 416 | ||
| 416 | - | The checkout (31 s) now spends most of its time building graphs: each |
|
| 417 | - | batch of 2 048 files reads the 81 MiB first change again for its inline |
|
| 418 | - | inserts. |
|
| 417 | + | A fetch between repositories on one machine now moves changes as they |
|
| 418 | + | are stored, their text blobs and chunks among the objects, instead of |
|
| 419 | + | decoding every change's text on one side and chunking, hashing and |
|
| 420 | + | compressing it again on the other; the import stores all the objects at |
|
| 421 | + | once, compressed on every core, and refuses a change whose text blob did |
|
| 422 | + | not come. |
|
| 419 | 423 |