refactor: centralize async runtime around shared loop and pools

Author: hrodmn

ARCHITECTURE.md

@@ -97,6 +97,20 @@ event loop. Multiple concurrent chunk reads overlap naturally, so the
 async path can be faster than the synchronous `da.compute()` when
 reading many chunks inside an already-running loop.
 
+### Concurrency notes
+
+- Sync callers submit work to one shared persistent lazycogs event loop.
+- CPU-bound reprojection runs on one bounded shared thread pool. Set
+  `LAZYCOGS_REPROJECT_WORKERS` before first use to change the default
+  `min(os.cpu_count(), 4)` limit.
+- DuckDB queries yield the event loop by running on a small explicit
+  executor instead of on the loop thread. On the local benchmark fixture,
+  DuckDB stayed under 2% of per-date chunk wall time, so there is no
+  separate per-thread DuckDB client pool today.
+- If you need to construct a loop-bound resource for lazycogs internals,
+  use `lazycogs.run_on_loop(...)`.
+- Low-level callers should use `await lazycogs.read_chunk_async(...)`.
+
 ## Documentation
 
 - [Home](https://developmentseed.github.io/lazycogs/) — quickstart and full usage guide

README.md

@@ -4,7 +4,7 @@
 
 ::: lazycogs.store_for
 
-::: lazycogs.set_reproject_workers
+::: lazycogs.run_on_loop
 
 ::: lazycogs.ExplainPlan
 

docs/api/utils.md

@@ -19,7 +19,7 @@ vals = da.sel(x=299965, y=2653947, method="nearest").sel(time=slice("2025-06", "
 
 ## When to add chunks
 
-Add `chunks={"time": 1}` when you want dask to distribute work across multiple workers. Without chunks, all time steps share one event loop and one reprojection thread pool. I/O is concurrent but CPU-bound reprojection is serialized within that pool. With temporal chunks, each time step becomes an independent dask task with its own event loop and thread pool, so reprojection work can run in true parallel across workers:
+Add `chunks={"time": 1}` when you want dask to distribute work across multiple workers. Without chunks, all time steps share one lazycogs event loop and one bounded reprojection pool. I/O is concurrent but CPU-bound reprojection is bounded by that shared pool. With temporal chunks, dask can run multiple chunk tasks in parallel across worker threads, all submitting to the same lazycogs loop while still sharing the same bounded reprojection pool:
 
 ```python
 da = lazycogs.open(
@@ -34,7 +34,7 @@ da.max(dim="time").compute()  # each time step runs in its own dask task
 
 ## Spatial chunks
 
-Avoid spatial chunks unless you are under memory pressure. lazycogs handles spatial I/O concurrency internally through its async event loop — adding spatial dask tasks layers DuckDB query overhead on top of I/O that was already happening concurrently.
+Avoid spatial chunks unless you are under memory pressure. lazycogs handles spatial I/O concurrency internally through its async event loop — adding spatial dask tasks layers extra DuckDB query overhead on top of I/O that was already happening concurrently.
 
 The one case where spatial chunks are useful is when a single time step is too large to fit in memory even at `max_concurrent_reads=1`. In that case, small spatial chunks limit how many pixels are in flight at once.
 
@@ -57,17 +57,16 @@ da = lazycogs.open(
 )
 ```
 
-## `set_reproject_workers`
+## `LAZYCOGS_REPROJECT_WORKERS`
 
-Controls how many threads each chunk's event loop uses for CPU-bound reprojection (pyproj + numpy). The default is `min(os.cpu_count(), 4)`.
+Controls how many threads the shared reprojection pool uses for CPU-bound reprojection (pyproj + numpy). The default is `min(os.cpu_count(), 4)`.
 
 Reprojection is memory-bandwidth-bound rather than compute-bound. Benchmarks show diminishing returns above 4 threads because concurrent large-array operations saturate the memory bus rather than adding throughput. Raising this beyond 4 is rarely useful.
 
-Each chunk gets its own independent thread pool, so dask tasks do not queue behind each other for reprojection.
+Set the environment variable before the first lazycogs chunk read:
 
-```python
-import lazycogs
-lazycogs.set_reproject_workers(2)   # reduce from default if memory-constrained
+```bash
+export LAZYCOGS_REPROJECT_WORKERS=2
 ```
 
 See also: [API reference for open()](../api/open.md), [API reference for utilities](../api/utils.md), [Architecture](../architecture.md)

docs/guides/chunking.md

@@ -1,8 +1,8 @@
 """lazycogs: lazy xarray DataArrays from STAC COG collections."""
 
-from lazycogs._chunk_reader import read_chunk, read_chunk_async
+from lazycogs._chunk_reader import read_chunk_async
 from lazycogs._core import open  # noqa: A004
-from lazycogs._executor import set_reproject_workers
+from lazycogs._executor import run_on_loop
 from lazycogs._explain import (  # noqa: F401 — registers da.lazycogs accessor
     ChunkRead,
     CogRead,
@@ -36,8 +36,7 @@
     "StdevMethod",
     "align_bbox",
     "open",
-    "read_chunk",
     "read_chunk_async",
-    "set_reproject_workers",
+    "run_on_loop",
     "store_for",
 ]

src/lazycogs/__init__.py

@@ -16,10 +16,7 @@
 
 from lazycogs._chunk_reader import read_chunk_async
 from lazycogs._cql2 import _extract_filter_fields, _sortby_fields
-from lazycogs._executor import (
-    _DUCKDB_EXECUTOR,
-    _run_coroutine,
-)
+from lazycogs._executor import _run_coroutine, get_duckdb_pool
 
 logger = logging.getLogger(__name__)
 
@@ -223,7 +220,7 @@ async def _search_items_async(
 
     """
     loop = asyncio.get_running_loop()
-    return await loop.run_in_executor(_DUCKDB_EXECUTOR, _search_items_sync, plan, date)
+    return await loop.run_in_executor(get_duckdb_pool(), _search_items_sync, plan, date)
 
 
 async def _run_one_date(

src/lazycogs/_backend.py

@@ -12,7 +12,7 @@
 from async_geotiff import GeoTIFF, Overview, RasterArray, Window
 from numpy import ma
 
-from lazycogs._executor import _run_coroutine
+from lazycogs._executor import get_reproject_pool
 from lazycogs._mosaic_methods import FirstMethod, MosaicMethodBase
 from lazycogs._reproject import (
     WarpMap,
@@ -442,7 +442,7 @@ async def _read_band(
     # Compute warp maps and apply, sharing maps across bands with identical geometry.
     loop = asyncio.get_running_loop()
     return await loop.run_in_executor(
-        None,
+        get_reproject_pool(),
         lambda: _apply_bands_with_warp_cache(
             band_rasters,
             ctx.chunk_affine,
@@ -615,62 +615,3 @@ def _error(idx: int, exc: BaseException) -> None:
                 dtype=np.float32,
             )
     return output
-
-
-def read_chunk(
-    items: list[dict],
-    bands: list[str],
-    chunk_affine: Affine,
-    dst_crs: CRS,
-    chunk_width: int,
-    chunk_height: int,
-    nodata: float | None = None,
-    mosaic_method_cls: type[MosaicMethodBase] | None = None,
-    store: ObjectStore | None = None,
-    max_concurrent_reads: int = 32,
-    warp_cache: dict | None = None,
-    path_fn: Callable[[str], str] | None = None,
-) -> dict[str, np.ndarray]:
-    """Run :func:`read_chunk_async` on the persistent per-thread background loop.
-
-    All arguments are identical to :func:`read_chunk_async`.
-
-    Args:
-        items: List of STAC item dicts to mosaic.  Processed in order.
-        bands: Asset keys identifying the bands to read from each item.
-        chunk_affine: Affine transform of the destination chunk.
-        dst_crs: CRS of the destination chunk.
-        chunk_width: Width of the destination chunk in pixels.
-        chunk_height: Height of the destination chunk in pixels.
-        nodata: No-data fill value.
-        mosaic_method_cls: Mosaic method class instantiated once per band.
-            Defaults to :class:`~lazycogs._mosaic_methods.FirstMethod`.
-        store: Optional pre-configured obstore ``ObjectStore`` instance.
-        max_concurrent_reads: Maximum number of COG reads to run concurrently.
-        warp_cache: Optional cache shared across calls for reusing warp maps
-            from earlier time steps.
-        path_fn: Optional callable that takes an asset HREF and returns the
-            object path to use with *store*.
-
-    Returns:
-        ``dict`` mapping each band name to an array of shape
-        ``(cog_bands, chunk_height, chunk_width)`` with dtype matching the
-        source COGs.
-
-    """
-    return _run_coroutine(
-        read_chunk_async(
-            items=items,
-            bands=bands,
-            chunk_affine=chunk_affine,
-            dst_crs=dst_crs,
-            chunk_width=chunk_width,
-            chunk_height=chunk_height,
-            nodata=nodata,
-            mosaic_method_cls=mosaic_method_cls,
-            store=store,
-            max_concurrent_reads=max_concurrent_reads,
-            warp_cache=warp_cache,
-            path_fn=path_fn,
-        ),
-    )