From e428b9a16b033fb7ca65020a489f2d05231d8ddf Mon Sep 17 00:00:00 2001 From: Dmitriy Kovalenko Date: Fri, 6 Nov 2026 19:59:29 +0000 Subject: [PATCH] Reduce contention in thread writer lock coordination (#59969) ## Why Exclusive coordination serialized writer acquisition across unrelated threads and could delay releasing writer ownership while waiting to remove a lock file. ## What changed Use shared coordination when opening thread writer locks. Attempt the initial stale-lock sweep only when exclusive coordination is immediately available, and keep rollout publication exclusively coordinated. When dropping a writer guard, remove its lock file only if exclusive coordination can be acquired immediately. Otherwise, release ownership and leave the file available for reuse or later cleanup. ## Testing Extend the writer-lock test to cover acquisition for different threads under shared coordination, rejection of competing owners for the same thread, and prompt ownership release and reacquisition when cleanup cannot acquire exclusive coordination. GitOrigin-RevId: 3e187b32dbef3bcfdd8c62e91eca58b615f3879e --- codex-rs/rollout/src/writer_lock.rs | 57 +++++++++++++++++------ codex-rs/rollout/src/writer_lock_tests.rs | 34 ++++++++++++-- 2 files changed, 73 insertions(+), 18 deletions(-) diff --git a/codex-rs/rollout/src/writer_lock.rs b/codex-rs/rollout/src/writer_lock.rs index 31068707a2..7ad0f8e619 100644 --- a/codex-rs/rollout/src/writer_lock.rs +++ b/codex-rs/rollout/src/writer_lock.rs @@ -1,5 +1,6 @@ //! Shared cross-process ownership for local thread writers and rollout publication. -//! Thread lock-file creation and removal uses a bounded wait for the home coordination lock. +//! Opening a thread lock file uses a bounded wait for home coordination. Cleanup +//! removes the file only if it can acquire exclusive coordination immediately. use std::fs; use std::fs::File; @@ -43,12 +44,22 @@ impl WriterLockCoordinator { /// Acquires exclusive writer ownership, returning `WouldBlock` for an active writer. pub fn acquire(self: &Arc, thread_id: ThreadId) -> io::Result { - let _coordination_lock = self.lock_coordination()?; - if !self.cleanup_attempted.swap(true, Ordering::Relaxed) - && let Err(err) = self.remove_stale_thread_locks() - { - warn!("failed to clean up stale thread writer locks: {err}"); + let file = self.open_coordination()?; + // Never queue a directory sweep behind another writer. Removal remains exclusive. + let have_exclusive = !self.cleanup_attempted.swap(true, Ordering::Relaxed) + && match file.try_lock() { + Ok(()) => true, + Err(fs::TryLockError::WouldBlock) => false, + Err(fs::TryLockError::Error(err)) => return Err(err), + }; + if have_exclusive { + if let Err(err) = self.remove_stale_thread_locks() { + warn!("failed to clean up stale thread writer locks: {err}"); + } + } else { + Self::wait_for_coordination(|| file.try_lock_shared())?; } + let _coordination_lock = file; let path = self.directory.join(format!("{thread_id}.lock")); let file = OpenOptions::new() @@ -107,18 +118,29 @@ impl WriterLockCoordinator { } } - fn lock_coordination(&self) -> io::Result { + fn open_coordination(&self) -> io::Result { fs::create_dir_all(&self.directory)?; - let file = OpenOptions::new() + OpenOptions::new() .read(true) .write(true) .create(true) .truncate(false) - .open(self.directory.join(COORDINATION_LOCK_FILE))?; + .open(self.directory.join(COORDINATION_LOCK_FILE)) + } + + fn lock_coordination(&self) -> io::Result { + let file = self.open_coordination()?; + Self::wait_for_coordination(|| file.try_lock())?; + Ok(file) + } + + fn wait_for_coordination( + try_lock: impl Fn() -> Result<(), fs::TryLockError>, + ) -> io::Result<()> { let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30); loop { - match file.try_lock() { - Ok(()) => break, + match try_lock() { + Ok(()) => return Ok(()), Err(fs::TryLockError::WouldBlock) if Instant::now() < deadline => { std::thread::sleep(Duration::from_millis(/*millis*/ 50)); } @@ -131,7 +153,6 @@ impl WriterLockCoordinator { Err(fs::TryLockError::Error(err)) => return Err(err), } } - Ok(file) } fn remove_stale_thread_locks(&self) -> io::Result<()> { @@ -186,7 +207,7 @@ impl WriterLockCoordinator { impl Drop for WriterLockGuard { fn drop(&mut self) { - let coordination_lock = match self.coordinator.lock_coordination() { + let coordination_lock = match self.coordinator.open_coordination() { Ok(lock) => lock, Err(err) => { warn!("failed to coordinate thread writer lock cleanup: {err}"); @@ -194,6 +215,16 @@ impl Drop for WriterLockGuard { } }; + match coordination_lock.try_lock() { + Ok(()) => {} + // Leave the file for the next writer. Dropping the guard still closes it. + Err(fs::TryLockError::WouldBlock) => return, + Err(fs::TryLockError::Error(err)) => { + warn!("failed to coordinate thread writer lock cleanup: {err}"); + return; + } + } + // Close the writer lock before deleting it so cleanup works on Windows too. drop(self.file.take()); if let Err(err) = fs::remove_file(&self.path) diff --git a/codex-rs/rollout/src/writer_lock_tests.rs b/codex-rs/rollout/src/writer_lock_tests.rs index 89e42c1f40..1b60d86c50 100644 --- a/codex-rs/rollout/src/writer_lock_tests.rs +++ b/codex-rs/rollout/src/writer_lock_tests.rs @@ -1,5 +1,7 @@ use std::fs; use std::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; use codex_protocol::ThreadId; use tempfile::TempDir; @@ -25,6 +27,18 @@ fn writer_locks_reject_competing_owners_and_release_their_files() { .join(format!("{thread_id}.lock")); assert!(lock_path.exists()); + // Another process is opening a writer. Distinct threads can start together, + // but even during this overlap the same thread must still have one owner. + let opening = fs::OpenOptions::new() + .read(true) + .write(true) + .open( + home.path() + .join(WRITER_LOCK_DIR) + .join(COORDINATION_LOCK_FILE), + ) + .expect("open coordination"); + opening.lock_shared().expect("another writer is opening"); let err = match secondary.acquire(thread_id) { Ok(_) => panic!("competing owner should fail"), Err(err) => err, @@ -34,11 +48,21 @@ fn writer_locks_reject_competing_owners_and_release_their_files() { .acquire(other_thread_id) .expect("other thread should acquire its own lock"); - drop(owner); - assert!(!lock_path.exists()); - let next_owner = secondary - .acquire(thread_id) - .expect("released thread should accept another owner"); + let next_owner = std::thread::scope(|scope| { + let _opening = opening; + let (closed, wait_for_close) = mpsc::channel(); + scope.spawn(move || { + drop(owner); + let _ = closed.send(()); + }); + wait_for_close + .recv_timeout(Duration::from_secs(/*secs*/ 10)) + .expect("writer should close while another writer is opening"); + assert!(lock_path.exists()); + secondary + .acquire(thread_id) + .expect("released thread should accept another owner") + }); drop(next_owner); drop(other_owner);