Bind the pipeline the batch source stands on

AGENTS.md

@@ -14,7 +14,7 @@ the same pull request that causes it.
 
 | Path | Holds |
 |---|---|
-| `src/` | The pybind11 binding, 42 `.cpp` and 42 `.hpp`, compiled into the `rexlib._binding` extension |
+| `src/` | The pybind11 binding, 47 `.cpp` and 47 `.hpp`, compiled into the `rexlib._binding` extension |
 | `python/rexlib/` | The Python package, 11 `.py`: everything the binding cannot express |
 | `tests/` | pytest suites, mirroring the binding's module structure |
 | `tests/assets/` | Two dummy plugins, built by CMake, that the plugin tests discover |
@@ -69,8 +69,9 @@ name that names them. See #143.
 | `_binding.ndarray` | `src/core/ndarray/` | `Array`, `ArrayDescriptor` |
 | `_binding.hardware` | `src/core/hardware/` | Devices, sessions, queues, events, memory resources |
 | `_binding.dispatch` | `src/core/dispatch/` | `ExecutionContext`, `Dispatcher`, `ProgramManager` |
+| `_binding.concurrency` | `src/core/concurrency/` | `Executor` and its two kinds, `Completion` |
 | `_binding.functional` | `src/functional/` | The operations, each taking an explicit context |
-| `_binding.em.image` | `src/em/image/` | `ImageLocation`, the read and write format managers, `read` and `write` |
+| `_binding.em.image` | `src/em/image/` | `ImageLocation`, the format managers and reader providers, the sources, `read`, `write` and the shape queries |
 
 A submodule is created by the `main.cpp` above it, which then hands it to the
 `bind_` function of the directory it stands for: `src/main.cpp` creates `em`
@@ -246,6 +247,19 @@ Where it does not, `device_index` being the one left, the hash has to be
 written, which is rexlib's to do rather than this binding's;
 `_session_pool.py` keys on a `(backend, id)` tuple in the meantime.
 
+A binding that blocks releases the GIL with
+`py::call_guard<py::gil_scoped_release>()`. `Completion.wait`, `Completion.get`
+and `ImageBatchSource.read` all do: held through a read, the GIL would freeze
+every other Python thread for its duration and there would be nothing
+asynchronous left about the interface. pybind11 converts the arguments before
+the guard is constructed and the return value after it is destroyed, so
+nothing touches Python without it.
+
+`rexlib::array` is move-only, and anything taking one by value has to be given
+`destination.share()` rather than the caller's own. Bound plainly, pybind11
+would move the array out of the Python object and hand the caller back an
+empty one.
+
 A type rexlib parses from a string carries a `from_string` static method and
 no constructor that parses. `DeviceIndex` and `ImageLocation` both do, and
 `ImageLocation` is why: its constructor already gives a lone string the

BACKLOG.md

@@ -56,55 +56,12 @@ One trap before `compare` lands: a type that defines `__eq__` and no
 rexlib, so the comparison operators need an answer about identity before they
 go in.
 
-## 3. Bind the batch source
-
-`image_batch_source` is the class meant for pipelines: asynchronous, batched,
-running its own threads, and where the performance is. The slow path
-(`em.image.read`) stays what it is, for one-off reads and tests.
-
-The chain:
-
-```
-ImageReadFormatManager                     ← bound
-  └→ DirectImageReaderProvider(formats)    │ CachingImageReaderProvider(backing, capacity)
-     └→ ImageSource(readers, executor)     ← ThreadPoolExecutor(n) │ SynchronousExecutor()
-        └→ ImageBatchSource(source)
-           └→ read(destination, locations) → Completion{wait, is_ready, get}
-```
-
-It splits in two, by risk rather than by size.
-
-**3a — geometry.** Bind `query_extents` and `query_core_extents` over a format
-manager. Read-only surface, no threads, no GIL, and useful for sizing any read
-rather than only a batch. Their overloads over a reader provider wait for 3b,
-which is what binds the provider; those take a non-const reference where the
-format manager ones take a const one.
-
-`query_core_extents` returns the shape of a single image or volume, which is
-what a batch destination carries beside its leading extent, so no arithmetic
-has to be reproduced here. Whether a `make_batch_descriptor` helper on top is
-worth its weight is decided when this is written, with the code in view; it
-would only save a caller from getting the leading extent wrong.
-
-**3b — the pipeline.** Executors, providers, source, batch source, completion,
-the Python layer, and the GIL work.
-
-Settled before starting, recorded in [DECISIONS.md](DECISIONS.md): the
-destination's data type is the caller's choice rather than the file's; the
-thread pool is per instance with an executor accepted as an override; the
-completion is mirrored rather than wrapped in a future; and `wait`, `get` and
-`read` release the GIL.
-
-Two facts to build against, both verified in rexlib's sources:
-
-- `rexlib::array` is move-only and `read` takes it by value. Bound plainly,
-  pybind11 would move the array out of the caller's Python object and leave
-  them holding an empty one. The binding takes `array&` and passes
-  `destination.share()`.
-- Nothing else needs keeping alive. The `locations` span is consumed
-  synchronously while the plan is built; the destination is a handle whose
-  storage outlives a dropped Python reference; and `image_source::read`
-  extracts what each task needs at submit time, so its plan does not dangle.
+## 3. Bind the patch source
+
+`image_patch_source` arrived with the batch source upstream and crops batches
+of patches out of an image, clipped at the borders. The batch source is bound;
+this one is not, and it stands on the same chain, so it is a smaller piece of
+work than the first was.
 
 ## 4. Translate the library's exceptions
 

