feat: fsspec source non-blocking chunks

src/uproot/source/fsspec.py

@@ -97,17 +97,21 @@ def chunks(self, ranges, notifications):
         self._num_requests += 1
         self._num_requested_chunks += len(ranges)
         self._num_requested_bytes += sum(stop - start for start, stop in ranges)
-        data = self._fs.cat_ranges(
-            [self._file_path] * len(ranges),
-            [start for start, _ in ranges],
-            [stop for _, stop in ranges],
-        )
+
+        executor = uproot.source.futures.ResourceThreadPoolExecutor(ranges)
+
         chunks = []
-        for item, (start, stop) in zip(data, ranges):
-            future = uproot.source.futures.TrivialFuture(item)
+        for start, stop in ranges:
+
+            def task(_range: tuple = (start, stop)):
+                return self._fs.cat_file(self._file_path, _range[0], _range[1])
+
+            future = uproot.source.futures.ResourceFuture(task)
             chunk = uproot.source.chunk.Chunk(self, start, stop, future)
-            uproot.source.chunk.notifier(chunk, notifications)()
+            future._set_notify(uproot.source.chunk.notifier(chunk, notifications))
+            executor.submit(future)
             chunks.append(chunk)
+
         return chunks
 
     @property

tests/test_0692_fsspec.py

@@ -1,11 +1,11 @@
 # BSD 3-Clause License; see https://github.com/scikit-hep/uproot4/blob/main/LICENSE
 
 import pytest
-
 import uproot
 import uproot.source.fsspec
 
 import skhep_testdata
+import queue
 
 
 @pytest.mark.network
@@ -66,3 +66,22 @@ def test_open_fsspec_xrootd():
         data = f["Events/run"].array(library="np", entry_stop=20)
         assert len(data) == 20
         assert (data == 194778).all()
+
+
+@pytest.mark.network
+def test_fsspec_chunks():
+    # Use the local HTTP server to serve test.root
+    url = "https://github.com/scikit-hep/scikit-hep-testdata/raw/v0.4.33/src/skhep_testdata/data/uproot-issue121.root"
+
+    notifications = queue.Queue()
+    with uproot.source.fsspec.FSSpecSource(url) as source:
+        chunks = source.chunks(
+            [(0, 100), (50, 55), (200, 400)], notifications=notifications
+        )
+        expected = {(chunk.start, chunk.stop): chunk for chunk in chunks}
+        while len(expected) > 0:
+            chunk = notifications.get()
+            expected.pop((chunk.start, chunk.stop))
+
+        chunk_data_sum = {sum(chunk.raw_data) for chunk in chunks}
+        assert chunk_data_sum == {3967, 413, 10985}, "Chunk data does not match"