From 5baf0ccb0e2eb0c18745f2aed17445a9d6f19f71 Mon Sep 17 00:00:00 2001 From: Daniel Faust Date: Sat, 9 Jul 2016 17:39:25 +0200 Subject: [PATCH] PollWatcher: Immediately scan path when adding a watch --- src/poll.rs | 257 +++++++++++++++++++++++++++------------------------- 1 file changed, 132 insertions(+), 125 deletions(-) diff --git a/src/poll.rs b/src/poll.rs index eaa44b1..81c6179 100644 --- a/src/poll.rs +++ b/src/poll.rs @@ -4,7 +4,7 @@ //! cross-platform APIs; it should function on any platform that the Rust standard library does. use std::collections::HashMap; -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, RwLock, Mutex}; use std::sync::mpsc::Sender; use std::fs; use std::thread; @@ -18,10 +18,20 @@ use filetime::FileTime; extern crate walkdir; extern crate time; +struct PathData { + mtime: u64, + last_check: f64, +} + +struct WatchData { + is_recursive: bool, + paths: HashMap, +} + /// Polling based `Watcher` implementation pub struct PollWatcher { tx: Sender, - watches: Arc>>, + watches: Arc>>, open: Arc>, } @@ -30,7 +40,7 @@ impl PollWatcher { pub fn with_delay(tx: Sender, delay: u32) -> Result { let mut p = PollWatcher { tx: tx, - watches: Arc::new(RwLock::new(HashMap::new())), + watches: Arc::new(Mutex::new(HashMap::new())), open: Arc::new(RwLock::new(true)), }; p.run(delay); @@ -46,122 +56,77 @@ impl PollWatcher { // TODO: handle chmod events // TODO: handle renames // TODO: DRY it up - let mut mtimes: HashMap = HashMap::new(); - - let mut current_time = time::precise_time_s(); - - for (watch, is_recursive) in watches.read().unwrap().iter() { - match fs::metadata(watch) { - Err(e) => { - let _ = tx.send(Event { - path: Some(watch.clone()), - op: Err(Error::Io(e)), - }); - continue; - } - Ok(metadata) => { - if !metadata.is_dir() { - let modified = FileTime::from_last_modification_time(&metadata).seconds(); - mtimes.insert(watch.clone(), (modified, current_time)); - } else { - let depth = if *is_recursive { - usize::max_value() - } else { - 1 - }; - for entry in WalkDir::new(watch) - .follow_links(true) - .max_depth(depth) - .into_iter() - .filter_map(|e| e.ok()) { - let path = entry.path(); - - match entry.metadata() { - Err(e) => { - let _ = tx.send(Event { - path: Some(path.to_path_buf()), - op: Err(Error::Io(e.into())), - }); - } - Ok(m) => { - let modified = FileTime::from_last_modification_time(&m).seconds(); - mtimes.insert(path.to_path_buf(), (modified, current_time)); - } - } - } - } - } - } - } loop { if !(*open.read().unwrap()) { break; } - current_time = time::precise_time_s(); + if let Ok(mut watches) = watches.lock() { + let current_time = time::precise_time_s(); - for (watch, is_recursive) in watches.read().unwrap().iter() { - match fs::metadata(watch) { - Err(e) => { - let _ = tx.send(Event { - path: Some(watch.clone()), - op: Err(Error::Io(e)), - }); - continue; - } - Ok(metadata) => { - if !metadata.is_dir() { - let modified = FileTime::from_last_modification_time(&metadata).seconds(); - match mtimes.insert(watch.clone(), (modified, current_time)) { - None => { - unreachable!(); - } - Some((old_modified, _)) => { - if modified > old_modified { - let _ = tx.send(Event { - path: Some(watch.clone()), - op: Ok(op::WRITE), - }); + for (watch, &mut WatchData{is_recursive, ref mut paths}) in watches.iter_mut() { + match fs::metadata(watch) { + Err(e) => { + let _ = tx.send(Event { + path: Some(watch.clone()), + op: Err(Error::Io(e)), + }); + continue; + } + Ok(metadata) => { + if !metadata.is_dir() { + let mtime = FileTime::from_last_modification_time(&metadata).seconds(); + match paths.insert(watch.clone(), PathData{mtime: mtime, last_check: current_time}) { + None => { + unreachable!(); + } + Some(PathData{mtime: old_mtime, ..}) => { + if mtime > old_mtime { + let _ = tx.send(Event { + path: Some(watch.clone()), + op: Ok(op::WRITE), + }); + } } } - } - } else { - let depth = if *is_recursive { - usize::max_value() } else { - 1 - }; - for entry in WalkDir::new(watch) - .follow_links(true) - .max_depth(depth) - .into_iter() - .filter_map(|e| e.ok()) { - let path = entry.path(); + let depth = if is_recursive { + usize::max_value() + } else { + 1 + }; + for entry in WalkDir::new(watch) + .follow_links(true) + .max_depth(depth) + .into_iter() + .filter_map(|e| e.ok()) { + let path = entry.path(); - match entry.metadata() { - Err(e) => { - let _ = tx.send(Event { - path: Some(path.to_path_buf()), - op: Err(Error::Io(e.into())), - }); - } - Ok(m) => { - let modified = FileTime::from_last_modification_time(&m).seconds(); - match mtimes.insert(path.to_path_buf(), (modified, current_time)) { - None => { - let _ = tx.send(Event { - path: Some(path.to_path_buf()), - op: Ok(op::CREATE), - }); - } - Some((old_modified, _)) => { - if modified > old_modified { + match entry.metadata() { + Err(e) => { + let _ = tx.send(Event { + path: Some(path.to_path_buf()), + op: Err(Error::Io(e.into())), + }); + } + Ok(m) => { + let mtime = FileTime::from_last_modification_time(&m).seconds(); + match paths.insert(path.to_path_buf(), PathData{mtime: mtime, last_check: current_time}) { + None => { let _ = tx.send(Event { path: Some(path.to_path_buf()), - op: Ok(op::WRITE), + op: Ok(op::CREATE), }); } + Some(PathData{mtime: old_mtime, ..}) => { + if mtime > old_mtime { + let _ = tx.send(Event { + path: Some(path.to_path_buf()), + op: Ok(op::WRITE), + }); + } + } } } } @@ -170,33 +135,26 @@ impl PollWatcher { } } } - } - let mut removed: Vec = Vec::new(); - - 'paths: for (ref path, &(_, last_checked)) in &mtimes { - for (watch, _) in watches.read().unwrap().iter() { - if path.starts_with(watch) { - if last_checked < current_time { + for (_, &mut WatchData{ref mut paths, ..}) in watches.iter_mut() { + let mut removed = Vec::new(); + for (path, &PathData{last_check, ..}) in paths.iter() { + if last_check < current_time { let _ = tx.send(Event { - path: Some(path.to_path_buf()), + path: Some(path.clone()), op: Ok(op::REMOVE), }); - removed.push(path.to_path_buf()); + removed.push(path.clone()); } - continue 'paths; + } + for path in removed { + (*paths).remove(&path); } } - // not found in watches - removed.push(path.to_path_buf()); - } - for path in removed { - mtimes.remove(&path); - } - - if delay != 0 { - thread::sleep(Duration::from_millis(delay as u64)); + if delay != 0 { + thread::sleep(Duration::from_millis(delay as u64)); + } } } }); @@ -209,12 +167,61 @@ impl Watcher for PollWatcher { } fn watch>(&mut self, path: P, recursive_mode: RecursiveMode) -> Result<()> { - (*self.watches).write().unwrap().insert(path.as_ref().to_path_buf(), recursive_mode.is_recursive()); + if let Ok(mut watches) = self.watches.lock() { + let current_time = time::precise_time_s(); + + let watch = path.as_ref().to_owned(); + + match fs::metadata(path) { + Err(e) => { + let _ = self.tx.send(Event { + path: Some(watch.clone()), + op: Err(Error::Io(e)), + }); + } + Ok(metadata) => { + if !metadata.is_dir() { + let mut paths = HashMap::new(); + let mtime = FileTime::from_last_modification_time(&metadata).seconds(); + paths.insert(watch.clone(), PathData{mtime: mtime, last_check: current_time}); + watches.insert(watch, WatchData{is_recursive: recursive_mode.is_recursive(), paths: paths}); + } else { + let mut paths = HashMap::new(); + let depth = if recursive_mode.is_recursive() { + usize::max_value() + } else { + 1 + }; + for entry in WalkDir::new(watch.clone()) + .follow_links(true) + .max_depth(depth) + .into_iter() + .filter_map(|e| e.ok()) { + let path = entry.path(); + + match entry.metadata() { + Err(e) => { + let _ = self.tx.send(Event { + path: Some(path.to_path_buf()), + op: Err(Error::Io(e.into())), + }); + } + Ok(m) => { + let mtime = FileTime::from_last_modification_time(&m).seconds(); + paths.insert(path.to_path_buf(), PathData{mtime: mtime, last_check: current_time}); + } + } + } + watches.insert(watch, WatchData{is_recursive: recursive_mode.is_recursive(), paths: paths}); + } + } + } + } Ok(()) } fn unwatch>(&mut self, path: P) -> Result<()> { - if (*self.watches).write().unwrap().remove(path.as_ref()).is_some() { + if (*self.watches).lock().unwrap().remove(path.as_ref()).is_some() { Ok(()) } else { Err(Error::WatchNotFound)