[WIP] Added max_block and request_batch to pyfive.File and backend

doc/optimising.rst

@@ -86,6 +86,28 @@ For example, you can use the `concurrent.futures` module to read data from multi
 
 You can do the same thing to parallelise manipulations within the variables, by for example using, ``dask``, but that is beyond the scope of this document.
 
+In addition, you can use the attributes ``max_request_block`` and ``batch_request_size`` to customise how requests are made to remote servers in the pyfive backend.
+
+.. code-block:: python
+
+    import pyfive
+
+    var_name = "var1"
+    max_request_block = 10e6 # 10 MB
+    batch_request_size = 80
+
+    with pyfive.File("data.h5", "r", 
+            max_request_block=max_request_block, 
+            batch_request_size=batch_request_size) as f:
+
+        dset = f[var_name]
+        data = dset[...]  # Read the entire variable
+        print('Min: ',data.min())
+
+The ``batch_request_size`` specifies the number of individual requests that can be made asynchronously in parallel by the backend session. Some remote services providing data access have limits applied to the rate of requests, such that if you attempt to make more than a certain number of requests simultaneously (or if you have more than that number of open/pending requests) you may be denied further requests with a ``HTTP 429: Too Many Requests`` error. If you encounter this problem, it may be effective to simply limit the requests your own client can make concurrently.
+
+In addition to the above limit, it may be more efficient for you to make fewer requests in general as each request will require some time for handling on the server end. If you use the ``max_request_block`` parameter, the backend part of the pyfive client will attempt to combine requests that represent a continuous block of memory, so instead of making many requests you can just make one larger request and reduce the overheads. Range-requests will be combined up to the limit you set here (10 Megabytes in the example above). For maximum efficiency, consider the bandwidth available to you, in relation to the maximum request size and the number of concurrent requests. The above example of 80 requests at 10MB per request, when combined with a typical non-fibre bandwidth of 30 Mbps would mean the whole batch would take 200 seconds which may be more than the timeout limit for the session, so you may want to reduce the number of requests or the block size for your situation.
+
 
 Using pyfive with S3
 --------------------

pyfive/h5d.py

@@ -1,4 +1,5 @@
 import numpy as np
+import fsspec.utils as futils
 from collections import namedtuple
 from operator import mul
 from pyfive.indexing import OrthogonalIndexer, ZarrArrayStub
@@ -111,7 +112,14 @@ def _decode_chunk(self, chunk_buffer, filter_mask, dtype):
             self.chunks, order=self._order
         )
 
-    def _select_chunks(self, indexer, out, dtype):
+    def _select_chunks(
+        self,
+        indexer,
+        out,
+        dtype,
+        max_block: int | None = None,
+        batch_size: int | None = None,
+    ):
         """
         Collect required chunks and dispatch I/O to the best strategy.
         Called by ``_get_selection_via_chunks`` in place of the serial loop.
@@ -121,18 +129,20 @@ def _select_chunks(self, indexer, out, dtype):
             return
 
         # Case A: fsspec - bulk parallel fetch via cat_ranges
-        if not self.posix and self._cat_range_allowed:
-            fh = self._fh
+        if not self.posix and self._cat_range_allowed:  # type: ignore[attr-defined]
+            fh = self._fh  # type: ignore[attr-defined]
             actual_fh = getattr(fh, "fh", fh)  # support wrapped file-like objects
             if hasattr(actual_fh, "fs") and hasattr(actual_fh.fs, "cat_ranges"):
                 logger.info(
                     f"[pyfive] chunk read strategy: fsspec_cat_ranges ({len(chunks)} chunks)"
                 )
-                self._read_bulk_fsspec(fh, chunks, out, dtype)
+                self._read_bulk_fsspec(
+                    fh, chunks, out, dtype, max_block=max_block, batch_size=batch_size
+                )
                 return
 
         # Case B: POSIX - thread-parallel reads via os.pread
-        if self.posix and hasattr(os, "pread") and self._thread_count != 0:
+        if self.posix and hasattr(os, "pread") and self._thread_count != 0:  # type: ignore[attr-defined]
             logger.info(
                 "[pyfive] chunk read strategy: posix_pread_threads workers=%s (%d chunks)",
                 self._thread_count,
@@ -197,7 +207,15 @@ def _read_one(item):
                 chunk_sel
             ]
 
-    def _read_bulk_fsspec(self, fh, chunks, out, dtype):
+    def _read_bulk_fsspec(
+        self,
+        fh,
+        chunks,
+        out,
+        dtype,
+        max_block: int | None = None,
+        batch_size: int | None = None,
+    ):
         """
         Bulk read via ``fsspec`` ``cat_ranges``.
 
@@ -207,10 +225,16 @@ def _read_bulk_fsspec(self, fh, chunks, out, dtype):
         (reaching through the MetadataBufferingWrapper).
         """
         actual_fh = getattr(fh, "fh", fh)  # support wrapped file-like objects
-        path = actual_fh.path
         starts = [si.byte_offset for _, _, _, si in chunks]
         stops = [si.byte_offset + si.size for _, _, _, si in chunks]
-        buffers = actual_fh.fs.cat_ranges([path] * len(chunks), starts, stops)
+
+        paths = [actual_fh.path] * len(chunks)
+        if max_block is not None:
+            paths, starts, stops = futils.merge_offset_ranges(
+                paths, starts, stops, max_block=max_block
+            )
+
+        buffers = actual_fh.fs.cat_ranges(paths, starts, stops, batch_size=batch_size)
 
         for (_coords, chunk_sel, out_sel, storeinfo), chunk_buffer in zip(
             chunks, buffers
@@ -240,6 +264,8 @@ def __init__(
         dataobject: "DataObjects",  # type: ignore[name-defined]  # noqa: F821
         noindex: bool = False,
         pseudo_chunking_size_MB: int = 4,
+        max_request_block: int | None = None,
+        batch_request_size: int | None = None,
     ) -> None:
         """
         Instantiated with the ``pyfive`` ``datasetdataobject``, we copy and cache everything
