Closes #544. (#607)

Prepare for release 0.10.1

CHANGELOG.md

@@ -3,6 +3,14 @@ Tantivy 0.11.0
 
 - Added f64 field. Internally reuse u64 code the same way i64 does (@fdb-hiroshima)
 
+Tantivy 0.10.1
+=====================
+
+- Closes #544.  A few users experienced problems with the directory watching system.
+Avoid watching the mmap directory until someone effectively creates a reader that uses
+this functionality.
+
+
 Tantivy 0.10.0
 =====================
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tantivy"
-version = "0.10.0"
+version = "0.10.1"
 authors = ["Paul Masurel <paul.masurel@gmail.com>"]
 license = "MIT"
 categories = ["database-implementations", "data-structures"]

src/directory/directory.rs

@@ -204,7 +204,7 @@ pub trait Directory: DirectoryClone + fmt::Debug + Send + Sync + 'static {
     /// Internally, tantivy only uses this API to detect new commits to implement the
     /// `OnCommit` `ReloadPolicy`. Not implementing watch in a `Directory` only prevents the
     /// `OnCommit` `ReloadPolicy` to work properly.
-    fn watch(&self, watch_callback: WatchCallback) -> WatchHandle;
+    fn watch(&self, watch_callback: WatchCallback) -> crate::Result<WatchHandle>;
 }
 
 /// DirectoryClone

src/directory/managed_directory.rs

@@ -241,7 +241,7 @@ impl Directory for ManagedDirectory {
         self.directory.acquire_lock(lock)
     }
 
-    fn watch(&self, watch_callback: WatchCallback) -> WatchHandle {
+    fn watch(&self, watch_callback: WatchCallback) -> crate::Result<WatchHandle> {
         self.directory.watch(watch_callback)
     }
 }

src/directory/mmap_directory.rs

@@ -161,7 +161,7 @@ impl InnerWatcherWrapper {
 }
 
 #[derive(Clone)]
-pub(crate) struct WatcherWrapper {
+struct WatcherWrapper {
     inner: Arc<InnerWatcherWrapper>,
 }
 
@@ -231,7 +231,7 @@ struct MmapDirectoryInner {
     root_path: PathBuf,
     mmap_cache: RwLock<MmapCache>,
     _temp_directory: Option<TempDir>,
-    watcher: RwLock<WatcherWrapper>,
+    watcher: RwLock<Option<WatcherWrapper>>,
 }
 
 impl MmapDirectoryInner {
@@ -239,19 +239,36 @@ impl MmapDirectoryInner {
         root_path: PathBuf,
         temp_directory: Option<TempDir>,
     ) -> Result<MmapDirectoryInner, OpenDirectoryError> {
-        let watch_wrapper = WatcherWrapper::new(&root_path)?;
         let mmap_directory_inner = MmapDirectoryInner {
             root_path,
             mmap_cache: Default::default(),
             _temp_directory: temp_directory,
-            watcher: RwLock::new(watch_wrapper),
+            watcher: RwLock::new(None),
         };
         Ok(mmap_directory_inner)
     }
 
-    fn watch(&self, watch_callback: WatchCallback) -> WatchHandle {
-        let mut wlock = self.watcher.write().unwrap();
-        wlock.watch(watch_callback)
+    fn watch(&self, watch_callback: WatchCallback) -> crate::Result<WatchHandle> {
+        // a lot of juggling here, to ensure we don't do anything that panics
+        // while the rwlock is held. That way we ensure that the rwlock cannot
+        // be poisoned.
+        //
+        // The downside is that we might create a watch wrapper that is not useful.
+        let need_initialization = self.watcher.read().unwrap().is_none();
+        if need_initialization {
+            let watch_wrapper = WatcherWrapper::new(&self.root_path)?;
+            let mut watch_wlock = self.watcher.write().unwrap();
+            // the watcher could have been initialized when we released the lock, and
+            // we do not want to lose the watched files that were set.
+            if watch_wlock.is_none() {
+                *watch_wlock = Some(watch_wrapper);
+            }
+        }
+        if let Some(watch_wrapper) = self.watcher.write().unwrap().as_mut() {
+            return Ok(watch_wrapper.watch(watch_callback));
+        } else {
+            unreachable!("At this point, watch wrapper is supposed to be initialized");
+        }
     }
 }
 
@@ -514,7 +531,7 @@ impl Directory for MmapDirectory {
         })))
     }
 
-    fn watch(&self, watch_callback: WatchCallback) -> WatchHandle {
+    fn watch(&self, watch_callback: WatchCallback) -> crate::Result<WatchHandle> {
         self.inner.watch(watch_callback)
     }
 }

src/directory/ram_directory.rs

@@ -193,7 +193,7 @@ impl Directory for RAMDirectory {
         Ok(())
     }
 
-    fn watch(&self, watch_callback: WatchCallback) -> WatchHandle {
-        self.fs.write().unwrap().watch(watch_callback)
+    fn watch(&self, watch_callback: WatchCallback) -> crate::Result<WatchHandle> {
+        Ok(self.fs.write().unwrap().watch(watch_callback))
     }
 }

src/directory/tests.rs

@@ -121,7 +121,7 @@ fn test_watch(directory: &mut dyn Directory) {
     thread::sleep(Duration::new(0, 10_000));
     assert_eq!(0, counter.load(Ordering::SeqCst));
 
-    let watch_handle = directory.watch(watch_callback);
+    let watch_handle = directory.watch(watch_callback).unwrap();
     for i in 0..10 {
         assert_eq!(i, counter.load(Ordering::SeqCst));
         assert!(directory

src/reader/mod.rs

@@ -85,7 +85,10 @@ impl IndexReaderBuilder {
                         );
                     }
                 };
-                let watch_handle = inner_reader_arc.index.directory().watch(Box::new(callback));
+                let watch_handle = inner_reader_arc
+                    .index
+                    .directory()
+                    .watch(Box::new(callback))?;
                 watch_handle_opt = Some(watch_handle);
             }
         }