Coverage Report

Created: 2024-10-22 12:33

/build/source/nativelink-store/src/memory_store.rs
Line
Count
Source (jump to first uncovered line)
1
// Copyright 2024 The NativeLink Authors. All rights reserved.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//    http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
15
use std::fmt::Debug;
16
use std::ops::Bound;
17
use std::pin::Pin;
18
use std::sync::Arc;
19
use std::time::SystemTime;
20
21
use async_trait::async_trait;
22
use bytes::{Bytes, BytesMut};
23
use nativelink_error::{Code, Error, ResultExt};
24
use nativelink_metric::MetricsComponent;
25
use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf};
26
use nativelink_util::evicting_map::{EvictingMap, LenEntry};
27
use nativelink_util::health_utils::{default_health_status_indicator, HealthStatusIndicator};
28
use nativelink_util::store_trait::{StoreDriver, StoreKey, UploadSizeInfo};
29
30
use crate::cas_utils::is_zero_digest;
31
32
#[derive(Clone)]
33
pub struct BytesWrapper(Bytes);
34
35
impl Debug for BytesWrapper {
36
0
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37
0
        f.write_str("BytesWrapper { -- Binary data -- }")
38
0
    }
39
}
40
41
impl LenEntry for BytesWrapper {
42
    #[inline]
43
10.2k
    fn len(&self) -> u64 {
44
10.2k
        Bytes::len(&self.0) as u64
45
10.2k
    }
46
47
    #[inline]
48
0
    fn is_empty(&self) -> bool {
49
0
        Bytes::is_empty(&self.0)
50
0
    }
51
}
52
53
0
#[derive(MetricsComponent)]
54
pub struct MemoryStore {
55
    #[metric(group = "evicting_map")]
56
    evicting_map: EvictingMap<StoreKey<'static>, BytesWrapper, SystemTime>,
57
}
58
59
impl MemoryStore {
60
300
    pub fn new(config: &nativelink_config::stores::MemoryStore) -> Arc<Self> {
61
300
        let empty_policy = nativelink_config::stores::EvictionPolicy::default();
62
300
        let eviction_policy = config.eviction_policy.as_ref().unwrap_or(&empty_policy);
63
300
        Arc::new(Self {
64
300
            evicting_map: EvictingMap::new(eviction_policy, SystemTime::now()),
65
300
        })
66
300
    }
67
68
    /// Returns the number of key-value pairs that are currently in the the cache.
69
    /// Function is not for production code paths.
70
30
    pub async fn len_for_test(&self) -> usize {
71
30
        self.evicting_map.len_for_test().
await0
72
30
    }
73
74
8
    
pub async fn remove_entry(&self, key: StoreKey<'_>) -> bool 0
{
75
8
        self.evicting_map.remove(&key.into_owned()).
await0
76
8
    }
77
}
78
79
#[async_trait]
80
impl StoreDriver for MemoryStore {
81
    async fn has_with_results(
82
        self: Pin<&Self>,
83
        keys: &[StoreKey<'_>],
84
        results: &mut [Option<u64>],
85
183
    ) -> Result<(), Error> {
86
        // TODO(allada): This is a dirty hack to get around the lifetime issues with the
87
        // evicting map.
88
225
        let 
digests: Vec<_> = keys.iter().map(183
|key| key.borrow().into_owned()).collect();
89
183
        self.evicting_map
90
183
            .sizes_for_keys(digests, results, false /* peek */)
91
0
            .await;
92
        // We need to do a special pass to ensure our zero digest exist.
93
183
        keys.iter()
94
183
            .zip(results.iter_mut())
95
225
            .for_each(|(key, result)| {
96
225
                if is_zero_digest(key.borrow()) {
  Branch (96:20): [True: 1, False: 224]
  Branch (96:20): [Folded - Ignored]
97
1
                    *result = Some(0);
98
224
                }
99
225
            });
100
183
        Ok(())
101
366
    }
102
103
    async fn list(
104
        self: Pin<&Self>,
105
        range: (Bound<StoreKey<'_>>, Bound<StoreKey<'_>>),
106
        handler: &mut (dyn for<'a> FnMut(&'a StoreKey) -> bool + Send + Sync + '_),
107
7
    ) -> Result<u64, Error> {
108
7
        let range = (
109
7
            range.0.map(StoreKey::into_owned),
110
7
            range.1.map(StoreKey::into_owned),
111
7
        );
112
7
        let iterations = self
113
7
            .evicting_map
114
13
            .range(range, move |key, _value| handler(key)
)7
115
0
            .await;
116
7
        Ok(iterations)
117
14
    }
118
119
    async fn update(
120
        self: Pin<&Self>,
121
        key: StoreKey<'_>,
122
        mut reader: DropCloserReadHalf,
123
        _size_info: UploadSizeInfo,
124
5.71k
    ) -> Result<(), Error> {
125
        // Internally Bytes might hold a reference to more data than just our data. To prevent
126
        // this potential case, we make a full copy of our data for long-term storage.
127
5.71k
        let final_buffer = {
128
5.71k
            let buffer = reader
129
5.71k
                .consume(None)
130
360
                .await
131
5.71k
                .err_tip(|| 
"Failed to collect all bytes from reader in memory_store::update"4
)
?4
;
132
5.71k
            let mut new_buffer = BytesMut::with_capacity(buffer.len());
133
5.71k
            new_buffer.extend_from_slice(&buffer[..]);
134
5.71k
            new_buffer.freeze()
135
5.71k
        };
136
5.71k
137
5.71k
        self.evicting_map
138
5.71k
            .insert(key.borrow().into_owned(), BytesWrapper(final_buffer))
139
0
            .await;
140
5.71k
        Ok(())
141
11.4k
    }
142
143
    async fn get_part(
144
        self: Pin<&Self>,
145
        key: StoreKey<'_>,
146
        writer: &mut DropCloserWriteHalf,
147
        offset: u64,
148
        length: Option<u64>,
149
4.46k
    ) -> Result<(), Error> {
150
4.46k
        let offset = usize::try_from(offset).err_tip(|| 
"Could not convert offset to usize"0
)
?0
;
151
4.46k
        let length = length
152
4.46k
            .map(|v| 
usize::try_from(v).err_tip(52
||
"Could not convert length to usize"0
)52
)
153
4.46k
            .transpose()
?0
;
154
155
4.46k
        if is_zero_digest(key.borrow()) {
  Branch (155:12): [True: 1, False: 4.45k]
  Branch (155:12): [Folded - Ignored]
156
1
            writer
157
1
                .send_eof()
158
1
                .err_tip(|| 
"Failed to send zero EOF in filesystem store get_part"0
)
?0
;
159
1
            return Ok(());
160
4.45k
        }
161
162
4.45k
        let 
value4.45k
= self
163
4.45k
            .evicting_map
164
4.45k
            .get(&key.borrow().into_owned())
165
0
            .await
166
4.45k
            .err_tip_with_code(|_| 
(Code::NotFound, format!("Key {key:?} not found"))7
)
?7
;
167
4.45k
        let default_len = usize::try_from(value.len())
168
4.45k
            .err_tip(|| 
"Could not convert value.len() to usize"0
)
?0
169
4.45k
            .saturating_sub(offset);
170
4.45k
        let length = length.unwrap_or(default_len).min(default_len);
171
4.45k
        if length > 0 {
  Branch (171:12): [True: 4.44k, False: 3]
  Branch (171:12): [Folded - Ignored]
172
4.44k
            writer
173
4.44k
                .send(value.0.slice(offset..(offset + length)))
174
0
                .await
175
4.44k
                .err_tip(|| 
"Failed to write data in memory store"0
)
?0
;
176
3
        }
177
4.45k
        writer
178
4.45k
            .send_eof()
179
4.45k
            .err_tip(|| 
"Failed to write EOF in memory store get_part"0
)
?0
;
180
4.45k
        Ok(())
181
8.92k
    }
182
183
83
    fn inner_store(&self, _digest: Option<StoreKey>) -> &dyn StoreDriver {
184
83
        self
185
83
    }
186
187
38
    fn as_any<'a>(&'a self) -> &'a (dyn std::any::Any + Sync + Send + 'static) {
188
38
        self
189
38
    }
190
191
0
    fn as_any_arc(self: Arc<Self>) -> Arc<dyn std::any::Any + Sync + Send + 'static> {
192
0
        self
193
0
    }
194
}
195
196
default_health_status_indicator!(MemoryStore);