1 //! Background worker that watches over the cache. 2 //! 3 //! It cleans up old cache, updates statistics and optimizes the cache. 4 //! We allow losing some messages (it doesn't hurt) and some races, 5 //! but we guarantee eventual consistency and fault tolerancy. 6 //! Background tasks can be CPU intensive, but the worker thread has low priority. 7 8 use super::{fs_write_atomic, CacheConfig}; 9 use log::{debug, info, trace, warn}; 10 use serde::{Deserialize, Serialize}; 11 use std::cmp; 12 use std::collections::HashMap; 13 use std::ffi::OsStr; 14 use std::fmt; 15 use std::fs; 16 use std::path::{Path, PathBuf}; 17 use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; 18 #[cfg(test)] 19 use std::sync::{Arc, Condvar, Mutex}; 20 use std::thread; 21 use std::time::Duration; 22 #[cfg(not(test))] 23 use std::time::SystemTime; 24 #[cfg(test)] 25 use tests::system_time_stub::SystemTimeStub as SystemTime; 26 27 #[derive(Clone)] 28 pub(super) struct Worker { 29 sender: SyncSender<CacheEvent>, 30 #[cfg(test)] 31 stats: Arc<(Mutex<WorkerStats>, Condvar)>, 32 } 33 34 struct WorkerThread { 35 receiver: Receiver<CacheEvent>, 36 cache_config: CacheConfig, 37 #[cfg(test)] 38 stats: Arc<(Mutex<WorkerStats>, Condvar)>, 39 } 40 41 #[cfg(test)] 42 #[derive(Default)] 43 struct WorkerStats { 44 dropped: u32, 45 sent: u32, 46 handled: u32, 47 } 48 49 #[derive(Debug, Clone)] 50 enum CacheEvent { 51 OnCacheGet(PathBuf), 52 OnCacheUpdate(PathBuf), 53 } 54 55 impl Worker { 56 pub(super) fn start_new( 57 cache_config: &CacheConfig, 58 init_file_per_thread_logger: Option<&'static str>, 59 ) -> Self { 60 let queue_size = match cache_config.worker_event_queue_size() { 61 num if num <= usize::max_value() as u64 => num as usize, 62 _ => usize::max_value(), 63 }; 64 let (tx, rx) = sync_channel(queue_size); 65 66 #[cfg(test)] 67 let stats = Arc::new((Mutex::new(WorkerStats::default()), Condvar::new())); 68 69 let worker_thread = WorkerThread { 70 receiver: rx, 71 cache_config: cache_config.clone(), 72 #[cfg(test)] 73 stats: stats.clone(), 74 }; 75 76 // when self is dropped, sender will be dropped, what will cause the channel 77 // to hang, and the worker thread to exit -- it happens in the tests 78 // non-tests binary has only a static worker, so Rust doesn't drop it 79 thread::spawn(move || worker_thread.run(init_file_per_thread_logger)); 80 81 Self { 82 sender: tx, 83 #[cfg(test)] 84 stats, 85 } 86 } 87 88 pub(super) fn on_cache_get_async(&self, path: impl AsRef<Path>) { 89 let event = CacheEvent::OnCacheGet(path.as_ref().to_path_buf()); 90 self.send_cache_event(event); 91 } 92 93 pub(super) fn on_cache_update_async(&self, path: impl AsRef<Path>) { 94 let event = CacheEvent::OnCacheUpdate(path.as_ref().to_path_buf()); 95 self.send_cache_event(event); 96 } 97 98 #[inline] 99 fn send_cache_event(&self, event: CacheEvent) { 100 let sent_event = self.sender.try_send(event.clone()); 101 102 if let Err(ref err) = sent_event { 103 info!( 104 "Failed to send asynchronously message to worker thread, \ 105 event: {:?}, error: {}", 106 event, err 107 ); 108 } 109 110 #[cfg(test)] 111 { 112 let mut stats = self 113 .stats 114 .0 115 .lock() 116 .expect("Failed to acquire worker stats lock"); 117 118 if sent_event.is_ok() { 119 stats.sent += 1; 120 } else { 121 stats.dropped += 1; 122 } 123 } 124 } 125 126 #[cfg(test)] 127 pub(super) fn events_dropped(&self) -> u32 { 128 let stats = self 129 .stats 130 .0 131 .lock() 132 .expect("Failed to acquire worker stats lock"); 133 stats.dropped 134 } 135 136 #[cfg(test)] 137 pub(super) fn wait_for_all_events_handled(&self) { 138 let (stats, condvar) = &*self.stats; 139 let mut stats = stats.lock().expect("Failed to acquire worker stats lock"); 140 while stats.handled != stats.sent { 141 stats = condvar 142 .wait(stats) 143 .expect("Failed to reacquire worker stats lock"); 144 } 145 } 146 } 147 148 impl fmt::Debug for Worker { 149 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { 150 f.debug_struct("Worker").finish() 151 } 152 } 153 154 #[derive(Serialize, Deserialize)] 155 struct ModuleCacheStatistics { 156 pub usages: u64, 157 #[serde(rename = "optimized-compression")] 158 pub compression_level: i32, 159 } 160 161 impl ModuleCacheStatistics { 162 fn default(cache_config: &CacheConfig) -> Self { 163 Self { 164 usages: 0, 165 compression_level: cache_config.baseline_compression_level(), 166 } 167 } 168 } 169 170 enum CacheEntry { 171 Recognized { 172 path: PathBuf, 173 mtime: SystemTime, 174 size: u64, 175 }, 176 Unrecognized { 177 path: PathBuf, 178 is_dir: bool, 179 }, 180 } 181 182 macro_rules! unwrap_or_warn { 183 ($result:expr, $cont:stmt, $err_msg:expr, $path:expr) => { 184 match $result { 185 Ok(val) => val, 186 Err(err) => { 187 warn!("{}, path: {}, msg: {}", $err_msg, $path.display(), err); 188 $cont 189 } 190 } 191 }; 192 } 193 194 impl WorkerThread { 195 fn run(self, init_file_per_thread_logger: Option<&'static str>) { 196 if let Some(prefix) = init_file_per_thread_logger { 197 file_per_thread_logger::initialize(prefix); 198 } 199 200 debug!("Cache worker thread started."); 201 202 Self::lower_thread_priority(); 203 204 #[cfg(test)] 205 let (stats, condvar) = &*self.stats; 206 207 for event in self.receiver.iter() { 208 match event { 209 CacheEvent::OnCacheGet(path) => self.handle_on_cache_get(path), 210 CacheEvent::OnCacheUpdate(path) => self.handle_on_cache_update(path), 211 } 212 213 #[cfg(test)] 214 { 215 let mut stats = stats.lock().expect("Failed to acquire worker stats lock"); 216 stats.handled += 1; 217 condvar.notify_all(); 218 } 219 } 220 } 221 222 #[cfg(target_os = "fuchsia")] 223 fn lower_thread_priority() { 224 // TODO This needs to use Fuchsia thread profiles 225 // https://fuchsia.dev/fuchsia-src/reference/kernel_objects/profile 226 warn!( 227 "Lowering thread priority on Fuchsia is currently a noop. It might affect application performance." 228 ); 229 } 230 231 #[cfg(target_os = "windows")] 232 fn lower_thread_priority() { 233 use std::convert::TryInto; 234 use winapi::um::processthreadsapi::{GetCurrentThread, SetThreadPriority}; 235 use winapi::um::winbase::THREAD_MODE_BACKGROUND_BEGIN; 236 237 // https://docs.microsoft.com/en-us/windows/win32/api/processthreadsapi/nf-processthreadsapi-setthreadpriority 238 // https://docs.microsoft.com/en-us/windows/win32/procthread/scheduling-priorities 239 240 if unsafe { 241 SetThreadPriority( 242 GetCurrentThread(), 243 THREAD_MODE_BACKGROUND_BEGIN.try_into().unwrap(), 244 ) 245 } == 0 246 { 247 warn!( 248 "Failed to lower worker thread priority. It might affect application performance." 249 ); 250 } 251 } 252 253 #[cfg(not(any(target_os = "windows", target_os = "fuchsia")))] 254 fn lower_thread_priority() { 255 // http://man7.org/linux/man-pages/man7/sched.7.html 256 257 const NICE_DELTA_FOR_BACKGROUND_TASKS: i32 = 3; 258 259 match rustix::process::nice(NICE_DELTA_FOR_BACKGROUND_TASKS) { 260 Ok(current_nice) => { 261 debug!("New nice value of worker thread: {}", current_nice); 262 } 263 Err(err) => { 264 warn!( 265 "Failed to lower worker thread priority ({:?}). It might affect application performance.", err); 266 } 267 }; 268 } 269 270 /// Increases the usage counter and recompresses the file 271 /// if the usage counter reached configurable treshold. 272 fn handle_on_cache_get(&self, path: PathBuf) { 273 trace!("handle_on_cache_get() for path: {}", path.display()); 274 275 // construct .stats file path 276 let filename = path.file_name().unwrap().to_str().unwrap(); 277 let stats_path = path.with_file_name(format!("{}.stats", filename)); 278 279 // load .stats file (default if none or error) 280 let mut stats = read_stats_file(stats_path.as_ref()) 281 .unwrap_or_else(|| ModuleCacheStatistics::default(&self.cache_config)); 282 283 // step 1: update the usage counter & write to the disk 284 // it's racy, but it's fine (the counter will be just smaller, 285 // sometimes will retrigger recompression) 286 stats.usages += 1; 287 if !write_stats_file(stats_path.as_ref(), &stats) { 288 return; 289 } 290 291 // step 2: recompress if there's a need 292 let opt_compr_lvl = self.cache_config.optimized_compression_level(); 293 if stats.compression_level >= opt_compr_lvl 294 || stats.usages 295 < self 296 .cache_config 297 .optimized_compression_usage_counter_threshold() 298 { 299 return; 300 } 301 302 let lock_path = if let Some(p) = acquire_task_fs_lock( 303 path.as_ref(), 304 self.cache_config.optimizing_compression_task_timeout(), 305 self.cache_config 306 .allowed_clock_drift_for_files_from_future(), 307 ) { 308 p 309 } else { 310 return; 311 }; 312 313 trace!("Trying to recompress file: {}", path.display()); 314 315 // recompress, write to other file, rename (it's atomic file content exchange) 316 // and update the stats file 317 let compressed_cache_bytes = unwrap_or_warn!( 318 fs::read(&path), 319 return, 320 "Failed to read old cache file", 321 path 322 ); 323 324 let cache_bytes = unwrap_or_warn!( 325 zstd::decode_all(&compressed_cache_bytes[..]), 326 return, 327 "Failed to decompress cached code", 328 path 329 ); 330 331 let recompressed_cache_bytes = unwrap_or_warn!( 332 zstd::encode_all(&cache_bytes[..], opt_compr_lvl), 333 return, 334 "Failed to compress cached code", 335 path 336 ); 337 338 unwrap_or_warn!( 339 fs::write(&lock_path, &recompressed_cache_bytes), 340 return, 341 "Failed to write recompressed cache", 342 lock_path 343 ); 344 345 unwrap_or_warn!( 346 fs::rename(&lock_path, &path), 347 { 348 if let Err(error) = fs::remove_file(&lock_path) { 349 warn!( 350 "Failed to clean up (remove) recompressed cache, path {}, err: {}", 351 lock_path.display(), 352 error 353 ); 354 } 355 356 return; 357 }, 358 "Failed to rename recompressed cache", 359 lock_path 360 ); 361 362 // update stats file (reload it! recompression can take some time) 363 if let Some(mut new_stats) = read_stats_file(stats_path.as_ref()) { 364 if new_stats.compression_level >= opt_compr_lvl { 365 // Rare race: 366 // two instances with different opt_compr_lvl: we don't know in which order they updated 367 // the cache file and the stats file (they are not updated together atomically) 368 // Possible solution is to use directories per cache entry, but it complicates the system 369 // and is not worth it. 370 debug!( 371 "DETECTED task did more than once (or race with new file): \ 372 recompression of {}. Note: if optimized compression level setting \ 373 has changed in the meantine, the stats file might contain \ 374 inconsistent compression level due to race.", 375 path.display() 376 ); 377 } else { 378 new_stats.compression_level = opt_compr_lvl; 379 let _ = write_stats_file(stats_path.as_ref(), &new_stats); 380 } 381 382 if new_stats.usages < stats.usages { 383 debug!( 384 "DETECTED lower usage count (new file or race with counter \ 385 increasing): file {}", 386 path.display() 387 ); 388 } 389 } else { 390 debug!( 391 "Can't read stats file again to update compression level (it might got \ 392 cleaned up): file {}", 393 stats_path.display() 394 ); 395 } 396 397 trace!("Task finished: recompress file: {}", path.display()); 398 } 399 400 fn handle_on_cache_update(&self, path: PathBuf) { 401 trace!("handle_on_cache_update() for path: {}", path.display()); 402 403 // ---------------------- step 1: create .stats file 404 405 // construct .stats file path 406 let filename = path 407 .file_name() 408 .expect("Expected valid cache file name") 409 .to_str() 410 .expect("Expected valid cache file name"); 411 let stats_path = path.with_file_name(format!("{}.stats", filename)); 412 413 // create and write stats file 414 let mut stats = ModuleCacheStatistics::default(&self.cache_config); 415 stats.usages += 1; 416 write_stats_file(&stats_path, &stats); 417 418 // ---------------------- step 2: perform cleanup task if needed 419 420 // acquire lock for cleanup task 421 // Lock is a proof of recent cleanup task, so we don't want to delete them. 422 // Expired locks will be deleted by the cleanup task. 423 let cleanup_file = self.cache_config.directory().join(".cleanup"); // some non existing marker file 424 if acquire_task_fs_lock( 425 &cleanup_file, 426 self.cache_config.cleanup_interval(), 427 self.cache_config 428 .allowed_clock_drift_for_files_from_future(), 429 ) 430 .is_none() 431 { 432 return; 433 } 434 435 trace!("Trying to clean up cache"); 436 437 let mut cache_index = self.list_cache_contents(); 438 let future_tolerance = SystemTime::now() 439 .checked_add( 440 self.cache_config 441 .allowed_clock_drift_for_files_from_future(), 442 ) 443 .expect("Brace your cache, the next Big Bang is coming (time overflow)"); 444 cache_index.sort_unstable_by(|lhs, rhs| { 445 // sort by age 446 use CacheEntry::*; 447 match (lhs, rhs) { 448 (Recognized { mtime: lhs_mt, .. }, Recognized { mtime: rhs_mt, .. }) => { 449 match (*lhs_mt > future_tolerance, *rhs_mt > future_tolerance) { 450 // later == younger 451 (false, false) => rhs_mt.cmp(lhs_mt), 452 // files from far future are treated as oldest recognized files 453 // we want to delete them, so the cache keeps track of recent files 454 // however, we don't delete them uncodintionally, 455 // because .stats file can be overwritten with a meaningful mtime 456 (true, false) => cmp::Ordering::Greater, 457 (false, true) => cmp::Ordering::Less, 458 (true, true) => cmp::Ordering::Equal, 459 } 460 } 461 // unrecognized is kind of infinity 462 (Recognized { .. }, Unrecognized { .. }) => cmp::Ordering::Less, 463 (Unrecognized { .. }, Recognized { .. }) => cmp::Ordering::Greater, 464 (Unrecognized { .. }, Unrecognized { .. }) => cmp::Ordering::Equal, 465 } 466 }); 467 468 // find "cut" boundary: 469 // - remove unrecognized files anyway, 470 // - remove some cache files if some quota has been exceeded 471 let mut total_size = 0u64; 472 let mut start_delete_idx = None; 473 let mut start_delete_idx_if_deleting_recognized_items: Option<usize> = None; 474 475 let total_size_limit = self.cache_config.files_total_size_soft_limit(); 476 let file_count_limit = self.cache_config.file_count_soft_limit(); 477 let tsl_if_deleting = total_size_limit 478 .checked_mul( 479 self.cache_config 480 .files_total_size_limit_percent_if_deleting() as u64, 481 ) 482 .unwrap() 483 / 100; 484 let fcl_if_deleting = file_count_limit 485 .checked_mul(self.cache_config.file_count_limit_percent_if_deleting() as u64) 486 .unwrap() 487 / 100; 488 489 for (idx, item) in cache_index.iter().enumerate() { 490 let size = if let CacheEntry::Recognized { size, .. } = item { 491 size 492 } else { 493 start_delete_idx = Some(idx); 494 break; 495 }; 496 497 total_size += size; 498 if start_delete_idx_if_deleting_recognized_items.is_none() 499 && (total_size > tsl_if_deleting || (idx + 1) as u64 > fcl_if_deleting) 500 { 501 start_delete_idx_if_deleting_recognized_items = Some(idx); 502 } 503 504 if total_size > total_size_limit || (idx + 1) as u64 > file_count_limit { 505 start_delete_idx = start_delete_idx_if_deleting_recognized_items; 506 break; 507 } 508 } 509 510 if let Some(idx) = start_delete_idx { 511 for item in &cache_index[idx..] { 512 let (result, path, entity) = match item { 513 CacheEntry::Recognized { path, .. } 514 | CacheEntry::Unrecognized { 515 path, 516 is_dir: false, 517 } => (fs::remove_file(path), path, "file"), 518 CacheEntry::Unrecognized { path, is_dir: true } => { 519 (fs::remove_dir_all(path), path, "directory") 520 } 521 }; 522 if let Err(err) = result { 523 warn!( 524 "Failed to remove {} during cleanup, path: {}, err: {}", 525 entity, 526 path.display(), 527 err 528 ); 529 } 530 } 531 } 532 533 trace!("Task finished: clean up cache"); 534 } 535 536 // Be fault tolerant: list as much as you can, and ignore the rest 537 fn list_cache_contents(&self) -> Vec<CacheEntry> { 538 fn enter_dir( 539 vec: &mut Vec<CacheEntry>, 540 dir_path: &Path, 541 level: u8, 542 cache_config: &CacheConfig, 543 ) { 544 macro_rules! add_unrecognized { 545 (file: $path:expr) => { 546 add_unrecognized!(false, $path) 547 }; 548 (dir: $path:expr) => { 549 add_unrecognized!(true, $path) 550 }; 551 ($is_dir:expr, $path:expr) => { 552 vec.push(CacheEntry::Unrecognized { 553 path: $path.to_path_buf(), 554 is_dir: $is_dir, 555 }) 556 }; 557 } 558 macro_rules! add_unrecognized_and { 559 ([ $( $ty:ident: $path:expr ),* ], $cont:stmt) => {{ 560 $( add_unrecognized!($ty: $path); )* 561 $cont 562 }}; 563 } 564 565 macro_rules! unwrap_or { 566 ($result:expr, $cont:stmt, $err_msg:expr) => { 567 unwrap_or!($result, $cont, $err_msg, dir_path) 568 }; 569 ($result:expr, $cont:stmt, $err_msg:expr, $path:expr) => { 570 unwrap_or_warn!( 571 $result, 572 $cont, 573 format!("{}, level: {}", $err_msg, level), 574 $path 575 ) 576 }; 577 } 578 579 // If we fail to list a directory, something bad is happening anyway 580 // (something touches our cache or we have disk failure) 581 // Try to delete it, so we can stay within soft limits of the cache size. 582 // This comment applies later in this function, too. 583 let it = unwrap_or!( 584 fs::read_dir(dir_path), 585 add_unrecognized_and!([dir: dir_path], return), 586 "Failed to list cache directory, deleting it" 587 ); 588 589 let mut cache_files = HashMap::new(); 590 for entry in it { 591 // read_dir() returns an iterator over results - in case some of them are errors 592 // we don't know their names, so we can't delete them. We don't want to delete 593 // the whole directory with good entries too, so we just ignore the erroneous entries. 594 let entry = unwrap_or!( 595 entry, 596 continue, 597 "Failed to read a cache dir entry (NOT deleting it, it still occupies space)" 598 ); 599 let path = entry.path(); 600 match (level, path.is_dir()) { 601 (0..=1, true) => enter_dir(vec, &path, level + 1, cache_config), 602 (0..=1, false) => { 603 if level == 0 604 && path.file_stem() == Some(OsStr::new(".cleanup")) 605 && path.extension().is_some() 606 // assume it's cleanup lock 607 && !is_fs_lock_expired( 608 Some(&entry), 609 &path, 610 cache_config.cleanup_interval(), 611 cache_config.allowed_clock_drift_for_files_from_future(), 612 ) 613 { 614 continue; // skip active lock 615 } 616 add_unrecognized!(file: path); 617 } 618 (2, false) => { 619 match path.extension().and_then(OsStr::to_str) { 620 // mod or stats file 621 None | Some("stats") => { 622 cache_files.insert(path, entry); 623 } 624 625 Some(ext) => { 626 // check if valid lock 627 let recognized = ext.starts_with("wip-") 628 && !is_fs_lock_expired( 629 Some(&entry), 630 &path, 631 cache_config.optimizing_compression_task_timeout(), 632 cache_config.allowed_clock_drift_for_files_from_future(), 633 ); 634 635 if !recognized { 636 add_unrecognized!(file: path); 637 } 638 } 639 } 640 } 641 (_, is_dir) => add_unrecognized!(is_dir, path), 642 } 643 } 644 645 // associate module with its stats & handle them 646 // assumption: just mods and stats 647 for (path, entry) in cache_files.iter() { 648 let path_buf: PathBuf; 649 let (mod_, stats_, is_mod) = match path.extension() { 650 Some(_) => { 651 path_buf = path.with_extension(""); 652 ( 653 cache_files.get(&path_buf).map(|v| (&path_buf, v)), 654 Some((path, entry)), 655 false, 656 ) 657 } 658 None => { 659 path_buf = path.with_extension("stats"); 660 ( 661 Some((path, entry)), 662 cache_files.get(&path_buf).map(|v| (&path_buf, v)), 663 true, 664 ) 665 } 666 }; 667 668 // construct a cache entry 669 match (mod_, stats_, is_mod) { 670 (Some((mod_path, mod_entry)), Some((stats_path, stats_entry)), true) => { 671 let mod_metadata = unwrap_or!( 672 mod_entry.metadata(), 673 add_unrecognized_and!([file: stats_path, file: mod_path], continue), 674 "Failed to get metadata, deleting BOTH module cache and stats files", 675 mod_path 676 ); 677 let stats_mtime = unwrap_or!( 678 stats_entry.metadata().and_then(|m| m.modified()), 679 add_unrecognized_and!( 680 [file: stats_path], 681 unwrap_or!( 682 mod_metadata.modified(), 683 add_unrecognized_and!( 684 [file: stats_path, file: mod_path], 685 continue 686 ), 687 "Failed to get mtime, deleting BOTH module cache and stats \ 688 files", 689 mod_path 690 ) 691 ), 692 "Failed to get metadata/mtime, deleting the file", 693 stats_path 694 ); 695 // .into() called for the SystemTimeStub if cfg(test) 696 #[allow(clippy::identity_conversion)] 697 vec.push(CacheEntry::Recognized { 698 path: mod_path.to_path_buf(), 699 mtime: stats_mtime.into(), 700 size: mod_metadata.len(), 701 }) 702 } 703 (Some(_), Some(_), false) => (), // was or will be handled by previous branch 704 (Some((mod_path, mod_entry)), None, _) => { 705 let (mod_metadata, mod_mtime) = unwrap_or!( 706 mod_entry 707 .metadata() 708 .and_then(|md| md.modified().map(|mt| (md, mt))), 709 add_unrecognized_and!([file: mod_path], continue), 710 "Failed to get metadata/mtime, deleting the file", 711 mod_path 712 ); 713 // .into() called for the SystemTimeStub if cfg(test) 714 #[allow(clippy::identity_conversion)] 715 vec.push(CacheEntry::Recognized { 716 path: mod_path.to_path_buf(), 717 mtime: mod_mtime.into(), 718 size: mod_metadata.len(), 719 }) 720 } 721 (None, Some((stats_path, _stats_entry)), _) => { 722 debug!("Found orphaned stats file: {}", stats_path.display()); 723 add_unrecognized!(file: stats_path); 724 } 725 _ => unreachable!(), 726 } 727 } 728 } 729 730 let mut vec = Vec::new(); 731 enter_dir( 732 &mut vec, 733 self.cache_config.directory(), 734 0, 735 &self.cache_config, 736 ); 737 vec 738 } 739 } 740 741 fn read_stats_file(path: &Path) -> Option<ModuleCacheStatistics> { 742 fs::read(path) 743 .map_err(|err| { 744 trace!( 745 "Failed to read stats file, path: {}, err: {}", 746 path.display(), 747 err 748 ) 749 }) 750 .and_then(|bytes| { 751 toml::from_slice::<ModuleCacheStatistics>(&bytes[..]).map_err(|err| { 752 trace!( 753 "Failed to parse stats file, path: {}, err: {}", 754 path.display(), 755 err, 756 ) 757 }) 758 }) 759 .ok() 760 } 761 762 fn write_stats_file(path: &Path, stats: &ModuleCacheStatistics) -> bool { 763 toml::to_string_pretty(&stats) 764 .map_err(|err| { 765 warn!( 766 "Failed to serialize stats file, path: {}, err: {}", 767 path.display(), 768 err 769 ) 770 }) 771 .and_then(|serialized| { 772 if fs_write_atomic(path, "stats", serialized.as_bytes()) { 773 Ok(()) 774 } else { 775 Err(()) 776 } 777 }) 778 .is_ok() 779 } 780 781 /// Tries to acquire a lock for specific task. 782 /// 783 /// Returns Some(path) to the lock if succeeds. The task path must not 784 /// contain any extension and have file stem. 785 /// 786 /// To release a lock you need either manually rename or remove it, 787 /// or wait until it expires and cleanup task removes it. 788 /// 789 /// Note: this function is racy. Main idea is: be fault tolerant and 790 /// never block some task. The price is that we rarely do some task 791 /// more than once. 792 fn acquire_task_fs_lock( 793 task_path: &Path, 794 timeout: Duration, 795 allowed_future_drift: Duration, 796 ) -> Option<PathBuf> { 797 assert!(task_path.extension().is_none()); 798 assert!(task_path.file_stem().is_some()); 799 800 // list directory 801 let dir_path = task_path.parent()?; 802 let it = fs::read_dir(dir_path) 803 .map_err(|err| { 804 warn!( 805 "Failed to list cache directory, path: {}, err: {}", 806 dir_path.display(), 807 err 808 ) 809 }) 810 .ok()?; 811 812 // look for existing locks 813 for entry in it { 814 let entry = entry 815 .map_err(|err| { 816 warn!( 817 "Failed to list cache directory, path: {}, err: {}", 818 dir_path.display(), 819 err 820 ) 821 }) 822 .ok()?; 823 824 let path = entry.path(); 825 if path.is_dir() || path.file_stem() != task_path.file_stem() { 826 continue; 827 } 828 829 // check extension and mtime 830 match path.extension() { 831 None => continue, 832 Some(ext) => { 833 if let Some(ext_str) = ext.to_str() { 834 // if it's None, i.e. not valid UTF-8 string, then that's not our lock for sure 835 if ext_str.starts_with("wip-") 836 && !is_fs_lock_expired(Some(&entry), &path, timeout, allowed_future_drift) 837 { 838 return None; 839 } 840 } 841 } 842 } 843 } 844 845 // create the lock 846 let lock_path = task_path.with_extension(format!("wip-{}", std::process::id())); 847 let _file = fs::OpenOptions::new() 848 .create_new(true) 849 .write(true) 850 .open(&lock_path) 851 .map_err(|err| { 852 warn!( 853 "Failed to create lock file (note: it shouldn't exists): path: {}, err: {}", 854 lock_path.display(), 855 err 856 ) 857 }) 858 .ok()?; 859 860 Some(lock_path) 861 } 862 863 // we have either both, or just path; dir entry is desirable since on some platforms we can get 864 // metadata without extra syscalls 865 // futhermore: it's better to get a path if we have it instead of allocating a new one from the dir entry 866 fn is_fs_lock_expired( 867 entry: Option<&fs::DirEntry>, 868 path: &PathBuf, 869 threshold: Duration, 870 allowed_future_drift: Duration, 871 ) -> bool { 872 let mtime = match entry 873 .map_or_else(|| path.metadata(), |e| e.metadata()) 874 .and_then(|metadata| metadata.modified()) 875 { 876 Ok(mt) => mt, 877 Err(err) => { 878 warn!( 879 "Failed to get metadata/mtime, treating as an expired lock, path: {}, err: {}", 880 path.display(), 881 err 882 ); 883 return true; // can't read mtime, treat as expired, so this task will not be starved 884 } 885 }; 886 887 // DON'T use: mtime.elapsed() -- we must call SystemTime directly for the tests to be deterministic 888 match SystemTime::now().duration_since(mtime) { 889 Ok(elapsed) => elapsed >= threshold, 890 Err(err) => { 891 trace!( 892 "Found mtime in the future, treating as a not expired lock, path: {}, err: {}", 893 path.display(), 894 err 895 ); 896 // the lock is expired if the time is too far in the future 897 // it is fine to have network share and not synchronized clocks, 898 // but it's not good when user changes time in their system clock 899 err.duration() > allowed_future_drift 900 } 901 } 902 } 903 904 #[cfg(test)] 905 mod tests; 906