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/09/09 06:43:14 UTC
(opendal) branch main updated: fix(bindings/python): release GIL during blocking File I/O (#8267)
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 8a4a4ed976 fix(bindings/python): release GIL during blocking File I/O (#8267)
8a4a4ed976 is described below
commit 8a4a4ed976b49afa582518df7e84c2e7b253e5c7
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Sep 9 14:43:09 2026 +0800
fix(bindings/python): release GIL during blocking File I/O (#8267)
---
bindings/python/src/file.rs | 102 ++++++++++++++------------
bindings/python/tests/test_read.py | 26 +++++++
bindings/python/tests/test_sync_gil.py | 130 +++++++++++++++++++++++++++++++++
3 files changed, 210 insertions(+), 48 deletions(-)
diff --git a/bindings/python/src/file.rs b/bindings/python/src/file.rs
index a66c62b20e..ed27b15f9e 100644
--- a/bindings/python/src/file.rs
+++ b/bindings/python/src/file.rs
@@ -39,6 +39,9 @@ use crate::*;
/// A file-like object for reading and writing data.
///
/// Created by the `open` method of the `Operator` class.
+///
+/// Storage I/O releases the Python GIL while waiting. Concurrent operations on
+/// the same file are rejected; use separate files for I/O from multiple threads.
#[pyclass(module = "opendal.file")]
pub struct File(FileState);
@@ -97,11 +100,12 @@ impl File {
}
};
- let buffer = match size {
- Some(size) => reader.read_buffer(size),
- None => reader.read_to_end_buffer(),
- }
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
+ let buffer = py
+ .detach(|| match size {
+ Some(size) => reader.read_buffer(size),
+ None => reader.read_to_end_buffer(),
+ })
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
buffer_into_py_bytes(py, buffer).map(Bound::into_any)
}
@@ -144,23 +148,16 @@ impl File {
}
};
- let buffer = match size {
- None => {
- let mut buffer = Vec::new();
- reader
- .read_until(b'\n', &mut buffer)
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- buffer
- }
- Some(size) => {
+ let buffer = py
+ .detach(|| -> std::io::Result<Vec<u8>> {
let mut buffer = Vec::new();
- let mut reader = reader.take(size as u64);
- reader
- .read_until(b'\n', &mut buffer)
- .map_err(|err| PyIOError::new_err(err.to_string()))?;
- buffer
- }
- };
+ match size {
+ None => reader.read_until(b'\n', &mut buffer)?,
+ Some(size) => reader.take(size as u64).read_until(b'\n', &mut buffer)?,
+ };
+ Ok(buffer)
+ })
+ .map_err(|err| PyIOError::new_err(err.to_string()))?;
buffer_into_py_bytes(py, buffer.into()).map(Bound::into_any)
}
@@ -177,7 +174,7 @@ impl File {
/// int
/// The number of bytes read.
#[pyo3(signature = (buffer: "bytearray | memoryview"))]
- pub fn readinto(&mut self, buffer: PyBuffer<u8>) -> PyResult<usize> {
+ pub fn readinto(&mut self, py: Python<'_>, buffer: PyBuffer<u8>) -> PyResult<usize> {
let reader = match &mut self.0 {
FileState::Reader(r) => r,
FileState::Writer(_) => {
@@ -200,15 +197,22 @@ impl File {
return Err(PyIOError::new_err("Buffer is not C contiguous."));
}
- Python::attach(|_py| {
- let ptr = buffer.buf_ptr();
- let nbytes = buffer.len_bytes();
- unsafe {
- let view: &mut [u8] = std::slice::from_raw_parts_mut(ptr as *mut u8, nbytes);
- let z = Read::read(reader, view)?;
- Ok(z)
+ let size = buffer.len_bytes();
+ // Wait using owned storage, without exposing the Python buffer to I/O.
+ let data = py.detach(|| reader.read_buffer(size))?;
+ let len = data.len();
+ let target = buffer.as_mut_slice(py).expect("buffer was validated above");
+ let mut offset = 0;
+ for chunk in data {
+ for (dst, src) in target[offset..offset + chunk.len()]
+ .iter()
+ .zip(chunk.iter())
+ {
+ dst.set(*src);
}
- })
+ offset += chunk.len();
+ }
+ Ok(len)
}
/// Write bytes to this file.
@@ -223,7 +227,7 @@ impl File {
/// int
/// The number of bytes written.
#[pyo3(signature = (bs: "bytes"))]
- pub fn write(&mut self, bs: &[u8]) -> PyResult<usize> {
+ pub fn write(&mut self, py: Python<'_>, bs: &Bound<PyBytes>) -> PyResult<usize> {
let writer = match &mut self.0 {
FileState::Reader(_) => {
return Err(PyIOError::new_err(
@@ -238,8 +242,8 @@ impl File {
}
};
- writer
- .write_all(bs)
+ let bs = PyBackedBytes::from(bs.clone());
+ py.detach(|| writer.write_all(&bs))
.map(|_| bs.len())
.map_err(|err| PyIOError::new_err(err.to_string()))
}
@@ -266,7 +270,7 @@ impl File {
/// int
/// The new absolute position.
#[pyo3(signature = (pos, whence = 0))]
- pub fn seek(&mut self, pos: i64, whence: u8) -> PyResult<u64> {
+ pub fn seek(&mut self, py: Python<'_>, pos: i64, whence: u8) -> PyResult<u64> {
if !self.seekable()? {
return Err(PyIOError::new_err(
"Seek operation is not supported by the backing service.",
@@ -293,8 +297,7 @@ impl File {
_ => return Err(PyValueError::new_err("invalid whence")),
};
- reader
- .seek(whence)
+ py.detach(|| reader.seek(whence))
.map_err(|err| PyIOError::new_err(err.to_string()))
}
@@ -304,7 +307,7 @@ impl File {
/// -------
/// int
/// The current absolute position.
- pub fn tell(&mut self) -> PyResult<u64> {
+ pub fn tell(&mut self, py: Python<'_>) -> PyResult<u64> {
let reader = match &mut self.0 {
FileState::Reader(r) => r,
FileState::Writer(_) => {
@@ -319,8 +322,7 @@ impl File {
}
};
- reader
- .stream_position()
+ py.detach(|| reader.stream_position())
.map_err(|err| PyIOError::new_err(err.to_string()))
}
@@ -331,12 +333,15 @@ impl File {
/// Notes
/// -----
/// A closed file cannot be used for further I/O operations.
- fn close(&mut self) -> PyResult<()> {
- if let FileState::Writer(w) = &mut self.0 {
- w.close().map_err(format_pyerr_from_io_error)?;
- };
- self.0 = FileState::Closed;
- Ok(())
+ fn close(&mut self, py: Python<'_>) -> PyResult<()> {
+ py.detach(|| {
+ if let FileState::Writer(w) = &mut self.0 {
+ w.close()?;
+ }
+ self.0 = FileState::Closed;
+ Ok(())
+ })
+ .map_err(format_pyerr_from_io_error)
}
pub fn __enter__(slf: PyRef<'_, Self>) -> Py<Self> {
@@ -350,11 +355,12 @@ impl File {
traceback: "types.TracebackType | None"))]
pub fn __exit__(
&mut self,
+ py: Python<'_>,
exc_type: Py<PyAny>,
exc_value: Py<PyAny>,
traceback: Py<PyAny>,
) -> PyResult<()> {
- self.close()
+ self.close(py)
}
/// Flush the underlying writer.
@@ -362,11 +368,11 @@ impl File {
/// Notes
/// -----
/// Is a no-op if the file is not `writable`.
- pub fn flush(&mut self) -> PyResult<()> {
+ pub fn flush(&mut self, py: Python<'_>) -> PyResult<()> {
if matches!(self.0, FileState::Reader(_)) {
Ok(())
} else if let FileState::Writer(w) = &mut self.0 {
- match w.flush() {
+ match py.detach(|| w.flush()) {
Ok(_) => Ok(()),
Err(e) => Err(e.into()),
}
diff --git a/bindings/python/tests/test_read.py b/bindings/python/tests/test_read.py
index c4ebc32351..b0b33b506a 100644
--- a/bindings/python/tests/test_read.py
+++ b/bindings/python/tests/test_read.py
@@ -338,3 +338,29 @@ def test_sync_conditional_reads(service_name, operator):
# Should fail: file was modified after `before`
with pytest.raises(ConditionNotMatch):
operator.read(path, if_unmodified_since=before)
+
+
+@pytest.mark.need_capability("read", "write", "delete")
+def test_sync_readinto_buffer_boundaries(operator):
+ filename = f"readinto_{uuid4()}"
+ operator.write(filename, b"abcdef")
+ try:
+ with operator.open(filename, "rb") as reader:
+ with pytest.raises(OSError, match="not writable"):
+ reader.readinto(b"readonly")
+ with pytest.raises(OSError, match="not C contiguous"):
+ reader.readinto(memoryview(bytearray(8))[::2])
+ assert reader.tell() == 0
+ assert reader.readinto(bytearray()) == 0
+ assert reader.tell() == 0
+ target = bytearray(b"??????????")
+ assert reader.readinto(memoryview(target)[2:5]) == 3
+ assert target == b"??abc?????"
+ assert reader.tell() == 3
+ assert reader.readinto(memoryview(target)[5:]) == 3
+ assert target == b"??abcdef??"
+ assert reader.tell() == 6
+ assert reader.readinto(target) == 0
+ assert target == b"??abcdef??"
+ finally:
+ operator.delete(filename)
diff --git a/bindings/python/tests/test_sync_gil.py b/bindings/python/tests/test_sync_gil.py
index 8b1d2482c8..58fede10b9 100644
--- a/bindings/python/tests/test_sync_gil.py
+++ b/bindings/python/tests/test_sync_gil.py
@@ -150,3 +150,133 @@ def test_blocking_operator_releases_gil(service_name, operator) -> None:
server_process.terminate()
server_process.join(TEST_TIMEOUT_SECONDS)
port_receiver.close()
+
+
+class _FileHandler(BaseHTTPRequestHandler):
+ def _respond(self, body=b"file contents\n") -> None:
+ self.server.request_started.set()
+ if not self.server.release_response.wait(SERVER_FALLBACK_SECONDS):
+ self.server.response_timed_out.set()
+ self.send_response(403 if self.path == "/denied" else 200)
+ self.send_header("Content-Length", str(len(body)))
+ self.end_headers()
+ if self.command == "GET":
+ self.wfile.write(body)
+
+ def do_GET(self) -> None:
+ self._respond()
+
+ def do_HEAD(self) -> None:
+ self._respond()
+
+ def do_PUT(self) -> None:
+ self.rfile.read(int(self.headers["Content-Length"]))
+ self._respond(b"")
+
+ def log_message(self, format_, *args: object) -> None:
+ pass
+
+
+def _serve_file_http(
+ port_sender, request_started, release_response, response_timed_out
+) -> None:
+ server = ThreadingHTTPServer(("127.0.0.1", 0), _FileHandler)
+ server.request_started = request_started
+ server.release_response = release_response
+ server.response_timed_out = response_timed_out
+ port_sender.send(server.server_address[1])
+ port_sender.close()
+ server.serve_forever(poll_interval=0.01)
+
+
+@pytest.mark.parametrize(
+ "operation",
+ ["read", "readline", "readinto", "seek", "close", "exit", "close_error"],
+)
+def test_blocking_file_releases_gil(operation, service_name):
+ if service_name != "memory":
+ pytest.skip("run the standalone GIL regression test once")
+
+ context = multiprocessing.get_context("spawn")
+ request_started = context.Event()
+ release_response = context.Event()
+ response_timed_out = context.Event()
+ receiver, sender = context.Pipe(duplex=False)
+ server = context.Process(
+ target=_serve_file_http,
+ args=(sender, request_started, release_response, response_timed_out),
+ )
+ server.start()
+ sender.close()
+ worker = None
+ results = []
+ errors = []
+ try:
+ assert receiver.poll(TEST_TIMEOUT_SECONDS)
+ endpoint = f"http://127.0.0.1:{receiver.recv()}"
+ if operation in {"close", "exit", "close_error"}:
+ op = opendal.Operator(
+ "webdav", endpoint=endpoint, disable_create_dir="true"
+ )
+ file = op.open("denied" if operation == "close_error" else "file", "wb")
+ file.write(b"file contents\n")
+ else:
+ op = opendal.Operator("http", endpoint=endpoint)
+ file = op.open("file", "rb")
+ target = bytearray(b"?" * 32)
+
+ def perform_io() -> None:
+ try:
+ if operation == "readinto":
+ results.append(file.readinto(target))
+ elif operation == "seek":
+ results.append(file.seek(0, 2))
+ elif operation == "close_error":
+ file.close()
+ elif operation == "exit":
+ results.append(file.__exit__(None, None, None))
+ else:
+ results.append(getattr(file, operation)())
+ except Exception as error: # noqa: BLE001
+ errors.append(error)
+
+ worker = threading.Thread(target=perform_io)
+ worker.start()
+ assert request_started.wait(TEST_TIMEOUT_SECONDS)
+ # This thread must run before the server unblocks the storage operation.
+ # The active mutable borrow must also reject a concurrent state change.
+ with pytest.raises(RuntimeError, match="borrow"):
+ file.close()
+ if operation == "readinto":
+ with pytest.raises(BufferError):
+ target.extend(b"resize")
+ target[:] = b"?" * len(target)
+ assert not response_timed_out.is_set()
+ release_response.set()
+ worker.join(TEST_TIMEOUT_SECONDS)
+ assert not worker.is_alive()
+ if operation == "close_error":
+ assert len(errors) == 1
+ assert isinstance(errors[0], opendal.exceptions.PermissionDenied)
+ assert not file.closed
+ return
+ assert errors == []
+ if operation in {"read", "readline"}:
+ assert results == [b"file contents\n"]
+ assert file.tell() == 14
+ elif operation == "readinto":
+ assert results == [14]
+ assert target == b"file contents\n" + b"?" * 18
+ elif operation == "seek":
+ assert results == [14]
+ else:
+ assert results == [None]
+ assert file.closed
+ file.close()
+ finally:
+ release_response.set()
+ if worker is not None:
+ worker.join(TEST_TIMEOUT_SECONDS)
+ server.terminate()
+ server.join(TEST_TIMEOUT_SECONDS)
+ receiver.close()