git_internal/internal/pack/
cache_object.rs

1//! Cache object representation plus disk-backed serialization helpers used by the pack decoder to
2//! bound memory while still serving delta reconstruction quickly.
3
4use std::{
5    borrow::Cow,
6    fs, io,
7    io::Write,
8    ops::Deref,
9    path::{Path, PathBuf},
10    sync::{
11        Arc,
12        atomic::{AtomicBool, AtomicUsize, Ordering},
13    },
14};
15
16use lru_mem::{HeapSize, MemSize};
17use rkyv::{
18    Archive, Deserialize as RkyvDeserialize, Serialize as RkyvSerialize,
19    api::high::{HighSerializer, HighValidator},
20    bytecheck::CheckBytes,
21    de::Pool,
22    rancor::{Error as RkyvError, Strategy},
23    ser::allocator::ArenaHandle,
24    util::AlignedVec,
25};
26use tempfile::NamedTempFile;
27use threadpool::ThreadPool;
28
29use crate::{
30    hash::ObjectHash,
31    internal::{
32        metadata::{EntryMeta, MetaAttached},
33        object::types::ObjectType,
34        pack::{entry::Entry, utils},
35    },
36};
37
38// static CACHE_OBJS_MEM_SIZE: AtomicUsize = AtomicUsize::new(0);
39
40/// file load&store trait
41pub trait FileLoadStore: Sized {
42    fn f_load(path: &Path) -> Result<Self, io::Error>;
43    fn f_save(&self, path: &Path) -> Result<(), io::Error>;
44}
45
46fn write_bytes_atomically(path: &Path, data: &[u8]) -> Result<(), io::Error> {
47    if path.exists() {
48        return Ok(());
49    }
50    let dir = path
51        .parent()
52        .filter(|parent| !parent.as_os_str().is_empty())
53        .unwrap_or_else(|| Path::new("."));
54
55    let mut temp_file = NamedTempFile::new_in(dir)?;
56    temp_file.write_all(data)?;
57
58    match temp_file.persist_noclobber(path) {
59        Ok(_persisted) => Ok(()),
60        Err(err) if err.error.kind() == io::ErrorKind::AlreadyExists && path.exists() => Ok(()),
61        Err(err) => Err(err.error),
62    }
63}
64
65impl<T> FileLoadStore for T
66where
67    T: Archive,
68    T: for<'a> RkyvSerialize<HighSerializer<AlignedVec, ArenaHandle<'a>, RkyvError>>,
69    T::Archived: for<'a> CheckBytes<HighValidator<'a, RkyvError>>
70        + RkyvDeserialize<T, Strategy<Pool, RkyvError>>,
71{
72    /// load object from file
73    fn f_load(path: &Path) -> Result<T, io::Error> {
74        let data = fs::read(path)?;
75        let obj = rkyv::from_bytes::<T, RkyvError>(&data).map_err(io::Error::other)?;
76        Ok(obj)
77    }
78
79    fn f_save(&self, path: &Path) -> Result<(), io::Error> {
80        if path.exists() {
81            return Ok(());
82        }
83
84        let data = rkyv::to_bytes::<RkyvError>(self).map_err(io::Error::other)?;
85        write_bytes_atomically(path, &data)
86    }
87}
88
89/// Represents the metadata of a cache object, indicating whether it is a delta or not.
90#[derive(
91    PartialEq,
92    Eq,
93    Clone,
94    Debug,
95    serde::Serialize,
96    serde::Deserialize,
97    rkyv::Archive,
98    rkyv::Serialize,
99    rkyv::Deserialize,
100)]
101pub(crate) enum CacheObjectInfo {
102    /// The object is one of the four basic types:
103    /// [`ObjectType::Blob`], [`ObjectType::Tree`], [`ObjectType::Commit`], or [`ObjectType::Tag`].
104    /// The metadata contains the [`ObjectType`] and the [`ObjectHash`] hash of the object.
105    BaseObject(ObjectType, ObjectHash),
106    /// The object is an offset delta with a specified offset delta [`usize`],
107    /// and the size of the expanded object (previously `delta_final_size`).
108    OffsetDelta(usize, usize),
109    /// Similar to [`OffsetDelta`], but delta algorithm is `zstd`.
110    OffsetZstdelta(usize, usize),
111    /// The object is a hash delta with a specified [`ObjectHash`] hash,
112    /// and the size of the expanded object (previously `delta_final_size`).
113    HashDelta(ObjectHash, usize),
114}
115
116impl CacheObjectInfo {
117    /// Get the [`ObjectType`] of the object.
118    pub(crate) fn object_type(&self) -> ObjectType {
119        match self {
120            CacheObjectInfo::BaseObject(obj_type, _) => *obj_type,
121            CacheObjectInfo::OffsetDelta(_, _) => ObjectType::OffsetDelta,
122            CacheObjectInfo::OffsetZstdelta(_, _) => ObjectType::OffsetZstdelta,
123            CacheObjectInfo::HashDelta(_, _) => ObjectType::HashDelta,
124        }
125    }
126}
127
128/// Represents a cached object in memory, which may be a delta or a base object.
129#[derive(Debug)]
130pub struct CacheObject {
131    pub(crate) info: CacheObjectInfo,
132    pub offset: usize,
133    pub crc32: u32,
134    pub data_decompressed: Vec<u8>,
135    pub mem_recorder: Option<Arc<AtomicUsize>>, // record mem-size of all CacheObjects of a Pack
136    pub is_delta_in_pack: bool,
137    pub(crate) known_hash: Option<ObjectHash>,
138}
139
140#[derive(Debug, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize)]
141struct CacheObjectOnDisk {
142    info: CacheObjectInfo,
143    offset: usize,
144    crc32: u32,
145    data_decompressed: Vec<u8>,
146    is_delta_in_pack: bool,
147}
148
149#[derive(Debug, rkyv::Archive, rkyv::Serialize)]
150struct CacheObjectOnDiskRef<'a> {
151    #[rkyv(with = rkyv::with::AsOwned)]
152    info: Cow<'a, CacheObjectInfo>,
153    offset: usize,
154    crc32: u32,
155    #[rkyv(with = rkyv::with::AsOwned)]
156    data_decompressed: Cow<'a, [u8]>,
157    is_delta_in_pack: bool,
158}
159
160impl<'a> From<&'a CacheObject> for CacheObjectOnDiskRef<'a> {
161    fn from(value: &'a CacheObject) -> Self {
162        Self {
163            info: Cow::Borrowed(&value.info),
164            offset: value.offset,
165            crc32: value.crc32,
166            data_decompressed: Cow::Borrowed(value.data_decompressed.as_slice()),
167            is_delta_in_pack: value.is_delta_in_pack,
168        }
169    }
170}
171
172impl From<CacheObjectOnDisk> for CacheObject {
173    fn from(value: CacheObjectOnDisk) -> Self {
174        Self {
175            info: value.info,
176            offset: value.offset,
177            crc32: value.crc32,
178            data_decompressed: value.data_decompressed,
179            mem_recorder: None,
180            is_delta_in_pack: value.is_delta_in_pack,
181            known_hash: None,
182        }
183    }
184}
185
186impl FileLoadStore for CacheObject {
187    fn f_load(path: &Path) -> Result<Self, io::Error> {
188        let obj = CacheObjectOnDisk::f_load(path)?;
189        Ok(obj.into())
190    }
191
192    fn f_save(&self, path: &Path) -> Result<(), io::Error> {
193        if path.exists() {
194            return Ok(());
195        }
196
197        let data = rkyv::to_bytes::<RkyvError>(&CacheObjectOnDiskRef::from(self))
198            .map_err(io::Error::other)?;
199        write_bytes_atomically(path, &data)
200    }
201}
202
203impl Clone for CacheObject {
204    fn clone(&self) -> Self {
205        let obj = CacheObject {
206            info: self.info.clone(),
207            offset: self.offset,
208            crc32: self.crc32,
209            data_decompressed: self.data_decompressed.clone(),
210            mem_recorder: self.mem_recorder.clone(),
211            is_delta_in_pack: self.is_delta_in_pack,
212            known_hash: self.known_hash,
213        };
214        obj.record_mem_size();
215        obj
216    }
217}
218
219// ! used by lru_mem to calculate the size of the object, limit the memory usage.
220// ! the implementation of HeapSize is not accurate, only calculate the size of the data_decompress
221// Note that: mem_size == value_size + heap_size, and we only need to impl HeapSize because value_size is known
222impl HeapSize for CacheObject {
223    /// If a [`CacheObject`] is [`ObjectType::HashDelta`] or [`ObjectType::OffsetDelta`],
224    /// it will expand to another [`CacheObject`] of other types. To prevent potential OOM,
225    /// we record the size of the expanded object as well as that of the object itself.
226    ///
227    /// Base objects, *i.e.*, [`ObjectType::Blob`], [`ObjectType::Tree`], [`ObjectType::Commit`],
228    /// and [`ObjectType::Tag`], will not be expanded, so the heap-size of the object is the same
229    /// as the size of the data.
230    ///
231    /// See [Comment in PR #755](https://github.com/web3infra-foundation/mega/pull/755#issuecomment-2543100481) for more details.
232    fn heap_size(&self) -> usize {
233        match &self.info {
234            CacheObjectInfo::BaseObject(_, _) => self.data_decompressed.heap_size(),
235            CacheObjectInfo::OffsetDelta(_, delta_final_size)
236            | CacheObjectInfo::OffsetZstdelta(_, delta_final_size)
237            | CacheObjectInfo::HashDelta(_, delta_final_size) => {
238                // To those who are concerned about why these two values are added,
239                // let's consider the lifetime of two `CacheObject`s, say `delta_obj`
240                // and `final_obj` in the function `Pack::rebuild_delta`.
241                //
242                // `delta_obj` is dropped only after `Pack::rebuild_delta` returns,
243                // but the space for `final_obj` is allocated in that function.
244                //
245                // Therefore, during the execution of `Pack::rebuild_delta`, both `delta_obj`
246                // and `final_obj` coexist. The maximum memory usage is the sum of the memory
247                // usage of `delta_obj` and `final_obj`.
248                self.data_decompressed.heap_size() + delta_final_size
249            }
250        }
251    }
252}
253
254impl Drop for CacheObject {
255    // Check: the heap-size subtracted when Drop is equal to the heap-size recorded
256    // (cannot change the heap-size during life cycle)
257    fn drop(&mut self) {
258        // (&*self).heap_size() != self.heap_size()
259        if let Some(mem_recorder) = &self.mem_recorder {
260            mem_recorder.fetch_sub((*self).mem_size(), Ordering::Release);
261        }
262    }
263}
264
265/// Heap-size recorder for a class(struct)
266/// <br> You should use a static Var to record mem-size
267/// and record mem-size after construction & minus it in `drop()`
268/// <br> So, variable-size fields in object should NOT be modified to keep heap-size stable.
269/// <br> Or, you can record the initial mem-size in this object
270/// <br> Or, update it (not impl)
271pub trait MemSizeRecorder: MemSize {
272    fn record_mem_size(&self);
273    fn set_mem_recorder(&mut self, mem_size: Arc<AtomicUsize>);
274    // fn get_mem_size() -> usize;
275}
276
277impl MemSizeRecorder for CacheObject {
278    /// record the mem-size of this `CacheObj` in a `static` `var`
279    /// <br> since that, DO NOT modify `CacheObj` after recording
280    fn record_mem_size(&self) {
281        if let Some(mem_recorder) = &self.mem_recorder {
282            mem_recorder.fetch_add(self.mem_size(), Ordering::Release);
283        }
284    }
285
286    fn set_mem_recorder(&mut self, mem_recorder: Arc<AtomicUsize>) {
287        self.mem_recorder = Some(mem_recorder);
288    }
289
290    // fn get_mem_size() -> usize {
291    //     CACHE_OBJS_MEM_SIZE.load(Ordering::Acquire)
292    // }
293}
294
295impl CacheObject {
296    /// Create a new CacheObject which is neither [`ObjectType::OffsetDelta`] nor [`ObjectType::HashDelta`].
297    pub fn new_for_undeltified(
298        obj_type: ObjectType,
299        data: Vec<u8>,
300        offset: usize,
301        crc32: u32,
302    ) -> Self {
303        let hash = utils::calculate_object_hash(obj_type, &data);
304        CacheObject {
305            info: CacheObjectInfo::BaseObject(obj_type, hash),
306            offset,
307            crc32,
308            data_decompressed: data,
309            mem_recorder: None,
310            is_delta_in_pack: false,
311            known_hash: None,
312        }
313    }
314
315    /// Get the [`ObjectType`] of the object.
316    pub fn object_type(&self) -> ObjectType {
317        self.info.object_type()
318    }
319
320    /// Get the [`ObjectHash`] hash of the object.
321    ///
322    /// If the object is a delta object, return [`None`].
323    pub fn base_object_hash(&self) -> Option<ObjectHash> {
324        match &self.info {
325            CacheObjectInfo::BaseObject(_, hash) => Some(*hash),
326            _ => None,
327        }
328    }
329
330    /// Get the offset delta of the object.
331    ///
332    /// If the object is not an offset delta, return [`None`].
333    pub fn offset_delta(&self) -> Option<usize> {
334        match &self.info {
335            CacheObjectInfo::OffsetDelta(offset, _) => Some(*offset),
336            _ => None,
337        }
338    }
339
340    /// Get the hash delta of the object.
341    ///
342    /// If the object is not a hash delta, return [`None`].
343    pub fn hash_delta(&self) -> Option<ObjectHash> {
344        match &self.info {
345            CacheObjectInfo::HashDelta(hash, _) => Some(*hash),
346            _ => None,
347        }
348    }
349
350    /// transform the CacheObject to Entry
351    pub fn to_entry(&self) -> Entry {
352        match self.info {
353            CacheObjectInfo::BaseObject(obj_type, hash) => Entry {
354                obj_type,
355                data: self.data_decompressed.clone(),
356                hash,
357                chain_len: 0,
358            },
359            _ => {
360                unreachable!("delta object should not persist!")
361            }
362        }
363    }
364
365    /// transform the CacheObject to MetaAttached<Entry, EntryMeta>
366    pub fn to_entry_metadata(&self) -> MetaAttached<Entry, EntryMeta> {
367        match self.info {
368            CacheObjectInfo::BaseObject(obj_type, hash) => {
369                let entry = Entry {
370                    obj_type,
371                    data: self.data_decompressed.clone(),
372                    hash,
373                    chain_len: 0,
374                };
375                let meta = EntryMeta {
376                    // pack_id:Some(pack_id),
377                    pack_offset: Some(self.offset),
378                    crc32: Some(self.crc32),
379                    is_delta: Some(self.is_delta_in_pack),
380                    ..Default::default()
381                };
382                MetaAttached { inner: entry, meta }
383            }
384
385            _ => {
386                unreachable!("delta object should not persist!")
387            }
388        }
389    }
390
391    /// transform the CacheObject to MetaAttached<Entry, EntryMeta> without cloning object data
392    pub fn into_entry_metadata(mut self) -> MetaAttached<Entry, EntryMeta> {
393        let recorded_size = self.mem_size();
394        if let Some(mem_recorder) = self.mem_recorder.take() {
395            mem_recorder.fetch_sub(recorded_size, Ordering::Release);
396        }
397
398        match self.info {
399            CacheObjectInfo::BaseObject(obj_type, hash) => {
400                let data = std::mem::take(&mut self.data_decompressed);
401                let entry = Entry {
402                    obj_type,
403                    data,
404                    hash,
405                    chain_len: 0,
406                };
407                let meta = EntryMeta {
408                    pack_offset: Some(self.offset),
409                    crc32: Some(self.crc32),
410                    is_delta: Some(self.is_delta_in_pack),
411                    ..Default::default()
412                };
413                MetaAttached { inner: entry, meta }
414            }
415
416            _ => {
417                unreachable!("delta object should not persist!")
418            }
419        }
420    }
421}
422
423/// trait alias for simple use
424pub trait ArcWrapperBounds: HeapSize + FileLoadStore + Send + Sync + 'static {}
425// You must impl `Alias Trait` for all the `T` satisfying Constraints
426// Or, `T` will not satisfy `Alias Trait` even if it satisfies the Original traits
427impl<T: HeapSize + FileLoadStore + Send + Sync + 'static> ArcWrapperBounds for T {}
428
429/// Implementing encapsulation of Arc to enable third-party Trait HeapSize implementation for the Arc type
430/// Because of use Arc in LruCache, the LruCache is not clear whether a pointer will drop the referenced
431/// content when it is ejected from the cache, the actual memory usage is not accurate
432pub struct ArcWrapper<T: ArcWrapperBounds> {
433    pub data: Arc<T>,
434    complete_signal: Arc<AtomicBool>,
435    pool: Option<Arc<ThreadPool>>,
436    pub store_path: Option<PathBuf>, // path to store when drop
437}
438impl<T: ArcWrapperBounds> ArcWrapper<T> {
439    /// Create a new ArcWrapper
440    pub fn new(data: Arc<T>, share_flag: Arc<AtomicBool>, pool: Option<Arc<ThreadPool>>) -> Self {
441        ArcWrapper {
442            data,
443            complete_signal: share_flag,
444            pool,
445            store_path: None,
446        }
447    }
448    /// Sets the file path where this object will be persisted when evicted from cache.
449    pub fn set_store_path(&mut self, path: PathBuf) {
450        self.store_path = Some(path);
451    }
452}
453
454impl<T: ArcWrapperBounds> HeapSize for ArcWrapper<T> {
455    fn heap_size(&self) -> usize {
456        self.data.heap_size()
457    }
458}
459
460impl<T: ArcWrapperBounds> Clone for ArcWrapper<T> {
461    /// clone won't clone the store_path
462    fn clone(&self) -> Self {
463        ArcWrapper {
464            data: self.data.clone(),
465            complete_signal: self.complete_signal.clone(),
466            pool: self.pool.clone(),
467            store_path: None,
468        }
469    }
470}
471
472impl<T: ArcWrapperBounds> Deref for ArcWrapper<T> {
473    type Target = Arc<T>;
474    fn deref(&self) -> &Self::Target {
475        &self.data
476    }
477}
478impl<T: ArcWrapperBounds> Drop for ArcWrapper<T> {
479    // `drop` will be called in `lru_cache.insert()` when cache full & eject the LRU
480    // `lru_cache.insert()` is protected by Mutex
481    fn drop(&mut self) {
482        if !self.complete_signal.load(Ordering::Acquire)
483            && let Some(path) = &self.store_path
484        {
485            match &self.pool {
486                Some(pool) => {
487                    let data_copy = self.data.clone();
488                    let path_copy = path.clone();
489                    let complete_signal = self.complete_signal.clone();
490                    // block entire process, wait for IO, Control Memory
491                    // queue size will influence the Memory usage
492                    while pool.queued_count() > 2000 {
493                        std::thread::yield_now();
494                    }
495                    pool.execute(move || {
496                        if !complete_signal.load(Ordering::Acquire) {
497                            let res = data_copy.f_save(&path_copy);
498                            if let Err(e) = res {
499                                println!("[f_save] {path_copy:?} error: {e:?}");
500                            }
501                        }
502                    });
503                }
504                None => {
505                    let res = self.data.f_save(path);
506                    if let Err(e) = res {
507                        println!("[f_save] {path:?} error: {e:?}");
508                    }
509                }
510            }
511        }
512    }
513}
514#[cfg(test)]
515mod test {
516    use std::{fs, sync::Mutex};
517
518    use lru_mem::LruCache;
519    use tempfile::tempdir;
520
521    use super::*;
522    use crate::hash::{HashKind, set_hash_kind_for_test};
523
524    /// Helper to build a base CacheObject with the given size.
525    fn make_obj(size: usize) -> CacheObject {
526        CacheObject {
527            info: CacheObjectInfo::BaseObject(ObjectType::Blob, ObjectHash::default()),
528            offset: 0,
529            crc32: 0,
530            data_decompressed: vec![0; size],
531            mem_recorder: None,
532            is_delta_in_pack: false,
533            known_hash: None,
534        }
535    }
536
537    /// Test that the memory size recording works correctly for CacheObject.
538    #[test]
539    fn test_heap_size_record() {
540        for (kind, size) in [(HashKind::Sha1, 1024usize), (HashKind::Sha256, 2048usize)] {
541            let _guard = set_hash_kind_for_test(kind);
542            let mut obj = make_obj(size);
543            let mem = Arc::new(AtomicUsize::default());
544            assert_eq!(mem.load(Ordering::Relaxed), 0);
545            obj.set_mem_recorder(mem.clone());
546            obj.record_mem_size();
547            assert_eq!(mem.load(Ordering::Relaxed), obj.mem_size());
548            drop(obj);
549            assert_eq!(mem.load(Ordering::Relaxed), 0);
550        }
551    }
552
553    /// Test that the heap size of CacheObject and ArcWrapper are the same.
554    #[test]
555    fn test_cache_object_with_same_size() {
556        for (kind, size) in [(HashKind::Sha1, 1024usize), (HashKind::Sha256, 2048usize)] {
557            let _guard = set_hash_kind_for_test(kind);
558            let a = make_obj(size);
559            assert_eq!(a.heap_size(), size);
560
561            let b = ArcWrapper::new(Arc::new(a.clone()), Arc::new(AtomicBool::new(false)), None);
562            assert_eq!(b.heap_size(), size);
563        }
564    }
565
566    /// Test that the LRU cache correctly ejects the least recently used object when capacity is exceeded.
567    #[test]
568    fn test_cache_object_with_lru() {
569        for (kind, cap, size_a, size_b) in [
570            (
571                HashKind::Sha1,
572                2048usize,
573                1024usize,
574                (1024.0 * 1.5) as usize,
575            ),
576            (
577                HashKind::Sha256,
578                4096usize,
579                2048usize,
580                (2048.0 * 1.5) as usize,
581            ),
582        ] {
583            let _guard = set_hash_kind_for_test(kind);
584            let mut cache = LruCache::new(cap);
585
586            let hash_a = ObjectHash::default();
587            let hash_b = ObjectHash::new(b"b"); // whatever different hash
588            let a = make_obj(size_a);
589            let b = make_obj(size_b);
590
591            {
592                let r = cache.insert(
593                    hash_a.to_string(),
594                    ArcWrapper::new(Arc::new(a.clone()), Arc::new(AtomicBool::new(true)), None),
595                );
596                assert!(r.is_ok());
597            }
598
599            {
600                let r = cache.try_insert(
601                    hash_b.to_string(),
602                    ArcWrapper::new(Arc::new(b.clone()), Arc::new(AtomicBool::new(true)), None),
603                );
604                assert!(r.is_err());
605                if let Err(lru_mem::TryInsertError::WouldEjectLru { .. }) = r {
606                    // expected
607                } else {
608                    panic!("Expected WouldEjectLru error");
609                }
610
611                let r = cache.insert(
612                    hash_b.to_string(),
613                    ArcWrapper::new(Arc::new(b.clone()), Arc::new(AtomicBool::new(true)), None),
614                );
615                assert!(r.is_ok());
616            }
617
618            {
619                let r = cache.get(&hash_a.to_string());
620                assert!(r.is_none());
621            }
622        }
623    }
624
625    /// test that the Drop trait is called when an object is ejected from the LRU cache
626    #[derive(
627        serde::Serialize, serde::Deserialize, rkyv::Archive, rkyv::Serialize, rkyv::Deserialize,
628    )]
629    struct Test {
630        a: usize,
631    }
632    impl Drop for Test {
633        fn drop(&mut self) {
634            println!("drop Test");
635        }
636    }
637    impl HeapSize for Test {
638        fn heap_size(&self) -> usize {
639            self.a
640        }
641    }
642    #[test]
643    fn test_lru_drop() {
644        println!("insert a");
645        let cache = LruCache::new(2048);
646        let cache = Arc::new(Mutex::new(cache));
647        {
648            let mut c = cache.as_ref().lock().unwrap();
649            let _ = c.insert(
650                "a",
651                ArcWrapper::new(
652                    Arc::new(Test { a: 1024 }),
653                    Arc::new(AtomicBool::new(true)),
654                    None,
655                ),
656            );
657        }
658        println!("insert b, a should be ejected");
659        {
660            let mut c = cache.as_ref().lock().unwrap();
661            let _ = c.insert(
662                "b",
663                ArcWrapper::new(
664                    Arc::new(Test { a: 1200 }),
665                    Arc::new(AtomicBool::new(true)),
666                    None,
667                ),
668            );
669        }
670        let b = {
671            let mut c = cache.as_ref().lock().unwrap();
672            c.get("b").cloned()
673        };
674        println!("insert c, b should not be ejected");
675        {
676            let mut c = cache.as_ref().lock().unwrap();
677            let _ = c.insert(
678                "c",
679                ArcWrapper::new(
680                    Arc::new(Test { a: 1200 }),
681                    Arc::new(AtomicBool::new(true)),
682                    None,
683                ),
684            );
685        }
686        println!("user b: {}", b.as_ref().unwrap().a);
687        println!("test over, enject all");
688    }
689
690    #[test]
691    fn test_cache_object_serialize() {
692        for (kind, size) in [(HashKind::Sha1, 1024usize), (HashKind::Sha256, 2048usize)] {
693            let _guard = set_hash_kind_for_test(kind);
694            let a = make_obj(size);
695            let s = rkyv::to_bytes::<RkyvError>(&CacheObjectOnDiskRef::from(&a)).unwrap();
696            let b: CacheObject = rkyv::from_bytes::<CacheObjectOnDisk, RkyvError>(&s)
697                .unwrap()
698                .into();
699            assert_eq!(a.info, b.info);
700            assert_eq!(a.data_decompressed, b.data_decompressed);
701            assert_eq!(a.offset, b.offset);
702            assert!(b.mem_recorder.is_none());
703        }
704    }
705
706    #[test]
707    fn test_write_bytes_atomically_creates_file_when_missing() {
708        let dir = tempdir().unwrap();
709        let path = dir.path().join("object");
710
711        write_bytes_atomically(&path, b"fresh").unwrap();
712
713        assert_eq!(fs::read(&path).unwrap(), b"fresh");
714    }
715
716    #[test]
717    fn test_write_bytes_atomically_returns_when_target_exists() {
718        let dir = tempdir().unwrap();
719        let path = dir.path().join("object");
720
721        fs::write(&path, b"existing").unwrap();
722        write_bytes_atomically(&path, b"new-data").unwrap();
723
724        assert_eq!(fs::read(&path).unwrap(), b"existing");
725    }
726
727    #[test]
728    fn test_cache_object_file_roundtrip() {
729        let dir = tempdir().unwrap();
730        let path = dir.path().join("object");
731        let a = make_obj(1024);
732
733        a.f_save(&path).unwrap();
734        let b = CacheObject::f_load(&path).unwrap();
735
736        assert_eq!(a.info, b.info);
737        assert_eq!(a.data_decompressed, b.data_decompressed);
738        assert_eq!(a.offset, b.offset);
739        assert!(b.mem_recorder.is_none());
740    }
741
742    #[test]
743    fn test_arc_wrapper_drop_store() {
744        let mut path = PathBuf::from(".cache_temp/test_arc_wrapper_drop_store");
745        fs::create_dir_all(&path).unwrap();
746        path.push("test_obj");
747        let mut a = ArcWrapper::new(Arc::new(1024), Arc::new(AtomicBool::new(false)), None);
748        a.set_store_path(path.clone());
749        drop(a);
750
751        assert!(path.exists());
752        path.pop();
753        fs::remove_dir_all(&path).unwrap();
754        // Try to remove parent .cache_temp directory if it's empty
755        let _ = fs::remove_dir(".cache_temp");
756    }
757
758    #[test]
759    /// test warpper can't correctly store the data when lru eject it
760    fn test_arc_wrapper_with_lru() {
761        let mut cache = LruCache::new(1500);
762        let path = PathBuf::from(".cache_temp/test_arc_wrapper_with_lru");
763        let _ = fs::remove_dir_all(&path);
764        fs::create_dir_all(&path).unwrap();
765        let shared_flag = Arc::new(AtomicBool::new(false));
766
767        // insert a, a not ejected
768        let a_path = path.join("a");
769        {
770            let mut a = ArcWrapper::new(Arc::new(Test { a: 1024 }), shared_flag.clone(), None);
771            a.set_store_path(a_path.clone());
772            let b = ArcWrapper::new(Arc::new(1024), shared_flag.clone(), None);
773            assert!(b.store_path.is_none());
774
775            println!("insert a with heap size: {:?}", a.heap_size());
776            let rt = cache.insert("a", a);
777            if let Err(e) = rt {
778                panic!("{}", format!("insert a failed: {:?}", e.to_string()));
779            }
780            println!("after insert a, cache used = {}", cache.current_size());
781        }
782        assert!(!a_path.exists());
783
784        let b_path = path.join("b");
785        // insert b, a should be ejected
786        {
787            let mut b = ArcWrapper::new(Arc::new(Test { a: 996 }), shared_flag.clone(), None);
788            b.set_store_path(b_path.clone());
789            let rt = cache.insert("b", b);
790            if let Err(e) = rt {
791                panic!("{}", format!("insert a failed: {:?}", e.to_string()));
792            }
793            println!("after insert b, cache used = {}", cache.current_size());
794        }
795        assert!(a_path.exists());
796        assert!(!b_path.exists());
797        shared_flag.store(true, Ordering::Release);
798        fs::remove_dir_all(path).unwrap();
799        // Try to remove parent .cache_temp directory if it's empty
800        let _ = fs::remove_dir(".cache_temp");
801        // should pass even b's path not exists
802    }
803}
git_internal/internal/pack/cache_object.rs

git_internal/internal/pack/
cache_object.rs
