exec-server: isolate sync remote path rpc

This commit is contained in:
starr-openai
2026-05-29 15:50:30 -07:00
parent 0942afd54a
commit 1b3cadae24
2 changed files with 65 additions and 22 deletions
+4
View File
@@ -214,6 +214,10 @@ impl LazyRemoteExecServerClient {
}
}
pub(crate) fn new_connection_client(&self) -> Self {
Self::new(self.transport_params.clone())
}
pub(crate) async fn get(&self) -> Result<ExecServerClient, ExecServerError> {
if let Some(client) = self.connected_client() {
return Ok(client);
+61 -22
View File
@@ -4,8 +4,11 @@ use base64::engine::general_purpose::STANDARD;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::future::Future;
use std::path::Path;
use std::sync::OnceLock;
use std::thread;
use tokio::io;
use tokio::runtime::Builder;
use tokio::runtime::Runtime;
use tracing::trace;
use crate::CopyOptions;
@@ -33,15 +36,20 @@ use crate::protocol::FsWriteFileParams;
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
const NOT_FOUND_ERROR_CODE: i64 = -32004;
#[derive(Clone)]
pub(crate) struct RemoteFileSystem {
client: LazyRemoteExecServerClient,
async_client: LazyRemoteExecServerClient,
sync_client: LazyRemoteExecServerClient,
sync_call_runtime: SyncCallRuntime,
}
impl RemoteFileSystem {
pub(crate) fn new(client: LazyRemoteExecServerClient) -> Self {
trace!("remote fs new");
Self { client }
Self {
async_client: client,
sync_client: client.new_connection_client(),
sync_call_runtime: SyncCallRuntime::default(),
}
}
fn run_sync_remote_call<T, F, Fut>(
@@ -54,24 +62,55 @@ impl RemoteFileSystem {
F: FnOnce(ExecServerClient) -> Fut + Send + 'static,
Fut: Future<Output = FileSystemResult<T>> + Send + 'static,
{
let runtime = tokio::runtime::Handle::try_current().map_err(|err| {
io::Error::other(format!("remote fs {operation} requires runtime: {err}"))
})?;
let client = self.client.clone();
let runtime = self.sync_call_runtime.get()?;
let client = self.sync_client.clone();
// ExecutorFileSystem exposes synchronous path operations, so remote callers have to hop
// through the active runtime from a helper thread to reach the JSON-RPC filesystem
// surface without blocking the runtime worker.
thread::spawn(move || {
runtime.block_on(async move {
let client = client.get().await.map_err(map_remote_error)?;
call(client).await
})
// through a dedicated exec-server client/runtime from a helper thread. Reusing the caller
// runtime can deadlock current-thread runtimes because the JSON-RPC reader also needs that
// runtime to make progress.
thread::scope(|scope| {
scope
.spawn(move || {
runtime.block_on(async move {
let client = client.get().await.map_err(map_remote_error)?;
call(client).await
})
})
.join()
})
.join()
.map_err(|_| io::Error::other(format!("remote fs {operation} thread panicked")))?
}
}
#[derive(Default)]
struct SyncCallRuntime {
runtime: OnceLock<Result<Runtime, String>>,
}
impl SyncCallRuntime {
fn get(&self) -> FileSystemResult<&Runtime> {
self.runtime
.get_or_init(|| {
Builder::new_multi_thread()
.worker_threads(1)
.thread_name("codex-remote-fs")
.enable_all()
.build()
.map_err(|err| err.to_string())
})
.as_ref()
.map_err(|err| io::Error::other(format!("failed to start remote fs runtime: {err}")))
}
}
impl Drop for SyncCallRuntime {
fn drop(&mut self) {
if let Some(Ok(runtime)) = self.runtime.take() {
runtime.shutdown_background();
}
}
}
#[async_trait]
impl ExecutorFileSystem for RemoteFileSystem {
fn canonicalize(&self, path: &AbsolutePathBuf) -> FileSystemResult<AbsolutePathBuf> {
@@ -124,7 +163,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<Vec<u8>> {
trace!("remote fs read_file");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
let response = client
.fs_read_file(FsReadFileParams {
path: path.clone(),
@@ -147,7 +186,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<()> {
trace!("remote fs write_file");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
client
.fs_write_file(FsWriteFileParams {
path: path.clone(),
@@ -166,7 +205,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<()> {
trace!("remote fs create_directory");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
client
.fs_create_directory(FsCreateDirectoryParams {
path: path.clone(),
@@ -184,7 +223,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<FileMetadata> {
trace!("remote fs get_metadata");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
let response = client
.fs_get_metadata(FsGetMetadataParams {
path: path.clone(),
@@ -207,7 +246,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<Vec<ReadDirectoryEntry>> {
trace!("remote fs read_directory");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
let response = client
.fs_read_directory(FsReadDirectoryParams {
path: path.clone(),
@@ -233,7 +272,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<()> {
trace!("remote fs remove");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
client
.fs_remove(FsRemoveParams {
path: path.clone(),
@@ -254,7 +293,7 @@ impl ExecutorFileSystem for RemoteFileSystem {
sandbox: Option<&FileSystemSandboxContext>,
) -> FileSystemResult<()> {
trace!("remote fs copy");
let client = self.client.get().await.map_err(map_remote_error)?;
let client = self.async_client.get().await.map_err(map_remote_error)?;
client
.fs_copy(FsCopyParams {
source_path: source_path.clone(),