@@ -263,6 +289,9 @@ def __init__(
 
         """
 
+        self._max_block = max_request_block
+        self._batch_size = batch_request_size
+
         self._order = dataobject.order
         fh = dataobject.fh
 
@@ -929,7 +958,13 @@ def _get_selection_via_chunks(self, args):
                 fh.close()
 
         else:
-            self._select_chunks(indexer, out, dtype)
+            self._select_chunks(
+                indexer,
+                out,
+                dtype,
+                max_block=self._max_block,
+                batch_size=self._batch_size,
+            )
 
         if isinstance(self._ptype, P5ReferenceType):
             to_reference = np.vectorize(Reference)

pyfive/high_level.py

@@ -40,7 +40,14 @@ class Group(Mapping):
 
     """
 
-    def __init__(self, name: str, dataobjects: DataObjects, parent: "Group") -> None:
+    def __init__(
+        self,
+        name: str,
+        dataobjects: DataObjects,
+        parent: "Group",
+        max_request_block: int | None = None,
+        batch_request_size: int | None = None,
+    ) -> None:
         """initalize."""
 
         self.parent = parent
@@ -51,6 +58,9 @@ def __init__(self, name: str, dataobjects: DataObjects, parent: "Group") -> None
         self._dataobjects = dataobjects
         self._attrs = None  # cached property
 
+        self._max_request_block = max_request_block
+        self._batch_request_size = batch_request_size
+
     def __repr__(self):
         return '<HDF5 group "%s" (%d members)>' % (self.name, len(self))
 
@@ -137,7 +147,16 @@ def __getitem_lazy_control(self, y, noindex):
         if dataobjs.is_dataset:
             if additional_obj != ".":
                 raise KeyError("%s is a dataset, not a group" % (obj_name))
-            return Dataset(obj_name, DatasetID(dataobjs, noindex=noindex), self)
+            return Dataset(
+                obj_name,
+                DatasetID(
+                    dataobjs,
+                    noindex=noindex,
+                    max_request_block=self._max_request_block,
+                    batch_request_size=self._batch_request_size,
+                ),
+                self,
+            )
 
         try:
             # if true, this may well raise a NotImplementedError, if so, we need
@@ -260,8 +279,10 @@ def __init__(
         filename: str | BinaryIO | MetadataBufferingWrapper,
         mode: str = "r",
         metadata_buffer_size: int = 1,
+        **kwargs,
     ) -> None:
         """initalize."""
+
         if mode != "r":
             raise NotImplementedError(
                 "pyfive only provides support for reading and treats all reads as binary"
@@ -306,7 +327,7 @@ def __init__(
         self.file = self
         self.mode = "r"
         self.userblock_size = 0
-        super(File, self).__init__("/", dataobjects, self)
+        super(File, self).__init__("/", dataobjects, self, **kwargs)
 
     @property
     def consolidated_metadata(self) -> bool:

tests/test_btree_parallel.py

@@ -4,6 +4,7 @@
 
 import fsspec
 import pytest
+import numpy as np
 from numpy.testing import assert_array_equal
 
 import pyfive
@@ -269,6 +270,32 @@ def cat_ranges_spy(paths, starts, stops, *args, **kwargs):
     assert_array_equal(fsspec_data, serial_data)
     assert fsspec_chunk_info == serial_chunk_info
 
+    standard_calls = list(calls)
+    calls = []
+
+    client_kwargs = {"auth": None, "ssl": False}
+    fs = fsspec.filesystem("http", **client_kwargs)
+
+    original_cat_ranges = fs.cat_ranges
+    fs.cat_ranges = cat_ranges_spy
+
+    uri = "https://esgf.ceda.ac.uk/thredds/fileServer/esg_cmip6/CMIP6/AerChemMIP/MOHC/UKESM1-0-LL/ssp370SST-lowNTCF/r1i1p1f2/Amon/cl/gn/latest/cl_Amon_UKESM1-0-LL_ssp370SST-lowNTCF_r1i1p1f2_gn_205001-209912.nc"
+
+    with fs.open(uri, "rb") as fh:
+        with pyfive.File(fh, max_request_block=10e6, batch_request_size=100) as hfile:
+            ds = np.array(hfile["cl"][:100])
+            assert ds.shape == (100, 85, 144, 192)
+
+    assert len(calls[0][0]) > 0, (
+        "Expected fsspec cat_ranges to be used for leaf-node reads"
+    )
+    assert len(standard_calls[0][0]) > 0, (
+        "Expected fsspec cat_ranges to be used for leaf-node reads"
+    )
+    assert len(calls[0][0]) != len(standard_calls[0][0]), (
+        "Expected differing cat_range spans"
+    )
+
 
 def test_parallel_s3fs_cat_ranges_matches_serial_results(s3fs_s3):
     with pyfive.File(DATASET_CHUNKED_HDF5_FILE) as hfile:

tests/test_h5d.py

@@ -9,6 +9,8 @@
 
 from conftest import s3_url_exists
 
+DIRNAME = os.path.dirname(__file__)
+DATASET_CHUNKED_HDF5_FILE = os.path.join(DIRNAME, "data", "chunked.hdf5")
 
 mypath = Path(__file__).parent
 filename = mypath / "data" / "compressed.hdf5"