Compare commits

...

1 Commits

Author SHA1 Message Date
Dmitriy Kovalenko 258afa374f fix: Nested background threads leaking 2025-08-04 16:46:53 +02:00
7 changed files with 160 additions and 116 deletions
+1
View File
@@ -27,6 +27,7 @@ M.get_scan_progress = rust_module.get_scan_progress
M.is_scanning = rust_module.is_scanning
M.refresh_git_status = rust_module.refresh_git_status
M.stop_background_monitor = rust_module.stop_background_monitor
M.cleanup_file_picker = rust_module.cleanup_file_picker
M.init_tracing = rust_module.init_tracing
M.wait_for_initial_scan = rust_module.wait_for_initial_scan
+27 -20
View File
@@ -107,8 +107,7 @@ function M.setup(config)
M.config = merged_config
M.setup_commands()
if merged_config.frecency.enabled then M.setup_global_file_tracking() end
M.setup_global_autocmds()
local git_utils = require('fff.git_utils')
git_utils.setup_highlights()
@@ -126,27 +125,29 @@ function M.setup(config)
return true
end
function M.setup_global_file_tracking()
function M.setup_global_autocmds()
local group = vim.api.nvim_create_augroup('fff_file_tracking', { clear = true })
vim.api.nvim_create_autocmd({ 'BufRead', 'BufNewFile' }, {
group = group,
callback = function(args)
local file_path = args.file
if M.config.frecency.enabled then
vim.api.nvim_create_autocmd({ 'BufRead', 'BufNewFile' }, {
group = group,
callback = function(args)
local file_path = args.file
if file_path and file_path ~= '' and not vim.startswith(file_path, 'term://') then
-- never block the UI
vim.schedule(function()
local stat = vim.uv.fs_stat(file_path)
if stat and stat.type == 'file' then
local relative_path = vim.fn.fnamemodify(file_path, ':.')
pcall(fuzzy.access_file, relative_path)
end
end)
end
end,
desc = 'Track file access for FFF frecency',
})
if file_path and file_path ~= '' and not vim.startswith(file_path, 'term://') then
-- never block the UI
vim.schedule(function()
local stat = vim.uv.fs_stat(file_path)
if stat and stat.type == 'file' then
local relative_path = vim.fn.fnamemodify(file_path, ':.')
pcall(fuzzy.access_file, relative_path)
end
end)
end
end,
desc = 'Track file access for FFF frecency',
})
end
-- make sure that this won't work correctly if autochdir plugins are enabled
-- using a pure :cd command but will work using lua api or :e command
@@ -167,6 +168,12 @@ function M.setup_global_file_tracking()
end,
desc = 'Automatically sync FFF directory changes',
})
vim.api.nvim_create_autocmd('VimLeavePre', {
group = group,
callback = function() pcall(fuzzy.cleanup_file_picker) end,
desc = 'Cleanup FFF background threads on Neovim exit',
})
end
function M.setup_commands()
+13 -18
View File
@@ -1,22 +1,18 @@
#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum Error {
#[error("Thread panicked")]
ThreadPanic,
#[error("Invalid path {0}")]
InvalidPath(std::path::PathBuf),
#[error("File picker not initialized")]
FilePickerMissing,
#[error("Failed to acquire lock for frecency")]
AcquireFrecencyLock,
#[error("Failed to acquire lock for items by provider")]
AcquireItemLock,
#[error("Attempted to use frecency before initialization")]
UseFrecencyBeforeInit,
#[error(
"Attempted to fuzzy match for provider {provider_id} before setting the provider's items"
)]
FuzzyBeforeSetItems { provider_id: String },
#[error("Failed to create frecency database directory: {0}")]
CreateDir(#[source] std::io::Error),
CreateDir(#[from] std::io::Error),
#[error("Failed to open frecency database env: {0}")]
EnvOpen(#[source] heed::Error),
#[error("Failed to create frecency database: {0}")]
@@ -35,16 +31,15 @@ pub enum Error {
DbWrite(#[source] heed::Error),
#[error("Failed to commit write transaction to frecency database: {0}")]
DbCommit(#[source] heed::Error),
#[error("Invalid file path: {0}")]
InvalidPath(String),
#[error("Failed to scan directory: {0}")]
DirectoryScan(String),
#[error("Failed to start file system watcher: {0}")]
FileSystemWatch(#[from] notify::Error),
}
impl From<Error> for mlua::Error {
fn from(value: Error) -> Self {
mlua::Error::RuntimeError(value.to_string())
let string_value = value.to_string();
::tracing::error!(string_value);
mlua::Error::RuntimeError(string_value)
}
}
+88 -59
View File
@@ -11,7 +11,7 @@ use rayon::prelude::*;
use std::path::{Path, PathBuf};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc, RwLock,
Arc, Mutex, RwLock,
};
use std::thread;
use std::time::{Duration, SystemTime};
@@ -27,6 +27,11 @@ struct FileSync {
scan_generation: u64,
}
type Debouncer = notify_debouncer_full::Debouncer<
notify::RecommendedWatcher,
notify_debouncer_full::RecommendedCache,
>;
impl FileSync {
fn new() -> Self {
Self {
@@ -170,9 +175,8 @@ pub struct FilePicker {
base_path: PathBuf,
git_workdir: Option<PathBuf>,
sync_data: Arc<RwLock<FileSync>>,
shutdown_signal: Arc<AtomicBool>,
is_scanning: Arc<AtomicBool>,
_background_handle: Option<thread::JoinHandle<()>>,
_debouncer: Arc<Mutex<Option<Debouncer>>>,
}
impl std::fmt::Debug for FilePicker {
@@ -190,7 +194,7 @@ impl FilePicker {
let path = PathBuf::from(&base_path);
if !path.exists() {
error!("Base path does not exist: {}", base_path);
return Err(Error::InvalidPath(path.to_string_lossy().into_owned()));
return Err(Error::InvalidPath(path));
}
let git_workdir = Repository::discover(&path)
@@ -204,25 +208,20 @@ impl FilePicker {
}
let sync_data = Arc::new(RwLock::new(FileSync::new()));
let shutdown = Arc::new(AtomicBool::new(false));
let scan_signal = Arc::new(AtomicBool::new(false));
let debouncer_holder = Arc::new(Mutex::new(None));
let background_handle = spawn_background_watcher(
path.clone(),
git_workdir.clone(),
Arc::clone(&sync_data),
Arc::clone(&shutdown),
Arc::clone(&scan_signal),
);
let picker = Self {
base_path: path.clone(),
git_workdir: git_workdir.clone(),
sync_data: Arc::clone(&sync_data),
is_scanning: Arc::clone(&scan_signal),
_debouncer: Arc::clone(&debouncer_holder),
};
Ok(Self {
base_path: path,
git_workdir,
sync_data,
shutdown_signal: shutdown,
is_scanning: scan_signal,
_background_handle: Some(background_handle),
})
spawn_async_initialization(path, git_workdir, sync_data, scan_signal, debouncer_holder);
Ok(picker)
}
pub fn fuzzy_search(
@@ -322,6 +321,16 @@ impl FilePicker {
self.get_cached_files()
}
pub fn stop_background_monitor(&self) -> Result<(), Error> {
if let Ok(mut debouncer_guard) = self._debouncer.lock() {
if let Some(debouncer) = debouncer_guard.take() {
debouncer.stop_nonblocking();
info!("File watcher stopped successfully");
}
}
Ok(())
}
pub fn trigger_rescan(&self) -> Result<(), Error> {
if self.is_scanning.load(Ordering::Relaxed) {
debug!("Scan already in progress, skipping trigger_rescan");
@@ -358,10 +367,6 @@ impl FilePicker {
pub fn is_scan_active(&self) -> bool {
self.is_scanning.load(Ordering::Relaxed)
}
pub fn stop_background_monitor(&self) {
self.shutdown_signal.store(true, Ordering::Relaxed);
}
}
#[allow(unused)]
@@ -372,21 +377,21 @@ pub struct ScanProgress {
pub is_scanning: bool,
}
fn spawn_background_watcher(
fn spawn_async_initialization(
base_path: PathBuf,
git_workdir: Option<PathBuf>,
sync_data: Arc<RwLock<FileSync>>,
shutdown: Arc<AtomicBool>,
scan_signal: Arc<AtomicBool>,
) -> thread::JoinHandle<()> {
debouncer_holder: Arc<Mutex<Option<Debouncer>>>,
) {
thread::spawn(move || {
scan_signal.store(true, Ordering::Relaxed);
info!("starting background watcher thread");
info!("Starting async initialization for file picker");
match scan_filesystem(&base_path, git_workdir.as_ref()) {
Ok((files, git_cache)) => {
info!(
"Initial parallel filesystem scan completed: found {} files",
"Initial filesystem scan completed: found {} files",
files.len()
);
if let Ok(mut data) = sync_data.write() {
@@ -395,43 +400,57 @@ fn spawn_background_watcher(
}
}
Err(e) => {
error!("Failed to scan filesystem: {:?}", e);
error!("Initial scan failed: {:?}", e);
}
}
scan_signal.store(false, Ordering::Relaxed);
error!("is_scanning = FALSE (initial scan completed)");
let mut debouncer = match new_debouncer(Duration::from_millis(500), None, {
let sync_data = Arc::clone(&sync_data);
let base_path = base_path.clone();
let git_workdir = git_workdir.clone();
move |result: DebounceEventResult| match result {
Ok(events) => {
handle_debounced_events(events, &sync_data, &base_path, &git_workdir);
}
Err(errors) => {
error!("File watcher errors: {:?}", errors);
match create_file_watcher_sync(base_path, git_workdir, Arc::clone(&sync_data)) {
Ok(debouncer) => {
if let Ok(mut holder) = debouncer_holder.lock() {
*holder = Some(debouncer);
info!("File watcher setup completed successfully");
} else {
error!("Failed to store debouncer - mutex poisoned");
}
}
}) {
Ok(debouncer) => debouncer,
Err(e) => {
error!("Failed to create debouncer: {:?}", e);
return;
error!("Failed to create file watcher: {:?}", e);
}
};
if let Err(e) = debouncer.watch(&base_path, RecursiveMode::Recursive) {
error!("Failed to start watching: {:?}", e);
return;
}
});
}
while !shutdown.load(Ordering::Relaxed) {
thread::sleep(Duration::from_millis(100));
fn create_file_watcher_sync(
base_path: PathBuf,
git_workdir: Option<PathBuf>,
sync_data: Arc<RwLock<FileSync>>,
) -> Result<Debouncer, Error> {
let mut debouncer = new_debouncer(Duration::from_millis(500), None, {
let sync_data = Arc::clone(&sync_data);
let base_path = base_path.clone();
let git_workdir = git_workdir.clone();
move |result: DebounceEventResult| match result {
Ok(events) => {
handle_debounced_events(events, &sync_data, &base_path, &git_workdir);
}
Err(errors) => {
error!("File watcher errors: {:?}", errors);
}
}
})
})?;
if let Err(e) = debouncer.watch(&base_path, RecursiveMode::Recursive) {
error!(
"Failed to start watching path: {}, error {e:?}",
base_path.display(),
);
return Err(e.into());
}
info!("File watcher started for path: {}", base_path.display());
Ok(debouncer)
}
fn handle_debounced_events(
@@ -616,9 +635,10 @@ fn scan_filesystem(
let walker_time = walker_start.elapsed();
info!("SCAN: File walking completed in {:?}", walker_time);
let git_cache = git_handle
.join()
.map_err(|_| Error::InvalidPath("Git status thread panicked".to_string()))?;
let git_cache = git_handle.join().map_err(|_| {
error!("Failed to join git status thread");
Error::ThreadPanic
})?;
if let Some(git_cache) = &git_cache {
files.par_iter_mut().for_each(|file| {
@@ -698,6 +718,15 @@ fn is_git_file(path: &Path) -> bool {
impl Drop for FilePicker {
fn drop(&mut self) {
self.shutdown_signal.store(true, Ordering::Relaxed);
info!("FilePicker is being dropped, stopping file watcher");
if let Ok(mut debouncer_guard) = self._debouncer.lock() {
if let Some(debouncer) = debouncer_guard.take() {
debouncer.stop();
info!("File watcher stopped successfully");
}
} else {
error!("Failed to acquire debouncer lock during drop");
}
}
}
+29 -12
View File
@@ -51,7 +51,7 @@ fn reinit_file_picker_internal(path: std::path::PathBuf) -> Result<(), Error> {
// drop should clean it anyway but just to be extra sure
if let Some(picker) = file_picker.take() {
picker.stop_background_monitor();
picker.stop_background_monitor()?;
}
let new_picker = FilePicker::new(path.to_string_lossy().to_string())?;
@@ -81,7 +81,7 @@ pub fn scan_files(_: &Lua, _: ()) -> LuaResult<()> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
picker.trigger_rescan()?;
::tracing::info!("scan_files trigger_rescan completed");
@@ -92,7 +92,7 @@ pub fn get_cached_files(_: &Lua, _: ()) -> LuaResult<Vec<FileItem>> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
Ok(picker.get_cached_files())
}
@@ -105,7 +105,7 @@ pub fn fuzzy_search_files(
::tracing::debug!("Fuzzy search started: {:?}", time.elapsed());
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
let results = picker.fuzzy_search(&query, max_results, max_threads, current_file.as_ref());
Ok(results)
@@ -124,7 +124,7 @@ pub fn get_scan_progress(lua: &Lua, _: ()) -> LuaResult<LuaValue> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
let progress = picker.get_scan_progress();
let table = lua.create_table()?;
@@ -138,7 +138,7 @@ pub fn is_scanning(_: &Lua, _: ()) -> LuaResult<bool> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
Ok(picker.is_scan_active())
}
@@ -146,20 +146,33 @@ pub fn refresh_git_status(_: &Lua, _: ()) -> LuaResult<Vec<FileItem>> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
Ok(picker.refresh_git_status())
}
pub fn stop_background_monitor(_: &Lua, _: ()) -> LuaResult<bool> {
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let mut file_picker = FILE_PICKER.write().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
picker.stop_background_monitor();
.as_mut()
.ok_or_else(|| Error::FilePickerMissing)?;
picker.stop_background_monitor()?;
Ok(true)
}
pub fn cleanup_file_picker(_: &Lua, _: ()) -> LuaResult<bool> {
let mut file_picker = FILE_PICKER.write().map_err(|_| Error::AcquireItemLock)?;
if let Some(picker) = file_picker.take() {
drop(picker);
::tracing::info!("FilePicker cleanup completed");
Ok(true)
} else {
Ok(false)
}
}
pub fn cancel_scan(_: &Lua, _: ()) -> LuaResult<bool> {
Ok(true)
}
@@ -168,7 +181,7 @@ pub fn wait_for_initial_scan(_: &Lua, timeout_ms: Option<u64>) -> LuaResult<bool
let file_picker = FILE_PICKER.read().map_err(|_| Error::AcquireItemLock)?;
let picker = file_picker
.as_ref()
.ok_or_else(|| Error::InvalidPath("File picker not initialized".to_string()))?;
.ok_or_else(|| Error::FilePickerMissing)?;
let timeout = Duration::from_millis(timeout_ms.unwrap_or(5000)); // Default 5s timeout
let start_time = std::time::Instant::now();
@@ -220,6 +233,10 @@ fn create_exports(lua: &Lua) -> LuaResult<LuaTable> {
"wait_for_initial_scan",
lua.create_function(wait_for_initial_scan)?,
)?;
exports.set(
"cleanup_file_picker",
lua.create_function(cleanup_file_picker)?,
)?;
Ok(exports)
}
+1 -7
View File
@@ -17,13 +17,7 @@ pub fn init_tracing(log_file_path: &str, log_level: &str) -> Result<String, Erro
let log_path = Path::new(log_file_path);
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent).map_err(|e| {
Error::InvalidPath(format!(
"Failed to create log directory {}: {}",
parent.display(),
e
))
})?;
std::fs::create_dir_all(parent)?;
}
let log_file_path_clone = log_file_path.to_string();
+1
View File
@@ -1,5 +1,6 @@
#![allow(clippy::all)]
#![allow(dead_code)]
#![allow(clippy::enum_variant_names)]
#[path = "../../lua/fff/rust/error.rs"]
mod error;