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/28 05:44:09 UTC
(opendal) branch main updated: perf(bindings/python): release GIL during blocking operator I/O (#8180)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new 5a9a58bb75 perf(bindings/python): release GIL during blocking operator I/O (#8180)
5a9a58bb75 is described below
commit 5a9a58bb75528fd9b360e9facbb7688e135e5f3c
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Fri Aug 28 13:43:54 2026 +0800
perf(bindings/python): release GIL during blocking operator I/O (#8180)
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()