You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/08/27 16:20:52 UTC
(opendal) branch xuanwo/python-blocking-operator-release-gil created (now 04f8e9b776)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/python-blocking-operator-release-gil
in repository https://gitbox.apache.org/repos/asf/opendal.git
at 04f8e9b776 perf(bindings/python): release GIL during blocking I/O
This branch includes the following new commits:
new 04f8e9b776 perf(bindings/python): release GIL during blocking I/O
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/01: perf(bindings/python): release GIL during blocking I/O
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/python-blocking-operator-release-gil
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit 04f8e9b7769259e9a5c2de113588a746aad6adba
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Thu Aug 27 17:24:58 2026 +0800
perf(bindings/python): release GIL during blocking I/O
---
bindings/python/benchmark/README.md | 19 +++
.../python/benchmark/blocking_gil_benchmark.py | 156 +++++++++++++++++++++
bindings/python/src/operator.rs | 88 +++++++-----
bindings/python/tests/test_sync_gil.py | 152 ++++++++++++++++++++
4 files changed, 379 insertions(+), 36 deletions(-)
diff --git a/bindings/python/benchmark/README.md b/bindings/python/benchmark/README.md
index d5f2024daf..98ddc75700 100644
--- a/bindings/python/benchmark/README.md
+++ b/bindings/python/benchmark/README.md
@@ -24,3 +24,22 @@ export AWS_S3_BUCKET=opendal
uv run async_opendal_benchmark.py
uv run async_origin_s3_benchmark_with_gevent.py
```
+
+## Blocking Operator GIL benchmark
+
+`blocking_gil_benchmark.py` uses a local HTTP subprocess with a configurable
+response delay, so it does not require service credentials. It reports the
+median and p95 latency for one blocking read, two sequential reads, and two
+reads started from independent Python threads.
+
+From `bindings/python`, build the binding from the revision to measure, then
+run:
+
+```shell
+uv run maturin develop --release
+uv run benchmark/blocking_gil_benchmark.py \
+ --delay-ms 50 --iterations 30 --warmups 3
+```
+
+Run the same command from an unchanged baseline checkout to compare
+single-thread overhead and concurrency speedup with identical local I/O.
diff --git a/bindings/python/benchmark/blocking_gil_benchmark.py b/bindings/python/benchmark/blocking_gil_benchmark.py
new file mode 100644
index 0000000000..ffa9fd9961
--- /dev/null
+++ b/bindings/python/benchmark/blocking_gil_benchmark.py
@@ -0,0 +1,156 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import argparse
+import json
+import multiprocessing
+import statistics
+import threading
+import time
+from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from multiprocessing.connection import Connection
+
+
+class _DelayedHandler(BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ time.sleep(self.server.response_delay_seconds)
+ body = b"benchmark response"
+ self.send_response(200)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+
+ def log_message(self, format_: str, *args: object) -> None:
+ pass
+
+
+def _serve_delayed_http(port_sender: Connection, delay_seconds: float) -> None:
+ server = ThreadingHTTPServer(("127.0.0.1", 0), _DelayedHandler)
+ server.response_delay_seconds = delay_seconds
+ port_sender.send(server.server_address[1])
+ port_sender.close()
+ server.serve_forever(poll_interval=0.01)
+
+
+def _measure(operation: Callable[[], object], iterations: int) -> list[float]:
+ samples = []
+ for _ in range(iterations):
+ started_at = time.perf_counter()
+ operation()
+ samples.append((time.perf_counter() - started_at) * 1_000)
+ return samples
+
+
+def _summarize(samples: list[float]) -> dict[str, float]:
+ sorted_samples = sorted(samples)
+ p95_index = max(0, (len(sorted_samples) * 95 + 99) // 100 - 1)
+ return {
+ "median_ms": round(statistics.median(sorted_samples), 3),
+ "p95_ms": round(sorted_samples[p95_index], 3),
+ }
+
+
+def _parse_args() -> argparse.Namespace:
+ parser = argparse.ArgumentParser(
+ description=(
+ "Benchmark blocking Operator GIL release with local delayed HTTP I/O."
+ )
+ )
+ parser.add_argument("--delay-ms", type=float, default=50.0)
+ parser.add_argument("--iterations", type=int, default=30)
+ parser.add_argument("--warmups", type=int, default=3)
+ return parser.parse_args()
+
+
+def main() -> None:
+ import opendal
+
+ args = _parse_args()
+ if args.delay_ms < 0:
+ message = "--delay-ms must be non-negative"
+ raise ValueError(message)
+ if args.iterations < 1:
+ message = "--iterations must be positive"
+ raise ValueError(message)
+ if args.warmups < 0:
+ message = "--warmups must be non-negative"
+ raise ValueError(message)
+
+ context = multiprocessing.get_context("spawn")
+ port_receiver, port_sender = context.Pipe(duplex=False)
+ server_process = context.Process(
+ target=_serve_delayed_http,
+ args=(port_sender, args.delay_ms / 1_000),
+ )
+ server_process.start()
+ port_sender.close()
+
+ try:
+ if not port_receiver.poll(10):
+ message = "delayed HTTP benchmark server did not start"
+ raise RuntimeError(message)
+ port = port_receiver.recv()
+ operator = opendal.Operator("http", endpoint=f"http://127.0.0.1:{port}")
+
+ for _ in range(args.warmups):
+ operator.read("warmup")
+
+ single_read = _measure(lambda: operator.read("single"), args.iterations)
+ sequential_pair = _measure(
+ lambda: (operator.read("first"), operator.read("second")),
+ args.iterations,
+ )
+
+ with ThreadPoolExecutor(max_workers=2) as executor:
+
+ def concurrent_pair() -> list[bytes]:
+ start = threading.Barrier(3)
+
+ def read(path: str) -> bytes:
+ start.wait()
+ return operator.read(path)
+
+ futures = [
+ executor.submit(read, "first"),
+ executor.submit(read, "second"),
+ ]
+ start.wait()
+ return [future.result() for future in futures]
+
+ concurrent_pair_samples = _measure(concurrent_pair, args.iterations)
+
+ sequential_median = statistics.median(sequential_pair)
+ concurrent_median = statistics.median(concurrent_pair_samples)
+ result = {
+ "delay_ms": args.delay_ms,
+ "iterations": args.iterations,
+ "single_read": _summarize(single_read),
+ "sequential_two_reads": _summarize(sequential_pair),
+ "concurrent_two_reads": _summarize(concurrent_pair_samples),
+ "concurrency_speedup": round(sequential_median / concurrent_median, 3),
+ }
+ print(json.dumps(result, indent=2, sort_keys=True))
+ finally:
+ server_process.terminate()
+ server_process.join(10)
+ port_receiver.close()
+
+
+if __name__ == "__main__":
+ main()
diff --git a/bindings/python/src/operator.rs b/bindings/python/src/operator.rs
index 7b75b39471..5c1fad2045 100644
--- a/bindings/python/src/operator.rs
+++ b/bindings/python/src/operator.rs
@@ -338,6 +338,7 @@ impl Operator {
#[pyo3(signature = (path, mode, *, **kwargs: "Unpack[OpenKwargs]"))]
pub fn open(
&self,
+ py: Python<'_>,
path: PathBuf,
mode: String,
kwargs: Option<&Bound<PyDict>>,
@@ -357,17 +358,16 @@ impl Operator {
if mode == "rb" {
let range = reader_opts.make_range();
- let reader = this
- .reader_options(&path, reader_opts.into())
- .map_err(format_pyerr)?;
-
- let r = reader
- .into_std_read(range.to_range())
+ let r = py
+ .detach(move || {
+ let reader = this.reader_options(&path, reader_opts.into())?;
+ reader.into_std_read(range.to_range())
+ })
.map_err(format_pyerr)?;
Ok(File::new_reader(r))
} else if mode == "wb" {
- let writer = this
- .writer_options(&path, writer_opts.into())
+ let writer = py
+ .detach(move || this.writer_options(&path, writer_opts.into()))
.map_err(format_pyerr)?;
Ok(File::new_writer(writer))
} else {
@@ -468,9 +468,9 @@ impl Operator {
cache_control,
content_disposition,
};
- let buffer = self
- .core
- .read_options(&path, opts.into())
+ let this = self.core.clone();
+ let buffer = py
+ .detach(move || this.read_options(&path, opts.into()))
.map_err(format_pyerr)?;
Ok(buffer_into_py_bytes(py, buffer)?.into_any())
@@ -524,6 +524,7 @@ impl Operator {
user_metadata = None))]
pub fn write(
&self,
+ py: Python<'_>,
path: PathBuf,
bs: &Bound<PyAny>,
append: Option<bool>,
@@ -554,8 +555,8 @@ impl Operator {
user_metadata,
};
- self.core
- .write_options(&path, bs, opts.into())
+ let this = self.core.clone();
+ py.detach(move || this.write_options(&path, bs, opts.into()))
.map(|_| ())
.map_err(format_pyerr)
}
@@ -599,6 +600,7 @@ impl Operator {
content_disposition=None))]
pub fn stat(
&self,
+ py: Python<'_>,
path: PathBuf,
version: Option<String>,
if_match: Option<String>,
@@ -620,8 +622,8 @@ impl Operator {
cache_control,
content_disposition,
};
- self.core
- .stat_options(&path, opts.into())
+ let this = self.core.clone();
+ py.detach(move || this.stat_options(&path, opts.into()))
.map_err(format_pyerr)
.map(Metadata::new)
}
@@ -634,11 +636,11 @@ impl Operator {
/// The path to the source file.
/// target : str
/// The path to the target file.
- pub fn copy(&self, source: PathBuf, target: PathBuf) -> PyResult<()> {
+ pub fn copy(&self, py: Python<'_>, source: PathBuf, target: PathBuf) -> PyResult<()> {
let source = source.to_string_lossy().to_string();
let target = target.to_string_lossy().to_string();
- self.core
- .copy(&source, &target)
+ let this = self.core.clone();
+ py.detach(move || this.copy(&source, &target))
.map(|_| ())
.map_err(format_pyerr)
}
@@ -651,10 +653,12 @@ impl Operator {
/// The path to the source file.
/// target : str
/// The path to the target file.
- pub fn rename(&self, source: PathBuf, target: PathBuf) -> PyResult<()> {
+ pub fn rename(&self, py: Python<'_>, source: PathBuf, target: PathBuf) -> PyResult<()> {
let source = source.to_string_lossy().to_string();
let target = target.to_string_lossy().to_string();
- self.core.rename(&source, &target).map_err(format_pyerr)
+ let this = self.core.clone();
+ py.detach(move || this.rename(&source, &target))
+ .map_err(format_pyerr)
}
/// Recursively remove all files and directories at the given path.
@@ -663,18 +667,20 @@ impl Operator {
/// ----------
/// path : str
/// The path to remove.
- pub fn remove_all(&self, path: PathBuf) -> PyResult<()> {
+ pub fn remove_all(&self, py: Python<'_>, path: PathBuf) -> PyResult<()> {
use ocore::options::DeleteOptions;
let path = path.to_string_lossy().to_string();
- self.core
- .delete_options(
+ let this = self.core.clone();
+ py.detach(move || {
+ this.delete_options(
&path,
DeleteOptions {
recursive: true,
..Default::default()
},
)
- .map_err(format_pyerr)
+ })
+ .map_err(format_pyerr)
}
/// Create a directory at the given path.
@@ -688,9 +694,11 @@ impl Operator {
/// ----------
/// path : str
/// The path to the directory.
- pub fn create_dir(&self, path: PathBuf) -> PyResult<()> {
+ pub fn create_dir(&self, py: Python<'_>, path: PathBuf) -> PyResult<()> {
let path = path.to_string_lossy().to_string();
- self.core.create_dir(&path).map_err(format_pyerr)
+ let this = self.core.clone();
+ py.detach(move || this.create_dir(&path))
+ .map_err(format_pyerr)
}
/// Delete a file at the given path.
@@ -713,12 +721,14 @@ impl Operator {
#[pyo3(signature = (path, *, version=None, recursive=None, if_match=None))]
pub fn delete(
&self,
+ py: Python<'_>,
path: PathBuf,
version: Option<String>,
recursive: Option<bool>,
if_match: Option<String>,
) -> PyResult<()> {
let path = path.to_string_lossy().to_string();
+ let this = self.core.clone();
if version.is_some() || recursive.is_some() || if_match.is_some() {
let opts = ocore::options::DeleteOptions {
version,
@@ -726,9 +736,10 @@ impl Operator {
if_match,
..Default::default()
};
- self.core.delete_options(&path, opts).map_err(format_pyerr)
+ py.detach(move || this.delete_options(&path, opts))
+ .map_err(format_pyerr)
} else {
- self.core.delete(&path).map_err(format_pyerr)
+ py.detach(move || this.delete(&path)).map_err(format_pyerr)
}
}
@@ -743,9 +754,10 @@ impl Operator {
/// -------
/// bool
/// True if the path exists, False otherwise.
- pub fn exists(&self, path: PathBuf) -> PyResult<bool> {
+ pub fn exists(&self, py: Python<'_>, path: PathBuf) -> PyResult<bool> {
let path = path.to_string_lossy().to_string();
- self.core.exists(&path).map_err(format_pyerr)
+ let this = self.core.clone();
+ py.detach(move || this.exists(&path)).map_err(format_pyerr)
}
/// List entries in the given directory.
@@ -769,6 +781,7 @@ impl Operator {
/// -------
/// BlockingLister
/// An iterator over the entries in the directory.
+ #[allow(clippy::too_many_arguments)]
#[pyo3(signature = (path, *,
limit=None,
start_after=None,
@@ -777,6 +790,7 @@ impl Operator {
deleted=None) -> "collections.abc.Iterable[Entry]")]
pub fn list(
&self,
+ py: Python<'_>,
path: PathBuf,
limit: Option<usize>,
start_after: Option<String>,
@@ -794,9 +808,9 @@ impl Operator {
deleted,
};
- let l = self
- .core
- .lister_options(&path, opts.into())
+ let this = self.core.clone();
+ let l = py
+ .detach(move || this.lister_options(&path, opts.into()))
.map_err(format_pyerr)?;
Ok(BlockingLister::new(l))
}
@@ -831,13 +845,14 @@ impl Operator {
deleted=None) -> "collections.abc.Iterable[Entry]")]
pub fn scan(
&self,
+ py: Python<'_>,
path: PathBuf,
limit: Option<usize>,
start_after: Option<String>,
versions: Option<bool>,
deleted: Option<bool>,
) -> PyResult<BlockingLister> {
- self.list(path, limit, start_after, Some(true), versions, deleted)
+ self.list(py, path, limit, start_after, Some(true), versions, deleted)
}
/// Get all capabilities of this operator.
@@ -856,8 +871,9 @@ impl Operator {
/// ------
/// Exception
/// If the operator is not able to work correctly.
- pub fn check(&self) -> PyResult<()> {
- self.core.check().map_err(format_pyerr)
+ pub fn check(&self, py: Python<'_>) -> PyResult<()> {
+ let this = self.core.clone();
+ py.detach(move || this.check()).map_err(format_pyerr)
}
/// Create a new `AsyncOperator` from this blocking operator.
diff --git a/bindings/python/tests/test_sync_gil.py b/bindings/python/tests/test_sync_gil.py
new file mode 100644
index 0000000000..8b1d2482c8
--- /dev/null
+++ b/bindings/python/tests/test_sync_gil.py
@@ -0,0 +1,152 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import multiprocessing
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+import pytest
+
+import opendal
+
+SERVER_FALLBACK_SECONDS = 2
+TEST_TIMEOUT_SECONDS = 6
+
+
+class _CoordinatedHandler(BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ server = self.server
+ with server.active_lock:
+ server.active_requests += 1
+ server.first_request_started.set()
+ if server.active_requests == 2:
+ server.two_requests_active.set()
+
+ try:
+ server.release_responses.wait(SERVER_FALLBACK_SECONDS)
+ body = b"blocking I/O completed"
+ self.send_response(200)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ self.wfile.write(body)
+ finally:
+ with server.active_lock:
+ server.active_requests -= 1
+
+ def log_message(self, format_, *args: object) -> None:
+ pass
+
+
+def _serve_coordinated_http(
+ port_sender,
+ first_request_started,
+ two_requests_active,
+ release_responses,
+) -> None:
+ server = ThreadingHTTPServer(("127.0.0.1", 0), _CoordinatedHandler)
+ server.active_lock = threading.Lock()
+ server.active_requests = 0
+ server.first_request_started = first_request_started
+ server.two_requests_active = two_requests_active
+ server.release_responses = release_responses
+ port_sender.send(server.server_address[1])
+ port_sender.close()
+ server.serve_forever(poll_interval=0.01)
+
+
+def test_blocking_operator_releases_gil(service_name, operator) -> None:
+ if service_name != "memory":
+ pytest.skip("run the standalone GIL regression test once")
+
+ context = multiprocessing.get_context("spawn")
+ first_request_started = context.Event()
+ two_requests_active = context.Event()
+ release_responses = context.Event()
+ port_receiver, port_sender = context.Pipe(duplex=False)
+ server_process = context.Process(
+ target=_serve_coordinated_http,
+ args=(
+ port_sender,
+ first_request_started,
+ two_requests_active,
+ release_responses,
+ ),
+ )
+ server_process.start()
+ port_sender.close()
+
+ read_completed = threading.Event()
+ python_thread_progressed = threading.Event()
+ progress_while_read_blocked = []
+ results = []
+ errors = []
+
+ try:
+ assert port_receiver.poll(TEST_TIMEOUT_SECONDS), (
+ "HTTP test server did not start"
+ )
+ port = port_receiver.recv()
+ op = opendal.Operator("http", endpoint=f"http://127.0.0.1:{port}")
+
+ def read(path) -> None:
+ try:
+ results.append(op.read(path))
+ except Exception as error: # noqa: BLE001
+ errors.append(error)
+ finally:
+ read_completed.set()
+
+ def record_python_progress() -> None:
+ request_started = first_request_started.wait(TEST_TIMEOUT_SECONDS)
+ progress_while_read_blocked.append(
+ request_started and not read_completed.is_set()
+ )
+ python_thread_progressed.set()
+
+ progress_thread = threading.Thread(target=record_python_progress)
+ read_threads = [
+ threading.Thread(target=read, args=("first",)),
+ threading.Thread(target=read, args=("second",)),
+ ]
+
+ progress_thread.start()
+ for thread in read_threads:
+ thread.start()
+
+ assert two_requests_active.wait(TEST_TIMEOUT_SECONDS), (
+ "two blocking reads did not overlap"
+ )
+ assert python_thread_progressed.wait(TEST_TIMEOUT_SECONDS), (
+ "another Python thread did not run during blocking I/O"
+ )
+ assert progress_while_read_blocked == [True]
+ assert all(thread.is_alive() for thread in read_threads)
+
+ release_responses.set()
+ for thread in read_threads:
+ thread.join(TEST_TIMEOUT_SECONDS)
+ progress_thread.join(TEST_TIMEOUT_SECONDS)
+
+ assert all(not thread.is_alive() for thread in read_threads)
+ assert not progress_thread.is_alive()
+ assert errors == []
+ assert results == [b"blocking I/O completed"] * 2
+ finally:
+ release_responses.set()
+ server_process.terminate()
+ server_process.join(TEST_TIMEOUT_SECONDS)
+ port_receiver.close()