From d12b99159b97ad25aec4ed55628cf1bd66f14e14 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Thu, 14 Aug 2025 21:43:11 -0700 Subject: [PATCH] fix: prefer RwLock to Mutex --- codex-rs/core/src/codex.rs | 44 +++++++++++++++++++------------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 194ac6a922..026dfd6c0d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -7,7 +7,7 @@ use std::collections::HashSet; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; -use std::sync::Mutex; +use std::sync::RwLock; use std::sync::atomic::AtomicU64; use std::time::Duration; @@ -241,8 +241,8 @@ pub(crate) struct Session { /// Optional rollout recorder for persisting the conversation transcript so /// sessions can be replayed or inspected later. - rollout: Mutex>, - state: Mutex, + rollout: RwLock>, + state: RwLock, codex_linux_sandbox_exe: Option, user_shell: shell::Shell, show_raw_agent_reasoning: bool, @@ -461,8 +461,8 @@ impl Session { cwd, mcp_connection_manager, notify, - state: Mutex::new(state), - rollout: Mutex::new(rollout_recorder), + state: RwLock::new(state), + rollout: RwLock::new(rollout_recorder), codex_linux_sandbox_exe: config.codex_linux_sandbox_exe.clone(), disable_response_storage, user_shell: default_shell, @@ -521,7 +521,7 @@ impl Session { } pub fn set_task(&self, task: AgentTask) { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); if let Some(current_task) = state.current_task.take() { current_task.abort(); } @@ -529,7 +529,7 @@ impl Session { } pub fn remove_task(&self, sub_id: &str) { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); if let Some(task) = &state.current_task { if task.sub_id == sub_id { state.current_task.take(); @@ -565,7 +565,7 @@ impl Session { }; let _ = self.tx_event.send(event).await; { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); state.pending_approvals.insert(sub_id, tx_approve); } rx_approve @@ -591,21 +591,21 @@ impl Session { }; let _ = self.tx_event.send(event).await; { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); state.pending_approvals.insert(sub_id, tx_approve); } rx_approve } pub fn notify_approval(&self, sub_id: &str, decision: ReviewDecision) { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); if let Some(tx_approve) = state.pending_approvals.remove(sub_id) { tx_approve.send(decision).ok(); } } pub fn add_approved_command(&self, cmd: Vec) { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); state.approved_commands.insert(cmd); } @@ -615,14 +615,14 @@ impl Session { debug!("Recording items for conversation: {items:?}"); self.record_state_snapshot(items).await; - self.state.lock().unwrap().history.record_items(items); + self.state.write().unwrap().history.record_items(items); } async fn record_state_snapshot(&self, items: &[ResponseItem]) { let snapshot = { crate::rollout::SessionStateSnapshot {} }; let recorder = { - let guard = self.rollout.lock().unwrap(); + let guard = self.rollout.read().unwrap(); guard.as_ref().cloned() }; @@ -800,12 +800,12 @@ impl Session { /// Build the full turn input by concatenating the current conversation /// history with additional items for this turn. pub fn turn_input_with_history(&self, extra: Vec) -> Vec { - [self.state.lock().unwrap().history.contents(), extra].concat() + [self.state.read().unwrap().history.contents(), extra].concat() } /// Returns the input if there was no task running to inject into pub fn inject_input(&self, input: Vec) -> Result<(), Vec> { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); if state.current_task.is_some() { state.pending_input.push(input.into()); Ok(()) @@ -815,7 +815,7 @@ impl Session { } pub fn get_pending_input(&self) -> Vec { - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); if state.pending_input.is_empty() { Vec::with_capacity(0) } else { @@ -839,7 +839,7 @@ impl Session { fn abort(&self) { info!("Aborting existing session"); - let mut state = self.state.lock().unwrap(); + let mut state = self.state.write().unwrap(); state.pending_approvals.clear(); state.pending_input.clear(); if let Some(task) = state.current_task.take() { @@ -1043,7 +1043,7 @@ async fn submission_loop(sess: Arc, config: Arc, rx_sub: Receiv // Gracefully flush and shutdown rollout recorder on session end so tests // that inspect the rollout file do not race with the background writer. - let recorder_opt = sess.rollout.lock().unwrap().take(); + let recorder_opt = sess.rollout.write().unwrap().take(); if let Some(rec) = recorder_opt { if let Err(e) = rec.shutdown().await { warn!("failed to shutdown rollout recorder: {e}"); @@ -1459,7 +1459,7 @@ async fn try_run_turn( } ResponseEvent::OutputTextDelta(delta) => { { - let mut st = sess.state.lock().unwrap(); + let mut st = sess.state.write().unwrap(); st.history.append_assistant_text(&delta); } @@ -1575,7 +1575,7 @@ async fn run_compact_task( }; sess.send_event(event).await; - let mut state = sess.state.lock().unwrap(); + let mut state = sess.state.write().unwrap(); state.history.keep_last_messages(1); } @@ -1886,7 +1886,7 @@ async fn handle_container_exec_with_params( } None => { let safety = { - let state = sess.state.lock().unwrap(); + let state = sess.state.read().unwrap(); assess_command_safety( ¶ms.command, sess.approval_policy, @@ -2226,7 +2226,7 @@ async fn drain_to_completed(sess: &Session, sub_id: &str, prompt: &Prompt) -> Co match event { Ok(ResponseEvent::OutputItemDone(item)) => { // Record only to in-memory conversation history; avoid state snapshot. - let mut state = sess.state.lock().unwrap(); + let mut state = sess.state.write().unwrap(); state.history.record_items(std::slice::from_ref(&item)); } Ok(ResponseEvent::Completed {