Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion python/strands_shell/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -358,7 +358,11 @@ def __init__(

def run(self, command: str) -> Output:
"""Run a command and capture its output. Never raises for command-level
failures — check :attr:`Output.status`."""
failures — check :attr:`Output.status`.

Blocking; releases the GIL while the command runs. A ``Shell`` must be
created, used and dropped on one thread, so from async code run it in a
single-worker executor rather than ``asyncio.to_thread``."""
return self._shell.run(command)

# ---- Environment ----
Expand Down
50 changes: 33 additions & 17 deletions src/python.rs
Original file line number Diff line number Diff line change
Expand Up @@ -423,6 +423,33 @@ impl ShellBuilder {
}
}

/// Drives `fut` to completion with the GIL released, so other Python threads
/// (and an asyncio loop on another thread) keep running while a command executes.
fn block_on_detached<F>(py: Python<'_>, runtime: &tokio::runtime::Runtime, fut: F) -> F::Output
where
F: std::future::Future,
F::Output: Send,
{
/// Lets a `!Send` value cross [`Python::detach`]. Scoped to this function
/// so it can't be reused somewhere the value would leave the thread.
struct SameThread<T>(T);
// SAFETY: `Python::detach` runs its closure synchronously on the calling
// thread, so the wrapped value never crosses a thread boundary. `T` must hold
// no Python references (`Py`, `Bound`, `Python`); the `Send` bound on `detach`
// is PyO3's stand-in for that check, and this impl bypasses it.
unsafe impl<T> Send for SameThread<T> {}

impl<T> SameThread<T> {
/// Unwraps by value, so a closure captures the whole wrapper rather than its `!Send` field.
fn into_inner(self) -> T {
self.0
}
}

let job = SameThread(fut);
py.detach(move || runtime.block_on(tokio::task::LocalSet::new().run_until(job.into_inner())))
}

/// A sandboxed shell environment.
#[pyclass(unsendable)]
pub struct Shell {
Expand Down Expand Up @@ -452,13 +479,12 @@ impl Shell {
}

/// Run a command and capture output.
fn run(&mut self, command: &str) -> PyResult<Output> {
fn run(&mut self, py: Python<'_>, command: &str) -> PyResult<Output> {
let shell = self
.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("shell consumed"))?;
let local = tokio::task::LocalSet::new();
let output = self.runtime.block_on(local.run_until(shell.run(command)));
let output = block_on_detached(py, &self.runtime, shell.run(command));
Ok(Output {
status: output.status,
stdout: output.stdout,
Expand Down Expand Up @@ -543,10 +569,7 @@ impl Shell {
.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("shell consumed"))?;
let local = tokio::task::LocalSet::new();
let bytes = self
.runtime
.block_on(local.run_until(shell.read_file(path)))
let bytes = block_on_detached(py, &self.runtime, shell.read_file(path))
.map_err(|e| native_file_error(py, path, &e))?;
Ok(PyBytes::new(py, &bytes))
}
Expand All @@ -560,9 +583,7 @@ impl Shell {
.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("shell consumed"))?;
let local = tokio::task::LocalSet::new();
self.runtime
.block_on(local.run_until(shell.write_file(path, content)))
block_on_detached(py, &self.runtime, shell.write_file(path, content))
.map_err(|e| native_file_error(py, path, &e))
}

Expand All @@ -572,9 +593,7 @@ impl Shell {
.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("shell consumed"))?;
let local = tokio::task::LocalSet::new();
self.runtime
.block_on(local.run_until(shell.remove_file(path)))
block_on_detached(py, &self.runtime, shell.remove_file(path))
.map_err(|e| native_file_error(py, path, &e))
}

Expand All @@ -587,10 +606,7 @@ impl Shell {
.inner
.as_mut()
.ok_or_else(|| PyRuntimeError::new_err("shell consumed"))?;
let local = tokio::task::LocalSet::new();
let infos = self
.runtime
.block_on(local.run_until(shell.list_files(path)))
let infos = block_on_detached(py, &self.runtime, shell.list_files(path))
.map_err(|e| native_file_error(py, path, &e))?;
Ok(infos
.into_iter()
Expand Down
26 changes: 26 additions & 0 deletions tests/python/test_bindings.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@
import os
import shutil
import tempfile
import threading
import time

import pytest

Expand Down Expand Up @@ -266,3 +268,27 @@ def test_omitted_and_positive_timeout_allowed():
# Omitted timeout => no limit; a positive value => bounded. Both build.
assert strands_shell.Shell().run("echo ok").stdout.strip() == "ok"
assert strands_shell.Shell(timeout=5.0).run("echo ok").stdout.strip() == "ok"


def test_run_releases_the_gil(shell):
"""Other Python threads keep running while a command executes (#119)."""
ticks = 0
stop = threading.Event()

def tick():
nonlocal ticks
while not stop.is_set():
ticks += 1
time.sleep(0.001)

worker = threading.Thread(target=tick)
worker.start()
try:
before = ticks
shell.run("sleep 0.5")
during = ticks - before
finally:
stop.set()
worker.join()

assert during > 50