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