Break reference cycles and harden async chunk reader

   - FieldsAccessor: use weakref to parent NDArray and create NDField
     instances on access instead of eagerly. Breaks NDArray↔NDField
     reference cycles so gen-0/gen-1 GC can collect them, reducing
     thread accumulation on Python 3.14.

   - sync_read_chunks: wrap reader thread in try/finally with
     thread.join() to prevent thread leaks. Propagate exceptions
     from the async reader instead of silently swallowing them.
     Detect dead reader thread to avoid infinite wait on empty queue.

   - ThreadPoolExecutor: cap max_workers to min(len(arrs), nthreads)
     instead of os.cpu_count(), avoiding excess threads when there
     are few operands.

Author: FrancescAlted

src/blosc2/lazyexpr.py

@@ -1182,7 +1182,8 @@ def get_chunk(arr, info, nchunk):
 async def async_read_chunks(arrs, info, queue):
     loop = asyncio.get_event_loop()
     shape, chunks_ = arrs[0].shape, arrs[0].chunks
-    with concurrent.futures.ThreadPoolExecutor(max_workers=os.cpu_count() or 1) as executor:
+    max_workers = max(1, min(len(arrs), int(getattr(blosc2, "nthreads", 1) or 1)))
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
         my_chunk_iter = range(arrs[0].schunk.nchunks)
         if len(info) == 5:
             if info[-1] is not None:
@@ -1213,20 +1214,38 @@ def async_read_chunks_thread(arrs, info, queue):
 def sync_read_chunks(arrs, info):
     queue_size = 2  # maximum number of chunks in the queue
     queue = Queue(maxsize=queue_size)
+    worker_exc = None
+
+    def _run_async_reader():
+        nonlocal worker_exc
+        try:
+            async_read_chunks_thread(arrs, info, queue)
+        except BaseException as exc:
+            worker_exc = exc
+            queue.put(None)
 
     # Start the async file reading in a separate thread
-    thread = threading.Thread(target=async_read_chunks_thread, args=(arrs, info, queue))
+    thread = threading.Thread(target=_run_async_reader)
     thread.start()
 
-    # Read the chunks synchronously from the queue
-    while True:
-        try:
-            chunks = queue.get(timeout=1)  # Wait for the next chunk
-            if chunks is None:  # End of chunks
-                break
-            yield chunks
-        except Empty:
-            continue
+    try:
+        # Read the chunks synchronously from the queue
+        while True:
+            try:
+                chunks = queue.get(timeout=1)  # Wait for the next chunk
+                if chunks is None:  # End of chunks
+                    if worker_exc is not None:
+                        raise worker_exc
+                    break
+                yield chunks
+            except Empty:
+                if not thread.is_alive():
+                    if worker_exc is not None:
+                        raise worker_exc
+                    break
+                continue
+    finally:
+        thread.join()
 
 
 def read_nchunk(arrs, info):

src/blosc2/ndarray.py

@@ -10,6 +10,7 @@
 import builtins
 import inspect
 import math
+import weakref
 from abc import abstractmethod
 from collections import OrderedDict, namedtuple
 from collections.abc import Mapping
@@ -146,23 +147,33 @@ def __getitem__(self, key: Any) -> Any:
 class FieldsAccessor(Mapping):
     """Read-only mapping of structured field views."""
 
-    def __init__(self, field_views: dict[str, Any]):
-        self._field_views = field_views
+    def __init__(self, ndarr, field_names: Sequence[str]):
+        self._ndarr_ref = weakref.ref(ndarr)
+        self._field_names = tuple(field_names)
+
+    def _ndarr(self):
+        ndarr = self._ndarr_ref()
+        if ndarr is None:
+            raise ReferenceError("owning NDArray has been released")
+        return ndarr
 
     def __getitem__(self, key: str) -> Any:
-        return self._field_views[key]
+        if key not in self._field_names:
+            raise KeyError(key)
+        return NDField(self._ndarr(), key)
 
     def __iter__(self) -> Iterator[str]:
-        return iter(self._field_views)
+        return iter(self._field_names)
 
     def __len__(self) -> int:
-        return len(self._field_views)
+        return len(self._field_names)
 
     def __setitem__(self, key: str, value: object) -> None:
         raise TypeError(f'assign through the field view, e.g. array.fields["{key}"][:] = values')
 
     def copy(self) -> dict[str, Any]:
-        return dict(self._field_views)
+        ndarr = self._ndarr()
+        return {field: NDField(ndarr, field) for field in self._field_names}
 
     def __or__(self, other: object) -> dict[str, Any]:
         if not isinstance(other, Mapping):
@@ -175,7 +186,7 @@ def __ror__(self, other: object) -> dict[str, Any]:
         return dict(other) | self.copy()
 
     def __repr__(self) -> str:
-        return repr(self._field_views)
+        return repr(dict(self))
 
 
 def is_documented_by(original):
@@ -3730,11 +3741,8 @@ def __init__(self, **kwargs):
         base = kwargs.pop("_base", None)
         super().__init__(kwargs["_array"], base=base)
         # Accessor to fields
-        field_views = {}
-        if self.dtype.fields:
-            for field in self.dtype.fields:
-                field_views[field] = NDField(self, field)
-        self._fields = FieldsAccessor(field_views)
+        field_names = tuple(self.dtype.fields) if self.dtype.fields else ()
+        self._fields = FieldsAccessor(self, field_names)
 
     @property
     def cparams(self) -> blosc2.CParams: