Address #656

Broke the reference loop to make sure that the watch_router can
be dropped, and the thread exits.

CHANGELOG.md

@@ -3,6 +3,20 @@ Tantivy 0.11.0
 
 - Added f64 field. Internally reuse u64 code the same way i64 does (@fdb-hiroshima)
 
+
+Tantivy 0.10.2
+=====================
+
+- Closes #656. Solving memory leak.
+
+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.2"
 authors = ["Paul Masurel <paul.masurel@gmail.com>"]
 license = "MIT"
 categories = ["database-implementations", "data-structures"]
@@ -98,4 +98,4 @@ features = ["failpoints"]
 [[test]]
 name = "failpoints"
 path = "tests/failpoints/mod.rs"
-required-features = ["fail/failpoints"]
+required-features = ["fail/failpoints"]

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

@@ -141,42 +141,28 @@ impl MmapCache {
     }
 }
 
-struct InnerWatcherWrapper {
+struct WatcherWrapper {
     _watcher: Mutex<notify::RecommendedWatcher>,
-    watcher_router: WatchCallbackList,
-}
-
-impl InnerWatcherWrapper {
-    pub fn new(path: &Path) -> Result<(Self, Receiver<notify::RawEvent>), notify::Error> {
-        let (tx, watcher_recv): (Sender<RawEvent>, Receiver<RawEvent>) = channel();
-        // We need to initialize the
-        let mut watcher = notify::raw_watcher(tx)?;
-        watcher.watch(path, RecursiveMode::Recursive)?;
-        let inner = InnerWatcherWrapper {
-            _watcher: Mutex::new(watcher),
-            watcher_router: Default::default(),
-        };
-        Ok((inner, watcher_recv))
-    }
-}
-
-#[derive(Clone)]
-pub(crate) struct WatcherWrapper {
-    inner: Arc<InnerWatcherWrapper>,
+    watcher_router: Arc<WatchCallbackList>,
 }
 
 impl WatcherWrapper {
     pub fn new(path: &Path) -> Result<Self, OpenDirectoryError> {
-        let (inner, watcher_recv) = InnerWatcherWrapper::new(path).map_err(|err| match err {
-            notify::Error::PathNotFound => OpenDirectoryError::DoesNotExist(path.to_owned()),
-            _ => {
-                panic!("Unknown error while starting watching directory {:?}", path);
-            }
-        })?;
-        let watcher_wrapper = WatcherWrapper {
-            inner: Arc::new(inner),
-        };
-        let watcher_wrapper_clone = watcher_wrapper.clone();
+        let (tx, watcher_recv): (Sender<RawEvent>, Receiver<RawEvent>) = channel();
+        // We need to initialize the
+        let watcher = notify::raw_watcher(tx)
+            .and_then(|mut watcher| {
+                watcher.watch(path, RecursiveMode::Recursive)?;
+                Ok(watcher)
+            })
+            .map_err(|err| match err {
+                notify::Error::PathNotFound => OpenDirectoryError::DoesNotExist(path.to_owned()),
+                _ => {
+                    panic!("Unknown error while starting watching directory {:?}", path);
+                }
+            })?;
+        let watcher_router: Arc<WatchCallbackList> = Default::default();
+        let watcher_router_clone = watcher_router.clone();
         thread::Builder::new()
             .name("meta-file-watch-thread".to_string())
             .spawn(move || {
@@ -187,7 +173,7 @@ impl WatcherWrapper {
                             // We might want to be more accurate than this at one point.
                             if let Some(filename) = changed_path.file_name() {
                                 if filename == *META_FILEPATH {
-                                    watcher_wrapper_clone.inner.watcher_router.broadcast();
+                                    watcher_router_clone.broadcast();
                                 }
                             }
                         }
@@ -200,13 +186,15 @@ impl WatcherWrapper {
                         }
                     }
                 }
-            })
-            .expect("Failed to spawn thread to watch meta.json");
-        Ok(watcher_wrapper)
+            })?;
+        Ok(WatcherWrapper {
+            _watcher: Mutex::new(watcher),
+            watcher_router,
+        })
     }
 
     pub fn watch(&mut self, watch_callback: WatchCallback) -> WatchHandle {
-        self.inner.watcher_router.subscribe(watch_callback)
+        self.watcher_router.subscribe(watch_callback)
     }
 }
 
@@ -231,7 +219,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 +227,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 +519,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);
             }
         }