python/rexlib/__init__.py

@@ -25,6 +25,7 @@
 from . import _paths as _paths
 
 from ._binding import (
+	concurrency as concurrency,
 	dispatch as dispatch,
 	hardware as hardware,
 	Plugin as Plugin,
@@ -91,6 +92,7 @@
 	"add",
 	"cast",
 	"cast_copy",
+	"concurrency",
 	"copy",
 	"device",
 	"dispatch",

python/rexlib/em/image/__init__.py

@@ -21,23 +21,35 @@
 from __future__ import annotations
 
 from ..._binding.em.image import (
+	CachingImageReaderProvider as CachingImageReaderProvider,
+	DirectImageReaderProvider as DirectImageReaderProvider,
+	ImageBatchSource as ImageBatchSource,
 	ImageLocation as ImageLocation,
+	ImageReaderProvider as ImageReaderProvider,
+	ImageSource as ImageSource,
 	ImageReadFormatManager as ImageReadFormatManager,
 	ImageWriteFormatManager as ImageWriteFormatManager,
 	get_image_read_format_manager as get_image_read_format_manager,
 	get_image_write_format_manager as get_image_write_format_manager,
 )
 from ._functions import (
+	batch_source as batch_source,
 	query_core_extents as query_core_extents,
 	query_extents as query_extents,
 	read as read,
 	write as write,
 )
 
 __all__ = [
+	"CachingImageReaderProvider",
+	"DirectImageReaderProvider",
+	"ImageBatchSource",
 	"ImageLocation",
 	"ImageReadFormatManager",
+	"ImageReaderProvider",
+	"ImageSource",
 	"ImageWriteFormatManager",
+	"batch_source",
 	"get_image_read_format_manager",
 	"get_image_write_format_manager",
 	"query_core_extents",

python/rexlib/em/image/_functions.py

@@ -10,10 +10,17 @@
 
 from __future__ import annotations
 
+import os
+
+from ..._binding.concurrency import Executor, ThreadPoolExecutor
 from ..._binding.dispatch import ExecutionContext
 from ..._binding.em import image as _raw
 from ..._binding.em.image import (
+	CachingImageReaderProvider,
+	DirectImageReaderProvider,
+	ImageBatchSource,
 	ImageLocation,
+	ImageSource,
 	ImageReadFormatManager,
 	ImageWriteFormatManager,
 	get_image_read_format_manager,
@@ -24,6 +31,9 @@
 from ..._catalog import get_default_catalog
 from ..._functional import _resolve_context
 
+def _default_worker_count() -> int:
+	return os.cpu_count() or 1
+
 def _resolve_read_manager(
 	manager: ImageReadFormatManager | None
 ) -> ImageReadFormatManager:
@@ -38,6 +48,45 @@ def _resolve_write_manager(
 		manager = get_image_write_format_manager(get_default_catalog())
 	return manager
 
+def batch_source(
+	workers: int | None = None,
+	cache: int | None = None,
+	manager: ImageReadFormatManager | None = None,
+	executor: Executor | None = None
+) -> ImageBatchSource:
+	"""
+	Assemble a batch source over a thread pool.
+
+	Wires the four objects a batch source stands on — the formats, a
+	reader provider, an executor and the source itself — so that a
+	caller reaches the pipeline without naming them.
+
+	Each source built this way owns its executor. Two of them means two
+	thread pools, each sized to the machine, which is worth avoiding by
+	passing one executor to both.
+
+	Args:
+		workers: How many threads to run reads on. Defaults to what the
+			machine reports. Ignored when `executor` is given.
+		cache: How many open readers to keep between reads. Defaults to
+			opening a file every time it is asked for, which is what
+			suits reads that do not revisit a file.
+		manager: The formats to recognize files with. Defaults to the
+			ones the default catalog holds.
+		executor: Where reads run. Defaults to a thread pool of its own.
+
+	Returns:
+		ImageBatchSource: The assembled source.
+	"""
+	readers = DirectImageReaderProvider(_resolve_read_manager(manager))
+	if cache is not None:
+		readers = CachingImageReaderProvider(readers, cache)
+	if executor is None:
+		executor = ThreadPoolExecutor(
+			workers if workers is not None else _default_worker_count()
+		)
+	return ImageBatchSource(ImageSource(readers, executor))
+
 def query_extents(
 	path: str,
 	manager: ImageReadFormatManager | None = None

src/core/concurrency/completion.cpp

@@ -0,0 +1,29 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "completion.hpp"
+
+namespace rexlib
+{
+
+namespace py = pybind11;
+
+completion_class declare_completion(pybind11::module_ &m)
+{
+	return completion_class(m, "Completion");
+}
+
+void define_completion(completion_class &c)
+{
+	c
+		.def(
+			"wait", &completion::wait,
+			py::call_guard<py::gil_scoped_release>()
+		)
+		.def(
+			"get", &completion::get,
+			py::call_guard<py::gil_scoped_release>()
+		)
+		.def_property_readonly("is_ready", &completion::is_ready);
+}
+
+} // namespace rexlib

src/core/concurrency/completion.hpp

@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <pybind11/pybind11.h>
+
+#include <rexlib/core/concurrency/completion.hpp>
+
+#include <memory>
+
+namespace rexlib
+{
+
+using completion_class =
+	pybind11::class_<completion, std::shared_ptr<completion>>;
+
+completion_class declare_completion(pybind11::module_ &m);
+void define_completion(completion_class &c);
+
+} // namespace rexlib

src/core/concurrency/executor.cpp

@@ -0,0 +1,37 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "executor.hpp"
+
+#include <cstddef>
+
+namespace rexlib
+{
+
+namespace py = pybind11;
+
+executor_class declare_executor(pybind11::module_ &m)
+{
+	return executor_class(m, "Executor");
+}
+
+synchronous_executor_class declare_synchronous_executor(pybind11::module_ &m)
+{
+	return synchronous_executor_class(m, "SynchronousExecutor");
+}
+
+thread_pool_executor_class declare_thread_pool_executor(pybind11::module_ &m)
+{
+	return thread_pool_executor_class(m, "ThreadPoolExecutor");
+}
+
+void define_synchronous_executor(synchronous_executor_class &c)
+{
+	c.def(py::init<>());
+}
+
+void define_thread_pool_executor(thread_pool_executor_class &c)
+{
+	c.def(py::init<std::size_t>(), py::arg("worker_count"));
+}
+
+} // namespace rexlib

src/core/concurrency/executor.hpp

@@ -0,0 +1,30 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <pybind11/pybind11.h>
+
+#include <rexlib/core/concurrency/executor.hpp>
+#include <rexlib/core/concurrency/synchronous_executor.hpp>
+#include <rexlib/core/concurrency/thread_pool_executor.hpp>
+
+#include <memory>
+
+namespace rexlib
+{
+
+using executor_class = pybind11::class_<executor, std::shared_ptr<executor>>;
+using synchronous_executor_class = pybind11::class_<
+	synchronous_executor, executor, std::shared_ptr<synchronous_executor>
+>;
+using thread_pool_executor_class = pybind11::class_<
+	thread_pool_executor, executor, std::shared_ptr<thread_pool_executor>
+>;
+
+executor_class declare_executor(pybind11::module_ &m);
+synchronous_executor_class declare_synchronous_executor(pybind11::module_ &m);
+thread_pool_executor_class declare_thread_pool_executor(pybind11::module_ &m);
+void define_synchronous_executor(synchronous_executor_class &c);
+void define_thread_pool_executor(thread_pool_executor_class &c);
+
+} // namespace rexlib

src/core/concurrency/main.cpp

@@ -0,0 +1,23 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "main.hpp"
+
+#include "completion.hpp"
+#include "executor.hpp"
+
+namespace rexlib
+{
+
+void bind_concurrency(pybind11::module_ &m)
+{
+	auto completion = declare_completion(m);
+	declare_executor(m);
+	auto synchronous_executor = declare_synchronous_executor(m);
+	auto thread_pool_executor = declare_thread_pool_executor(m);
+
+	define_completion(completion);
+	define_synchronous_executor(synchronous_executor);
+	define_thread_pool_executor(thread_pool_executor);
+}
+
+} // namespace rexlib

src/core/concurrency/main.hpp

@@ -0,0 +1,12 @@
+// SPDX-License-Identifier: GPL-3.0-only
+
+#pragma once
+
+#include <pybind11/pybind11.h>
+
+namespace rexlib
+{
+
+void bind_concurrency(pybind11::module_ &m);
+
+} // namespace rexlib