/build/source/nativelink-store/src/filesystem_store.rs
Line | Count | Source |
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 core::fmt::{Debug, Formatter}; |
16 | | use core::pin::Pin; |
17 | | use core::sync::atomic::{AtomicU64, Ordering}; |
18 | | use std::borrow::Cow; |
19 | | use std::ffi::{OsStr, OsString}; |
20 | | use std::sync::{Arc, Weak}; |
21 | | use std::time::SystemTime; |
22 | | |
23 | | use async_lock::RwLock; |
24 | | use async_trait::async_trait; |
25 | | use bytes::BytesMut; |
26 | | use futures::stream::{StreamExt, TryStreamExt}; |
27 | | use futures::{Future, TryFutureExt}; |
28 | | use nativelink_config::stores::FilesystemSpec; |
29 | | use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err}; |
30 | | use nativelink_metric::MetricsComponent; |
31 | | use nativelink_util::background_spawn; |
32 | | use nativelink_util::buf_channel::{ |
33 | | DropCloserReadHalf, DropCloserWriteHalf, make_buf_channel_pair, |
34 | | }; |
35 | | use nativelink_util::common::{DigestInfo, fs}; |
36 | | use nativelink_util::evicting_map::{EvictingMap, LenEntry}; |
37 | | use nativelink_util::health_utils::{HealthRegistryBuilder, HealthStatus, HealthStatusIndicator}; |
38 | | use nativelink_util::store_trait::{ |
39 | | StoreDriver, StoreKey, StoreKeyBorrow, StoreOptimizations, UploadSizeInfo, |
40 | | }; |
41 | | use tokio::io::{AsyncReadExt, AsyncWriteExt, Take}; |
42 | | use tokio_stream::wrappers::ReadDirStream; |
43 | | use tracing::{debug, error, warn}; |
44 | | |
45 | | use crate::cas_utils::is_zero_digest; |
46 | | |
47 | | // Default size to allocate memory of the buffer when reading files. |
48 | | const DEFAULT_BUFF_SIZE: usize = 32 * 1024; |
49 | | // Default block size of all major filesystems is 4KB |
50 | | const DEFAULT_BLOCK_SIZE: u64 = 4 * 1024; |
51 | | |
52 | | pub const STR_FOLDER: &str = "s"; |
53 | | pub const DIGEST_FOLDER: &str = "d"; |
54 | | |
55 | | #[derive(Clone, Copy, Debug)] |
56 | | pub enum FileType { |
57 | | Digest, |
58 | | String, |
59 | | } |
60 | | |
61 | | #[derive(Debug, MetricsComponent)] |
62 | | pub struct SharedContext { |
63 | | // Used in testing to know how many active drop() spawns are running. |
64 | | // TODO(aaronmondal) It is probably a good idea to use a spin lock during |
65 | | // destruction of the store to ensure that all files are actually |
66 | | // deleted (similar to how it is done in tests). |
67 | | #[metric(help = "Number of active drop spawns")] |
68 | | pub active_drop_spawns: AtomicU64, |
69 | | #[metric(help = "Path to the configured temp path")] |
70 | | temp_path: String, |
71 | | #[metric(help = "Path to the configured content path")] |
72 | | content_path: String, |
73 | | } |
74 | | |
75 | | #[derive(Eq, PartialEq, Debug)] |
76 | | enum PathType { |
77 | | Content, |
78 | | Temp, |
79 | | Custom(OsString), |
80 | | } |
81 | | |
82 | | /// [`EncodedFilePath`] stores the path to the file |
83 | | /// including the context, path type and key to the file. |
84 | | /// The whole [`StoreKey`] is stored as opposed to solely |
85 | | /// the [`DigestInfo`] so that it is more usable for things |
86 | | /// such as BEP -see Issue #1108 |
87 | | #[derive(Debug)] |
88 | | pub struct EncodedFilePath { |
89 | | shared_context: Arc<SharedContext>, |
90 | | path_type: PathType, |
91 | | key: StoreKey<'static>, |
92 | | } |
93 | | |
94 | | impl EncodedFilePath { |
95 | | #[inline] |
96 | 330 | fn get_file_path(&self) -> Cow<'_, OsStr> { |
97 | 330 | get_file_path_raw(&self.path_type, self.shared_context.as_ref(), &self.key) |
98 | 330 | } |
99 | | } |
100 | | |
101 | | #[inline] |
102 | 447 | fn get_file_path_raw<'a>( |
103 | 447 | path_type: &'a PathType, |
104 | 447 | shared_context: &SharedContext, |
105 | 447 | key: &StoreKey<'a>, |
106 | 447 | ) -> Cow<'a, OsStr> { |
107 | 447 | let folder438 = match path_type { |
108 | 195 | PathType::Content => &shared_context.content_path, |
109 | 243 | PathType::Temp => &shared_context.temp_path, |
110 | 9 | PathType::Custom(path) => return Cow::Borrowed(path), |
111 | | }; |
112 | 438 | Cow::Owned(to_full_path_from_key(folder, key)) |
113 | 447 | } |
114 | | |
115 | | impl Drop for EncodedFilePath { |
116 | 118 | fn drop(&mut self) { |
117 | | // `drop()` can be called during shutdown, so we use `path_type` flag to know if the |
118 | | // file actually needs to be deleted. |
119 | 118 | if self.path_type == PathType::Content { Branch (119:12): [True: 92, False: 26]
Branch (119:12): [Folded - Ignored]
|
120 | 92 | return; |
121 | 26 | } |
122 | | |
123 | 26 | let file_path = self.get_file_path().to_os_string(); |
124 | 26 | let shared_context = self.shared_context.clone(); |
125 | 26 | shared_context |
126 | 26 | .active_drop_spawns |
127 | 26 | .fetch_add(1, Ordering::Relaxed); |
128 | 26 | background_spawn!("filesystem_delete_file", async move {24 |
129 | 24 | debug!(?file_path, "File deleted"0 ,); |
130 | 24 | let result23 = fs::remove_file(&file_path) |
131 | 24 | .await |
132 | 23 | .err_tip(|| format!("Failed to remove file {}"0 , file_path.display()0 )); |
133 | 23 | if let Err(err0 ) = result { Branch (133:20): [True: 0, False: 23]
Branch (133:20): [Folded - Ignored]
|
134 | 0 | error!(?file_path, ?err, "Failed to delete file",); |
135 | 23 | } |
136 | 23 | shared_context |
137 | 23 | .active_drop_spawns |
138 | 23 | .fetch_sub(1, Ordering::Relaxed); |
139 | 23 | }); |
140 | 118 | } |
141 | | } |
142 | | |
143 | | /// This creates the file path from the [`StoreKey`]. If |
144 | | /// it is a string, the string, prefixed with [`STR_PREFIX`] |
145 | | /// for backwards compatibility, is stored. |
146 | | /// |
147 | | /// If it is a [`DigestInfo`], it is prefixed by [`DIGEST_PREFIX`] |
148 | | /// followed by the string representation of a digest - the hash in hex, |
149 | | /// a hyphen then the size in bytes |
150 | | /// |
151 | | /// Previously, only the string representation of the [`DigestInfo`] was |
152 | | /// used with no prefix |
153 | | #[inline] |
154 | 465 | fn to_full_path_from_key(folder: &str, key: &StoreKey<'_>) -> OsString { |
155 | 465 | match key { |
156 | 3 | StoreKey::Str(str) => format!("{folder}/{STR_FOLDER}/{str}"), |
157 | 462 | StoreKey::Digest(digest_info) => format!("{folder}/{DIGEST_FOLDER}/{digest_info}"), |
158 | | } |
159 | 465 | .into() |
160 | 465 | } |
161 | | |
162 | | pub trait FileEntry: LenEntry + Send + Sync + Debug + 'static { |
163 | | /// Responsible for creating the underlying `FileEntry`. |
164 | | fn create(data_size: u64, block_size: u64, encoded_file_path: RwLock<EncodedFilePath>) -> Self; |
165 | | |
166 | | /// Creates a (usually) temp file, opens it and returns the path to the temp file. |
167 | | fn make_and_open_file( |
168 | | block_size: u64, |
169 | | encoded_file_path: EncodedFilePath, |
170 | | ) -> impl Future<Output = Result<(Self, fs::FileSlot, OsString), Error>> + Send |
171 | | where |
172 | | Self: Sized; |
173 | | |
174 | | /// Returns the underlying reference to the size of the data in bytes |
175 | | fn data_size_mut(&mut self) -> &mut u64; |
176 | | |
177 | | /// Returns the actual size of the underlying file on the disk after accounting for filesystem block size. |
178 | | fn size_on_disk(&self) -> u64; |
179 | | |
180 | | /// Gets the underlying `EncodedfilePath`. |
181 | | fn get_encoded_file_path(&self) -> &RwLock<EncodedFilePath>; |
182 | | |
183 | | /// Returns a reader that will read part of the underlying file. |
184 | | fn read_file_part( |
185 | | &self, |
186 | | offset: u64, |
187 | | length: u64, |
188 | | ) -> impl Future<Output = Result<Take<fs::FileSlot>, Error>> + Send; |
189 | | |
190 | | /// This function is a safe way to extract the file name of the underlying file. To protect users from |
191 | | /// accidentally creating undefined behavior we encourage users to do the logic they need to do with |
192 | | /// the filename inside this function instead of extracting the filename and doing the logic outside. |
193 | | /// This is because the filename is not guaranteed to exist after this function returns, however inside |
194 | | /// the callback the file is always guaranteed to exist and immutable. |
195 | | /// DO NOT USE THIS FUNCTION TO EXTRACT THE FILENAME AND STORE IT FOR LATER USE. |
196 | | fn get_file_path_locked< |
197 | | T, |
198 | | Fut: Future<Output = Result<T, Error>> + Send, |
199 | | F: FnOnce(OsString) -> Fut + Send, |
200 | | >( |
201 | | &self, |
202 | | handler: F, |
203 | | ) -> impl Future<Output = Result<T, Error>> + Send; |
204 | | } |
205 | | |
206 | | pub struct FileEntryImpl { |
207 | | data_size: u64, |
208 | | block_size: u64, |
209 | | encoded_file_path: RwLock<EncodedFilePath>, |
210 | | } |
211 | | |
212 | | impl FileEntryImpl { |
213 | 9 | pub fn get_shared_context_for_test(&mut self) -> Arc<SharedContext> { |
214 | 9 | self.encoded_file_path.get_mut().shared_context.clone() |
215 | 9 | } |
216 | | } |
217 | | |
218 | | impl FileEntry for FileEntryImpl { |
219 | 118 | fn create(data_size: u64, block_size: u64, encoded_file_path: RwLock<EncodedFilePath>) -> Self { |
220 | 118 | Self { |
221 | 118 | data_size, |
222 | 118 | block_size, |
223 | 118 | encoded_file_path, |
224 | 118 | } |
225 | 118 | } |
226 | | |
227 | | /// This encapsulates the logic for the edge case of if the file fails to create |
228 | | /// the cleanup of the file is handled without creating a `FileEntry`, which would |
229 | | /// try to cleanup the file as well during `drop()`. |
230 | 108 | async fn make_and_open_file( |
231 | 108 | block_size: u64, |
232 | 108 | encoded_file_path: EncodedFilePath, |
233 | 108 | ) -> Result<(Self, fs::FileSlot, OsString), Error> { |
234 | 108 | let temp_full_path = encoded_file_path.get_file_path().to_os_string(); |
235 | 108 | let temp_file_result = fs::create_file(temp_full_path.clone()) |
236 | 108 | .or_else(|mut err| async {0 |
237 | 0 | let remove_result = fs::remove_file(&temp_full_path).await.err_tip(|| { |
238 | 0 | format!( |
239 | 0 | "Failed to remove file {} in filesystem store", |
240 | 0 | temp_full_path.display() |
241 | | ) |
242 | 0 | }); |
243 | 0 | if let Err(remove_err) = remove_result { Branch (243:24): [True: 0, False: 0]
Branch (243:24): [True: 0, False: 0]
Branch (243:24): [True: 0, False: 0]
Branch (243:24): [Folded - Ignored]
Branch (243:24): [True: 0, False: 0]
|
244 | 0 | err = err.merge(remove_err); |
245 | 0 | } |
246 | 0 | warn!(?err, ?block_size, ?temp_full_path, "Failed to create file",); |
247 | 0 | Err(err).err_tip(|| { |
248 | 0 | format!( |
249 | 0 | "Failed to create {} in filesystem store", |
250 | 0 | temp_full_path.display() |
251 | | ) |
252 | 0 | }) |
253 | 0 | }) |
254 | 108 | .await?0 ; |
255 | | |
256 | 108 | Ok(( |
257 | 108 | <Self as FileEntry>::create( |
258 | 108 | 0, /* Unknown yet, we will fill it in later */ |
259 | 108 | block_size, |
260 | 108 | RwLock::new(encoded_file_path), |
261 | 108 | ), |
262 | 108 | temp_file_result, |
263 | 108 | temp_full_path, |
264 | 108 | )) |
265 | 108 | } |
266 | | |
267 | 108 | fn data_size_mut(&mut self) -> &mut u64 { |
268 | 108 | &mut self.data_size |
269 | 108 | } |
270 | | |
271 | 245 | fn size_on_disk(&self) -> u64 { |
272 | 245 | self.data_size.div_ceil(self.block_size) * self.block_size |
273 | 245 | } |
274 | | |
275 | 169 | fn get_encoded_file_path(&self) -> &RwLock<EncodedFilePath> { |
276 | 169 | &self.encoded_file_path |
277 | 169 | } |
278 | | |
279 | 46 | fn read_file_part( |
280 | 46 | &self, |
281 | 46 | offset: u64, |
282 | 46 | length: u64, |
283 | 46 | ) -> impl Future<Output = Result<Take<fs::FileSlot>, Error>> + Send { |
284 | 46 | self.get_file_path_locked(move |full_content_path| async move { |
285 | 46 | let file44 = fs::open_file(&full_content_path, offset, length) |
286 | 46 | .await |
287 | 46 | .err_tip(|| {2 |
288 | 2 | format!( |
289 | 2 | "Failed to open file in filesystem store {}", |
290 | 2 | full_content_path.display() |
291 | | ) |
292 | 2 | })?; |
293 | 44 | Ok(file) |
294 | 92 | }) |
295 | 46 | } |
296 | | |
297 | 52 | async fn get_file_path_locked< |
298 | 52 | T, |
299 | 52 | Fut: Future<Output = Result<T, Error>> + Send, |
300 | 52 | F: FnOnce(OsString) -> Fut + Send, |
301 | 52 | >( |
302 | 52 | &self, |
303 | 52 | handler: F, |
304 | 52 | ) -> Result<T, Error> { |
305 | 52 | let encoded_file_path = self.get_encoded_file_path().read().await; |
306 | 52 | handler(encoded_file_path.get_file_path().to_os_string()).await |
307 | 52 | } |
308 | | } |
309 | | |
310 | | impl Debug for FileEntryImpl { |
311 | 0 | fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), core::fmt::Error> { |
312 | 0 | f.debug_struct("FileEntryImpl") |
313 | 0 | .field("data_size", &self.data_size) |
314 | 0 | .field("encoded_file_path", &"<behind mutex>") |
315 | 0 | .finish() |
316 | 0 | } |
317 | | } |
318 | | |
319 | 135 | fn make_temp_digest(mut digest: DigestInfo) -> DigestInfo { |
320 | | static DELETE_FILE_COUNTER: AtomicU64 = AtomicU64::new(0); |
321 | 135 | let mut hash = *digest.packed_hash(); |
322 | 135 | hash[24..].clone_from_slice( |
323 | 135 | &DELETE_FILE_COUNTER |
324 | 135 | .fetch_add(1, Ordering::Relaxed) |
325 | 135 | .to_le_bytes(), |
326 | | ); |
327 | 135 | digest.set_packed_hash(*hash); |
328 | 135 | digest |
329 | 135 | } |
330 | | |
331 | 135 | fn make_temp_key(key: &StoreKey) -> StoreKey<'static> { |
332 | 135 | StoreKey::Digest(make_temp_digest(key.borrow().into_digest())) |
333 | 135 | } |
334 | | |
335 | | impl LenEntry for FileEntryImpl { |
336 | | #[inline] |
337 | 243 | fn len(&self) -> u64 { |
338 | 243 | self.size_on_disk() |
339 | 243 | } |
340 | | |
341 | 0 | fn is_empty(&self) -> bool { |
342 | 0 | self.data_size == 0 |
343 | 0 | } |
344 | | |
345 | | // unref() only triggers when an item is removed from the eviction_map. It is possible |
346 | | // that another place in code has a reference to `FileEntryImpl` and may later read the |
347 | | // file. To support this edge case, we first move the file to a temp file and point |
348 | | // target file location to the new temp file. `unref()` should only ever be called once. |
349 | | #[inline] |
350 | 28 | async fn unref(&self) { |
351 | | { |
352 | 28 | let mut encoded_file_path = self.encoded_file_path.write().await; |
353 | 28 | if encoded_file_path.path_type == PathType::Temp { Branch (353:16): [True: 0, False: 0]
Branch (353:16): [True: 1, False: 9]
Branch (353:16): [True: 0, False: 0]
Branch (353:16): [True: 0, False: 0]
Branch (353:16): [Folded - Ignored]
Branch (353:16): [True: 0, False: 18]
|
354 | | // We are already a temp file that is now marked for deletion on drop. |
355 | | // This is very rare, but most likely the rename into the content path failed. |
356 | 1 | return; |
357 | 27 | } |
358 | 27 | let from_path = encoded_file_path.get_file_path(); |
359 | 27 | let new_key = make_temp_key(&encoded_file_path.key); |
360 | | |
361 | 27 | let to_path = |
362 | 27 | to_full_path_from_key(&encoded_file_path.shared_context.temp_path, &new_key); |
363 | | |
364 | 27 | if let Err(err2 ) = fs::rename(&from_path, &to_path).await { Branch (364:20): [True: 0, False: 0]
Branch (364:20): [True: 2, False: 7]
Branch (364:20): [True: 0, False: 0]
Branch (364:20): [True: 0, False: 0]
Branch (364:20): [Folded - Ignored]
Branch (364:20): [True: 0, False: 18]
|
365 | 2 | warn!( |
366 | 2 | key = ?encoded_file_path.key, |
367 | | ?from_path, |
368 | | ?to_path, |
369 | | ?err, |
370 | 2 | "Failed to rename file", |
371 | | ); |
372 | | } else { |
373 | 25 | debug!( |
374 | 0 | key = ?encoded_file_path.key, |
375 | | ?from_path, |
376 | | ?to_path, |
377 | 0 | "Renamed file", |
378 | | ); |
379 | 25 | encoded_file_path.path_type = PathType::Temp; |
380 | 25 | encoded_file_path.key = new_key; |
381 | | } |
382 | | } |
383 | 28 | } |
384 | | } |
385 | | |
386 | | #[inline] |
387 | 2 | fn digest_from_filename(file_name: &str) -> Result<DigestInfo, Error> { |
388 | 2 | let (hash, size) = file_name.split_once('-').err_tip(|| "")?0 ; |
389 | 2 | let size = size.parse::<i64>()?0 ; |
390 | 2 | DigestInfo::try_new(hash, size) |
391 | 2 | } |
392 | | |
393 | 2 | pub fn key_from_file(file_name: &str, file_type: FileType) -> Result<StoreKey<'_>, Error> { |
394 | 2 | match file_type { |
395 | 0 | FileType::String => Ok(StoreKey::new_str(file_name)), |
396 | 2 | FileType::Digest => digest_from_filename(file_name).map(StoreKey::Digest), |
397 | | } |
398 | 2 | } |
399 | | |
400 | | /// The number of files to read the metadata for at the same time when running |
401 | | /// `add_files_to_cache`. |
402 | | const SIMULTANEOUS_METADATA_READS: usize = 200; |
403 | | |
404 | 41 | async fn add_files_to_cache<Fe: FileEntry>( |
405 | 41 | evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>, |
406 | 41 | anchor_time: &SystemTime, |
407 | 41 | shared_context: &Arc<SharedContext>, |
408 | 41 | block_size: u64, |
409 | 41 | rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, |
410 | 41 | ) -> Result<(), Error> { |
411 | | #[expect(clippy::too_many_arguments)] |
412 | 1 | async fn process_entry<Fe: FileEntry>( |
413 | 1 | evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>, |
414 | 1 | file_name: &str, |
415 | 1 | file_type: FileType, |
416 | 1 | atime: SystemTime, |
417 | 1 | data_size: u64, |
418 | 1 | block_size: u64, |
419 | 1 | anchor_time: &SystemTime, |
420 | 1 | shared_context: &Arc<SharedContext>, |
421 | 1 | ) -> Result<(), Error> { |
422 | 1 | let key = key_from_file(file_name, file_type)?0 ; |
423 | | |
424 | 1 | let file_entry = Fe::create( |
425 | 1 | data_size, |
426 | 1 | block_size, |
427 | 1 | RwLock::new(EncodedFilePath { |
428 | 1 | shared_context: shared_context.clone(), |
429 | 1 | path_type: PathType::Content, |
430 | 1 | key: key.borrow().into_owned(), |
431 | 1 | }), |
432 | | ); |
433 | 1 | let time_since_anchor = anchor_time |
434 | 1 | .duration_since(atime) |
435 | 1 | .map_err(|_| make_input_err!("File access time newer than now"))?0 ; |
436 | 1 | evicting_map |
437 | 1 | .insert_with_time( |
438 | 1 | key.into_owned().into(), |
439 | 1 | Arc::new(file_entry), |
440 | 1 | time_since_anchor.as_secs() as i32, |
441 | 1 | ) |
442 | 1 | .await; |
443 | 1 | Ok(()) |
444 | 1 | } |
445 | | |
446 | 123 | async fn read_files( |
447 | 123 | folder: Option<&str>, |
448 | 123 | shared_context: &SharedContext, |
449 | 123 | ) -> Result<Vec<(String, SystemTime, u64, bool)>, Error> { |
450 | | // Note: In Dec 2024 this is for backwards compatibility with the old |
451 | | // way files were stored on disk. Previously all files were in a single |
452 | | // folder regardless of the StoreKey type. This allows old versions of |
453 | | // nativelink file layout to be upgraded at startup time. |
454 | | // This logic can be removed once more time has passed. |
455 | 123 | let read_dir = folder.map_or_else( |
456 | 41 | || format!("{}/", shared_context.content_path), |
457 | 82 | |folder| format!("{}/{folder}/", shared_context.content_path), |
458 | | ); |
459 | | |
460 | 123 | let (_permit, dir_handle) = fs::read_dir(read_dir) |
461 | 123 | .await |
462 | 123 | .err_tip(|| "Failed opening content directory for iterating in filesystem store")?0 |
463 | 123 | .into_inner(); |
464 | | |
465 | 123 | let read_dir_stream = ReadDirStream::new(dir_handle); |
466 | 123 | read_dir_stream |
467 | 123 | .map(|dir_entry| async move {83 |
468 | 83 | let dir_entry = dir_entry.unwrap(); |
469 | 83 | let file_name = dir_entry.file_name().into_string().unwrap(); |
470 | 83 | let metadata = dir_entry |
471 | 83 | .metadata() |
472 | 83 | .await |
473 | 83 | .err_tip(|| "Failed to get metadata in filesystem store")?0 ; |
474 | | // We need to filter out folders - we do not want to try to cache the s and d folders. |
475 | 83 | let is_file = |
476 | 83 | metadata.is_file() || !(file_name == STR_FOLDER82 || file_name == DIGEST_FOLDER41 ); Branch (476:21): [True: 0, False: 0]
Branch (476:45): [True: 0, False: 0]
Branch (476:21): [True: 1, False: 32]
Branch (476:45): [True: 16, False: 16]
Branch (476:21): [True: 0, False: 4]
Branch (476:45): [True: 2, False: 2]
Branch (476:21): [Folded - Ignored]
Branch (476:45): [Folded - Ignored]
Branch (476:21): [True: 0, False: 46]
Branch (476:45): [True: 23, False: 23]
|
477 | | // Using access time is not perfect, but better than random. We do not update the |
478 | | // atime when a file is actually "touched", we rely on whatever the filesystem does |
479 | | // when we read the file (usually update on read). |
480 | 83 | let atime = metadata |
481 | 83 | .accessed() |
482 | 83 | .or_else(|_| metadata0 .modified0 ()) |
483 | 83 | .unwrap_or(SystemTime::UNIX_EPOCH); |
484 | 83 | Result::<(String, SystemTime, u64, bool), Error>::Ok(( |
485 | 83 | file_name, |
486 | 83 | atime, |
487 | 83 | metadata.len(), |
488 | 83 | is_file, |
489 | 83 | )) |
490 | 166 | }) |
491 | 123 | .buffer_unordered(SIMULTANEOUS_METADATA_READS) |
492 | 123 | .try_collect() |
493 | 123 | .await |
494 | 123 | } |
495 | | |
496 | | /// Note: In Dec 2024 this is for backwards compatibility with the old |
497 | | /// way files were stored on disk. Previously all files were in a single |
498 | | /// folder regardless of the [`StoreKey`] type. This moves files from the old cache |
499 | | /// location to the new cache location, under [`DIGEST_FOLDER`]. |
500 | 41 | async fn move_old_cache( |
501 | 41 | shared_context: &Arc<SharedContext>, |
502 | 41 | rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, |
503 | 41 | ) -> Result<(), Error> { |
504 | 41 | let file_infos = read_files(None, shared_context).await?0 ; |
505 | | |
506 | 41 | let from_path = shared_context.content_path.to_string(); |
507 | | |
508 | 41 | let to_path = format!("{}/{DIGEST_FOLDER}", shared_context.content_path); |
509 | | |
510 | 41 | for (file_name0 , _, _, _) in file_infos.into_iter().filter(|x| x.3) { |
511 | 0 | let from_file: OsString = format!("{from_path}/{file_name}").into(); |
512 | 0 | let to_file: OsString = format!("{to_path}/{file_name}").into(); |
513 | | |
514 | 0 | if let Err(err) = rename_fn(&from_file, &to_file) { Branch (514:20): [True: 0, False: 0]
Branch (514:20): [True: 0, False: 0]
Branch (514:20): [True: 0, False: 0]
Branch (514:20): [Folded - Ignored]
Branch (514:20): [True: 0, False: 0]
|
515 | 0 | warn!(?from_file, ?to_file, ?err, "Failed to rename file",); |
516 | | } else { |
517 | 0 | debug!(?from_file, ?to_file, "Renamed file",); |
518 | | } |
519 | | } |
520 | 41 | Ok(()) |
521 | 41 | } |
522 | | |
523 | 82 | async fn add_files_to_cache<Fe: FileEntry>( |
524 | 82 | evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>, |
525 | 82 | anchor_time: &SystemTime, |
526 | 82 | shared_context: &Arc<SharedContext>, |
527 | 82 | block_size: u64, |
528 | 82 | folder: &str, |
529 | 82 | ) -> Result<(), Error> { |
530 | 82 | let file_infos = read_files(Some(folder), shared_context).await?0 ; |
531 | 82 | let file_type = match folder { |
532 | 82 | STR_FOLDER => FileType::String41 , |
533 | 41 | DIGEST_FOLDER => FileType::Digest, |
534 | 0 | _ => panic!("Invalid folder type"), |
535 | | }; |
536 | | |
537 | 82 | let path_root = format!("{}/{folder}", shared_context.content_path); |
538 | | |
539 | 82 | for (file_name1 , atime1 , data_size1 , _) in file_infos.into_iter().filter(|x| x.3) { |
540 | 1 | let result = process_entry( |
541 | 1 | evicting_map, |
542 | 1 | &file_name, |
543 | 1 | file_type, |
544 | 1 | atime, |
545 | 1 | data_size, |
546 | 1 | block_size, |
547 | 1 | anchor_time, |
548 | 1 | shared_context, |
549 | 1 | ) |
550 | 1 | .await; |
551 | 1 | if let Err(err0 ) = result { Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [True: 0, False: 1]
Branch (551:20): [True: 0, False: 0]
Branch (551:20): [Folded - Ignored]
Branch (551:20): [True: 0, False: 0]
|
552 | 0 | warn!(?file_name, ?err, "Failed to add file to eviction cache",); |
553 | | // Ignore result. |
554 | 0 | drop(fs::remove_file(format!("{path_root}/{file_name}")).await); |
555 | 1 | } |
556 | | } |
557 | 82 | Ok(()) |
558 | 82 | } |
559 | | |
560 | 41 | move_old_cache(shared_context, rename_fn).await?0 ; |
561 | | |
562 | 41 | add_files_to_cache( |
563 | 41 | evicting_map, |
564 | 41 | anchor_time, |
565 | 41 | shared_context, |
566 | 41 | block_size, |
567 | 41 | DIGEST_FOLDER, |
568 | 41 | ) |
569 | 41 | .await?0 ; |
570 | | |
571 | 41 | add_files_to_cache( |
572 | 41 | evicting_map, |
573 | 41 | anchor_time, |
574 | 41 | shared_context, |
575 | 41 | block_size, |
576 | 41 | STR_FOLDER, |
577 | 41 | ) |
578 | 41 | .await?0 ; |
579 | 41 | Ok(()) |
580 | 41 | } |
581 | | |
582 | 41 | async fn prune_temp_path(temp_path: &str) -> Result<(), Error> { |
583 | 82 | async fn prune_temp_inner(temp_path: &str, subpath: &str) -> Result<(), Error> { |
584 | 82 | let (_permit, dir_handle) = fs::read_dir(format!("{temp_path}/{subpath}")) |
585 | 82 | .await |
586 | 82 | .err_tip( |
587 | | || "Failed opening temp directory to prune partial downloads in filesystem store", |
588 | 0 | )? |
589 | 82 | .into_inner(); |
590 | | |
591 | 82 | let mut read_dir_stream = ReadDirStream::new(dir_handle); |
592 | 82 | while let Some(dir_entry0 ) = read_dir_stream.next().await { Branch (592:19): [True: 0, False: 0]
Branch (592:19): [True: 0, False: 32]
Branch (592:19): [True: 0, False: 4]
Branch (592:19): [Folded - Ignored]
Branch (592:19): [True: 0, False: 46]
|
593 | 0 | let path = dir_entry?.path(); |
594 | 0 | if let Err(err) = fs::remove_file(&path).await { Branch (594:20): [True: 0, False: 0]
Branch (594:20): [True: 0, False: 0]
Branch (594:20): [True: 0, False: 0]
Branch (594:20): [Folded - Ignored]
Branch (594:20): [True: 0, False: 0]
|
595 | 0 | warn!(?path, ?err, "Failed to delete file",); |
596 | 0 | } |
597 | | } |
598 | 82 | Ok(()) |
599 | 82 | } |
600 | | |
601 | 41 | prune_temp_inner(temp_path, STR_FOLDER).await?0 ; |
602 | 41 | prune_temp_inner(temp_path, DIGEST_FOLDER).await?0 ; |
603 | 41 | Ok(()) |
604 | 41 | } |
605 | | |
606 | | #[derive(Debug, MetricsComponent)] |
607 | | pub struct FilesystemStore<Fe: FileEntry = FileEntryImpl> { |
608 | | #[metric] |
609 | | shared_context: Arc<SharedContext>, |
610 | | #[metric(group = "evicting_map")] |
611 | | evicting_map: Arc<EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>>, |
612 | | #[metric(help = "Block size of the configured filesystem")] |
613 | | block_size: u64, |
614 | | #[metric(help = "Size of the configured read buffer size")] |
615 | | read_buffer_size: usize, |
616 | | weak_self: Weak<Self>, |
617 | | rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, |
618 | | } |
619 | | |
620 | | impl<Fe: FileEntry> FilesystemStore<Fe> { |
621 | 34 | pub async fn new(spec: &FilesystemSpec) -> Result<Arc<Self>, Error> { |
622 | 107 | Self::new_with_timeout_and_rename_fn34 (spec34 , |from, to| std::fs::rename(from, to)).await34 |
623 | 34 | } |
624 | | |
625 | 41 | pub async fn new_with_timeout_and_rename_fn( |
626 | 41 | spec: &FilesystemSpec, |
627 | 41 | rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>, |
628 | 41 | ) -> Result<Arc<Self>, Error> { |
629 | 82 | async fn create_subdirs(path: &str) -> Result<(), Error> { |
630 | 82 | fs::create_dir_all(format!("{path}/{STR_FOLDER}")) |
631 | 82 | .await |
632 | 82 | .err_tip(|| format!("Failed to create directory {path}/{STR_FOLDER}"0 ))?0 ; |
633 | 82 | fs::create_dir_all(format!("{path}/{DIGEST_FOLDER}")) |
634 | 82 | .await |
635 | 82 | .err_tip(|| format!("Failed to create directory {path}/{DIGEST_FOLDER}"0 )) |
636 | 82 | } |
637 | | |
638 | 41 | let now = SystemTime::now(); |
639 | | |
640 | 41 | let empty_policy = nativelink_config::stores::EvictionPolicy::default(); |
641 | 41 | let eviction_policy = spec.eviction_policy.as_ref().unwrap_or(&empty_policy); |
642 | 41 | let evicting_map = Arc::new(EvictingMap::new(eviction_policy, now)); |
643 | | |
644 | | // Create temp and content directories and the s and d subdirectories. |
645 | | |
646 | 41 | create_subdirs(&spec.temp_path).await?0 ; |
647 | 41 | create_subdirs(&spec.content_path).await?0 ; |
648 | | |
649 | 41 | let shared_context = Arc::new(SharedContext { |
650 | 41 | active_drop_spawns: AtomicU64::new(0), |
651 | 41 | temp_path: spec.temp_path.clone(), |
652 | 41 | content_path: spec.content_path.clone(), |
653 | 41 | }); |
654 | | |
655 | 41 | let block_size = if spec.block_size == 0 { Branch (655:29): [True: 0, False: 0]
Branch (655:29): [True: 0, False: 1]
Branch (655:29): [True: 0, False: 1]
Branch (655:29): [True: 1, False: 0]
Branch (655:29): [True: 0, False: 1]
Branch (655:29): [True: 1, False: 0]
Branch (655:29): [True: 10, False: 1]
Branch (655:29): [True: 2, False: 0]
Branch (655:29): [Folded - Ignored]
Branch (655:29): [True: 23, False: 0]
|
656 | 37 | DEFAULT_BLOCK_SIZE |
657 | | } else { |
658 | 4 | spec.block_size |
659 | | }; |
660 | 41 | add_files_to_cache( |
661 | 41 | evicting_map.as_ref(), |
662 | 41 | &now, |
663 | 41 | &shared_context, |
664 | 41 | block_size, |
665 | 41 | rename_fn, |
666 | 41 | ) |
667 | 41 | .await?0 ; |
668 | 41 | prune_temp_path(&shared_context.temp_path).await?0 ; |
669 | | |
670 | 41 | let read_buffer_size = if spec.read_buffer_size == 0 { Branch (670:35): [True: 0, False: 0]
Branch (670:35): [True: 1, False: 0]
Branch (670:35): [True: 0, False: 1]
Branch (670:35): [True: 1, False: 0]
Branch (670:35): [True: 0, False: 1]
Branch (670:35): [True: 1, False: 0]
Branch (670:35): [True: 4, False: 7]
Branch (670:35): [True: 2, False: 0]
Branch (670:35): [Folded - Ignored]
Branch (670:35): [True: 23, False: 0]
|
671 | 32 | DEFAULT_BUFF_SIZE |
672 | | } else { |
673 | 9 | spec.read_buffer_size as usize |
674 | | }; |
675 | 41 | Ok(Arc::new_cyclic(|weak_self| Self { |
676 | 41 | shared_context, |
677 | 41 | evicting_map, |
678 | 41 | block_size, |
679 | 41 | read_buffer_size, |
680 | 41 | weak_self: weak_self.clone(), |
681 | 41 | rename_fn, |
682 | 41 | })) |
683 | 41 | } |
684 | | |
685 | 24 | pub fn get_arc(&self) -> Option<Arc<Self>> { |
686 | 24 | self.weak_self.upgrade() |
687 | 24 | } |
688 | | |
689 | 9 | pub async fn get_file_entry_for_digest(&self, digest: &DigestInfo) -> Result<Arc<Fe>, Error> { |
690 | 9 | self.evicting_map |
691 | 9 | .get::<StoreKey<'static>>(&digest.into()) |
692 | 9 | .await |
693 | 9 | .ok_or_else(|| make_err!(Code::NotFound0 , "{digest} not found in filesystem store")) |
694 | 9 | } |
695 | | |
696 | 108 | async fn update_file( |
697 | 108 | self: Pin<&Self>, |
698 | 108 | mut entry: Fe, |
699 | 108 | mut temp_file: fs::FileSlot, |
700 | 108 | final_key: StoreKey<'static>, |
701 | 108 | mut reader: DropCloserReadHalf, |
702 | 108 | ) -> Result<(), Error> { |
703 | 108 | let mut data_size = 0; |
704 | | loop { |
705 | 183 | let mut data = reader |
706 | 183 | .recv() |
707 | 183 | .await |
708 | 183 | .err_tip(|| "Failed to receive data in filesystem store")?0 ; |
709 | 183 | let data_len = data.len(); |
710 | 183 | if data_len == 0 { Branch (710:16): [True: 0, False: 0]
Branch (710:16): [True: 2, False: 2]
Branch (710:16): [True: 2, False: 2]
Branch (710:16): [True: 2, False: 2]
Branch (710:16): [True: 2, False: 2]
Branch (710:16): [True: 1, False: 1]
Branch (710:16): [True: 13, False: 11]
Branch (710:16): [True: 0, False: 0]
Branch (710:16): [Folded - Ignored]
Branch (710:16): [True: 86, False: 55]
|
711 | 108 | break; // EOF. |
712 | 75 | } |
713 | 75 | temp_file |
714 | 75 | .write_all_buf(&mut data) |
715 | 75 | .await |
716 | 75 | .err_tip(|| "Failed to write data into filesystem store")?0 ; |
717 | 75 | data_size += data_len as u64; |
718 | | } |
719 | | |
720 | 108 | temp_file |
721 | 108 | .as_ref() |
722 | 108 | .sync_all() |
723 | 108 | .await |
724 | 108 | .err_tip(|| "Failed to sync_data in filesystem store")?0 ; |
725 | | |
726 | 108 | drop(temp_file); |
727 | | |
728 | 108 | *entry.data_size_mut() = data_size; |
729 | 108 | self.emplace_file(final_key, Arc::new(entry)).await |
730 | 107 | } |
731 | | |
732 | 117 | async fn emplace_file(&self, key: StoreKey<'static>, entry: Arc<Fe>) -> Result<(), Error> { |
733 | | // This sequence of events is quite ticky to understand due to the amount of triggers that |
734 | | // happen, async'ness of it and the locking. So here is a breakdown of what happens: |
735 | | // 1. Here will hold a write lock on any file operations of this FileEntry. |
736 | | // 2. Then insert the entry into the evicting map. This may trigger an eviction of other |
737 | | // entries. |
738 | | // 3. Eviction triggers `unref()`, which grabs a write lock on the evicted FileEntry |
739 | | // during the rename. |
740 | | // 4. It should be impossible for items to be added while eviction is happening, so there |
741 | | // should not be a deadlock possibility. However, it is possible for the new FileEntry |
742 | | // to be evicted before the file is moved into place. Eviction of the newly inserted |
743 | | // item is not possible within the `insert()` call because the write lock inside the |
744 | | // eviction map. If an eviction of new item happens after `insert()` but before |
745 | | // `rename()` then we get to finish our operation because the `unref()` of the new item |
746 | | // will be blocked on us because we currently have the lock. |
747 | | // 5. Move the file into place. Since we hold a write lock still anyone that gets our new |
748 | | // FileEntry (which has not yet been placed on disk) will not be able to read the file's |
749 | | // contents until we release the lock. |
750 | 117 | let evicting_map = self.evicting_map.clone(); |
751 | 117 | let rename_fn = self.rename_fn; |
752 | | |
753 | | // We need to guarantee that this will get to the end even if the parent future is dropped. |
754 | | // See: https://github.com/TraceMachina/nativelink/issues/495 |
755 | 117 | background_spawn!("filesystem_store_emplace_file", async move { |
756 | 117 | let mut encoded_file_path = entry.get_encoded_file_path().write().await; |
757 | 117 | let final_path = get_file_path_raw( |
758 | 117 | &PathType::Content, |
759 | 117 | encoded_file_path.shared_context.as_ref(), |
760 | 117 | &key, |
761 | | ); |
762 | | |
763 | 117 | evicting_map |
764 | 117 | .insert(key.borrow().into_owned().into(), entry.clone()) |
765 | 117 | .await; |
766 | | |
767 | 117 | let from_path = encoded_file_path.get_file_path(); |
768 | | // Internally tokio spawns fs commands onto a blocking thread anyways. |
769 | | // Since we are already on a blocking thread, we just need the `fs` wrapper to manage |
770 | | // an open-file permit (ensure we don't open too many files at once). |
771 | 117 | let result = (rename_fn)(&from_path, &final_path).err_tip(|| {1 |
772 | 1 | format!( |
773 | 1 | "Failed to rename temp file to final path {}", |
774 | 1 | final_path.display() |
775 | | ) |
776 | 1 | }); |
777 | | |
778 | | // In the event our move from temp file to final file fails we need to ensure we remove |
779 | | // the entry from our map. |
780 | | // Remember: At this point it is possible for another thread to have a reference to |
781 | | // `entry`, so we can't delete the file, only drop() should ever delete files. |
782 | 117 | if let Err(err1 ) = result { Branch (782:20): [True: 0, False: 0]
Branch (782:20): [True: 0, False: 2]
Branch (782:20): [True: 0, False: 2]
Branch (782:20): [True: 0, False: 2]
Branch (782:20): [True: 0, False: 2]
Branch (782:20): [True: 1, False: 0]
Branch (782:20): [True: 0, False: 15]
Branch (782:20): [True: 0, False: 0]
Branch (782:20): [Folded - Ignored]
Branch (782:20): [True: 0, False: 93]
|
783 | 1 | error!(?err, ?from_path, ?final_path, "Failed to rename file",); |
784 | | // Warning: To prevent deadlock we need to release our lock or during `remove_if()` |
785 | | // it will call `unref()`, which triggers a write-lock on `encoded_file_path`. |
786 | 1 | drop(encoded_file_path); |
787 | | // It is possible that the item in our map is no longer the item we inserted, |
788 | | // So, we need to conditionally remove it only if the pointers are the same. |
789 | | |
790 | 1 | evicting_map |
791 | 1 | .remove_if(&key, |map_entry| Arc::<Fe>::ptr_eq(map_entry, &entry)) |
792 | 1 | .await; |
793 | 1 | return Err(err); |
794 | 116 | } |
795 | 116 | encoded_file_path.path_type = PathType::Content; |
796 | 116 | encoded_file_path.key = key; |
797 | 116 | Ok(()) |
798 | 117 | }) |
799 | 117 | .await |
800 | 116 | .err_tip(|| "Failed to create spawn in filesystem store update_file")?0 |
801 | 116 | } |
802 | | } |
803 | | |
804 | | #[async_trait] |
805 | | impl<Fe: FileEntry> StoreDriver for FilesystemStore<Fe> { |
806 | | async fn has_with_results( |
807 | | self: Pin<&Self>, |
808 | | keys: &[StoreKey<'_>], |
809 | | results: &mut [Option<u64>], |
810 | 170 | ) -> Result<(), Error> { |
811 | 85 | self.evicting_map |
812 | 85 | .sizes_for_keys::<_, StoreKey<'_>, &StoreKey<'_>>( |
813 | 85 | keys.iter(), |
814 | 85 | results, |
815 | 85 | false, /* peek */ |
816 | 85 | ) |
817 | 85 | .await; |
818 | | // We need to do a special pass to ensure our zero files exist. |
819 | | // If our results failed and the result was a zero file, we need to |
820 | | // create the file by spec. |
821 | 85 | for (key, result) in keys.iter().zip(results.iter_mut()) { |
822 | 85 | if result.is_some() || !16 is_zero_digest16 (key.borrow()) { Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [True: 0, False: 1]
Branch (822:36): [True: 1, False: 0]
Branch (822:16): [True: 1, False: 2]
Branch (822:36): [True: 0, False: 2]
Branch (822:16): [True: 0, False: 0]
Branch (822:36): [True: 0, False: 0]
Branch (822:16): [Folded - Ignored]
Branch (822:36): [Folded - Ignored]
Branch (822:16): [True: 68, False: 13]
Branch (822:36): [True: 13, False: 0]
|
823 | 83 | continue; |
824 | 2 | } |
825 | 2 | let (mut tx, rx) = make_buf_channel_pair(); |
826 | 2 | let send_eof_result = tx.send_eof(); |
827 | 2 | self.update(key.borrow(), rx, UploadSizeInfo::ExactSize(0)) |
828 | 2 | .await |
829 | 2 | .err_tip(|| format!("Failed to create zero file for key {}"0 , key0 .as_str0 ())) |
830 | 2 | .merge( |
831 | 2 | send_eof_result |
832 | 2 | .err_tip(|| "Failed to send zero file EOF in filesystem store has"), |
833 | 0 | )?; |
834 | | |
835 | 2 | *result = Some(0); |
836 | | } |
837 | 85 | Ok(()) |
838 | 170 | } |
839 | | |
840 | | async fn update( |
841 | | self: Pin<&Self>, |
842 | | key: StoreKey<'_>, |
843 | | reader: DropCloserReadHalf, |
844 | | _upload_size: UploadSizeInfo, |
845 | 216 | ) -> Result<(), Error> { |
846 | 108 | let temp_key = make_temp_key(&key); |
847 | 108 | let (entry, temp_file, temp_full_path) = Fe::make_and_open_file( |
848 | 108 | self.block_size, |
849 | 108 | EncodedFilePath { |
850 | 108 | shared_context: self.shared_context.clone(), |
851 | 108 | path_type: PathType::Temp, |
852 | 108 | key: temp_key, |
853 | 108 | }, |
854 | 108 | ) |
855 | 108 | .await?0 ; |
856 | | |
857 | 108 | self.update_file(entry, temp_file, key.into_owned(), reader) |
858 | 108 | .await |
859 | 107 | .err_tip(|| {1 |
860 | 1 | format!( |
861 | 1 | "While processing with temp file {}", |
862 | 1 | temp_full_path.display() |
863 | | ) |
864 | 1 | }) |
865 | 215 | } |
866 | | |
867 | 84 | fn optimized_for(&self, optimization: StoreOptimizations) -> bool { |
868 | 84 | optimization == StoreOptimizations::FileUpdates |
869 | 84 | } |
870 | | |
871 | | async fn update_with_whole_file( |
872 | | self: Pin<&Self>, |
873 | | key: StoreKey<'_>, |
874 | | path: OsString, |
875 | | file: fs::FileSlot, |
876 | | upload_size: UploadSizeInfo, |
877 | 18 | ) -> Result<Option<fs::FileSlot>, Error> { |
878 | 9 | let file_size = match upload_size { |
879 | 9 | UploadSizeInfo::ExactSize(size) => size, |
880 | 0 | UploadSizeInfo::MaxSize(_) => file |
881 | 0 | .as_ref() |
882 | 0 | .metadata() |
883 | 0 | .await |
884 | 0 | .err_tip(|| format!("While reading metadata for {}", path.display()))? |
885 | 0 | .len(), |
886 | | }; |
887 | 9 | let entry = Fe::create( |
888 | 9 | file_size, |
889 | 9 | self.block_size, |
890 | 9 | RwLock::new(EncodedFilePath { |
891 | 9 | shared_context: self.shared_context.clone(), |
892 | 9 | path_type: PathType::Custom(path), |
893 | 9 | key: key.borrow().into_owned(), |
894 | 9 | }), |
895 | | ); |
896 | | // We are done with the file, if we hold a reference to the file here, it could |
897 | | // result in a deadlock if `emplace_file()` also needs file descriptors. |
898 | 9 | drop(file); |
899 | 9 | self.emplace_file(key.into_owned(), Arc::new(entry)) |
900 | 9 | .await |
901 | 9 | .err_tip(|| "Could not move file into store in upload_file_to_store, maybe dest is on different volume?")?0 ; |
902 | 9 | return Ok(None); |
903 | 18 | } |
904 | | |
905 | | async fn get_part( |
906 | | self: Pin<&Self>, |
907 | | key: StoreKey<'_>, |
908 | | writer: &mut DropCloserWriteHalf, |
909 | | offset: u64, |
910 | | length: Option<u64>, |
911 | 120 | ) -> Result<(), Error> { |
912 | 60 | if is_zero_digest(key.borrow()) { Branch (912:12): [True: 0, False: 0]
Branch (912:12): [True: 0, False: 0]
Branch (912:12): [True: 0, False: 1]
Branch (912:12): [True: 0, False: 0]
Branch (912:12): [True: 0, False: 1]
Branch (912:12): [True: 0, False: 0]
Branch (912:12): [True: 1, False: 4]
Branch (912:12): [True: 0, False: 0]
Branch (912:12): [Folded - Ignored]
Branch (912:12): [True: 15, False: 38]
|
913 | 16 | self.has(key.borrow()) |
914 | 16 | .await |
915 | 16 | .err_tip(|| "Failed to check if zero digest exists in filesystem store")?0 ; |
916 | 16 | writer |
917 | 16 | .send_eof() |
918 | 16 | .err_tip(|| "Failed to send zero EOF in filesystem store get_part")?0 ; |
919 | 16 | return Ok(()); |
920 | 44 | } |
921 | 44 | let entry = self.evicting_map.get(&key).await.ok_or_else(|| {0 |
922 | 0 | make_err!( |
923 | 0 | Code::NotFound, |
924 | | "{} not found in filesystem store here", |
925 | 0 | key.as_str() |
926 | | ) |
927 | 0 | })?; |
928 | 44 | let read_limit = length.unwrap_or(u64::MAX); |
929 | 44 | let mut temp_file42 = entry.read_file_part(offset, read_limit).or_else(|err| async move {2 |
930 | | // If the file is not found, we need to remove it from the eviction map. |
931 | 2 | if err.code == Code::NotFound { Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [True: 2, False: 0]
Branch (931:16): [True: 0, False: 0]
Branch (931:16): [Folded - Ignored]
Branch (931:16): [True: 0, False: 0]
|
932 | 2 | error!( |
933 | | ?err, |
934 | | ?key, |
935 | 2 | "Entry was in our map, but not found on disk. Removing from map as a precaution, but process probably need restarted." |
936 | | ); |
937 | 2 | self.evicting_map.remove(&key).await; |
938 | 0 | } |
939 | 2 | Err(err) |
940 | 44 | }4 ).await?2 ; |
941 | | |
942 | | loop { |
943 | 1.12k | let mut buf = BytesMut::with_capacity(self.read_buffer_size); |
944 | 1.12k | temp_file |
945 | 1.12k | .read_buf(&mut buf) |
946 | 1.12k | .await |
947 | 1.12k | .err_tip(|| "Failed to read data in filesystem store")?0 ; |
948 | 1.12k | if buf.is_empty() { Branch (948:16): [True: 0, False: 0]
Branch (948:16): [True: 0, False: 0]
Branch (948:16): [True: 1, False: 10]
Branch (948:16): [True: 0, False: 0]
Branch (948:16): [True: 1, False: 10]
Branch (948:16): [True: 0, False: 0]
Branch (948:16): [True: 1, False: 1.02k]
Branch (948:16): [True: 0, False: 0]
Branch (948:16): [Folded - Ignored]
Branch (948:16): [True: 38, False: 38]
|
949 | 41 | break; // EOF. |
950 | 1.08k | } |
951 | 1.08k | writer |
952 | 1.08k | .send(buf.freeze()) |
953 | 1.08k | .await |
954 | 1.08k | .err_tip(|| "Failed to send chunk in filesystem store get_part")?0 ; |
955 | | } |
956 | 41 | writer |
957 | 41 | .send_eof() |
958 | 41 | .err_tip(|| "Filed to send EOF in filesystem store get_part")?0 ; |
959 | | |
960 | 41 | Ok(()) |
961 | 119 | } |
962 | | |
963 | 101 | fn inner_store(&self, _digest: Option<StoreKey>) -> &dyn StoreDriver { |
964 | 101 | self |
965 | 101 | } |
966 | | |
967 | 24 | fn as_any<'a>(&'a self) -> &'a (dyn core::any::Any + Sync + Send + 'static) { |
968 | 24 | self |
969 | 24 | } |
970 | | |
971 | 0 | fn as_any_arc(self: Arc<Self>) -> Arc<dyn core::any::Any + Sync + Send + 'static> { |
972 | 0 | self |
973 | 0 | } |
974 | | |
975 | 0 | fn register_health(self: Arc<Self>, registry: &mut HealthRegistryBuilder) { |
976 | 0 | registry.register_indicator(self); |
977 | 0 | } |
978 | | } |
979 | | |
980 | | #[async_trait] |
981 | | impl<Fe: FileEntry> HealthStatusIndicator for FilesystemStore<Fe> { |
982 | 0 | fn get_name(&self) -> &'static str { |
983 | 0 | "FilesystemStore" |
984 | 0 | } |
985 | | |
986 | 0 | async fn check_health(&self, namespace: Cow<'static, str>) -> HealthStatus { |
987 | 0 | StoreDriver::check_health(Pin::new(self), namespace).await |
988 | 0 | } |
989 | | } |