add logging and checks

Author: googlyrahman

gcsfs/prefetcher.py

@@ -1,10 +1,13 @@
 import asyncio
 import ctypes
+import logging
 import threading
 from collections import deque
 
 import fsspec.asyn
 
+logger = logging.getLogger(__name__)
+
 PyBytes_FromStringAndSize = ctypes.pythonapi.PyBytes_FromStringAndSize
 PyBytes_FromStringAndSize.restype = ctypes.py_object
 PyBytes_FromStringAndSize.argtypes = [ctypes.c_void_p, ctypes.c_ssize_t]
@@ -63,11 +66,26 @@ def __init__(
         concurrency=4,
         **kwargs,
     ):
+        logger.debug(
+            "Initializing BackgroundPrefetcher: blocksize=%d, size=%d, max_prefetch_size=%s, concurrency=%d",
+            blocksize,
+            size,
+            max_prefetch_size,
+            concurrency,
+        )
         self.blocksize = blocksize
         self.size = size
         self.fetcher = fetcher
         self.concurrency = concurrency
+
+        if max_prefetch_size is not None and max_prefetch_size < blocksize:
+            # Ensure we have a positive prefetch window.
+            raise ValueError(
+                f"User max_prefetch_size ({max_prefetch_size}) is smaller than blocksize ({blocksize})."
+            )
+
         self._user_max_prefetch_size = max_prefetch_size
+
         self.sequential_streak = 0
         self.user_offset = 0
         self.current_offset = 0
@@ -82,6 +100,7 @@ def __init__(
         self._error = None
 
         async def _start_producer():
+            logger.debug("Starting background producer task")
             self.queue = asyncio.Queue()
             self._wakeup_producer = asyncio.Event()
             self._producer_task = asyncio.create_task(self._producer_loop())
@@ -111,6 +130,10 @@ def max_prefetch_size(self) -> int:
         return max(2 * self._get_adaptive_blocksize(), self.DEFAULT_PREFETCH_SIZE)
 
     async def _cancel_all_tasks(self):
+        logger.debug(
+            "Cancelling all active producer tasks. Task count: %d",
+            len(self._active_tasks),
+        )
         self.is_stopped = True
         self._wakeup_producer.set()
 
@@ -142,9 +165,11 @@ async def _cancel_all_tasks(self):
                     break
 
         if tasks_to_wait:
+            logger.debug("Waiting for %d cancelled tasks to finish", len(tasks_to_wait))
             await asyncio.gather(*tasks_to_wait, return_exceptions=True)
 
     async def _restart_producer(self):
+        logger.info("Restarting producer loop (likely due to seek operation)")
         await self._cancel_all_tasks()
         self.is_stopped = False
         self._error = None
@@ -154,6 +179,7 @@ async def _restart_producer(self):
         self._producer_task = asyncio.create_task(self._producer_loop())
 
     async def _producer_loop(self):
+        logger.debug("Producer loop running")
         try:
             while not self.is_stopped:
                 await self._wakeup_producer.wait()
@@ -164,6 +190,12 @@ async def _producer_loop(self):
                     (self.sequential_streak + 1) * block_size,
                     self.max_prefetch_size,
                 )
+                logger.debug(
+                    "Producer awake. Current offset: %d, User offset: %d, Prefetch size: %d",
+                    self.current_offset,
+                    self.user_offset,
+                    prefetch_size,
+                )
 
                 while (
                     not self.is_stopped
@@ -207,6 +239,12 @@ async def _producer_loop(self):
                             else 1
                         )  # sequential usecase
 
+                    logger.debug(
+                        "Producer spawning fetch task at offset %d, size %d, split_factor %d",
+                        self.current_offset,
+                        actual_size,
+                        sfactor,
+                    )
                     download_task = asyncio.create_task(
                         self.fetcher(
                             self.current_offset, actual_size, split_factor=sfactor
@@ -219,27 +257,37 @@ async def _producer_loop(self):
                     self.current_offset += actual_size
 
         except asyncio.CancelledError:
+            logger.debug("Producer loop cancelled")
             pass
         except Exception as e:
+            logger.error("Producer loop encountered an error: %s", e, exc_info=True)
             self.is_stopped = True
             self._error = e
             await self.queue.put(e)
 
     async def read(self):
         """Reads the next chunk from the object."""
         if self.user_offset >= self.size:
+            logger.debug(
+                "Read requested but user_offset (%d) is >= size (%d). Returning empty bytes.",
+                self.user_offset,
+                self.size,
+            )
             return b""
         if self.is_stopped and self.queue.empty():
             # This may happen if user read despite previous read produced an exception.
+            logger.error("Read attempted on a stopped producer with an empty queue.")
             raise RuntimeError("Could not fetch data, the producer is stopped")
 
         if self.queue.empty():
+            logger.debug("Queue empty during read, waking up producer.")
             self._wakeup_producer.set()
 
         task = await self.queue.get()
 
         # Check if the producer pushed an exception
         if isinstance(task, Exception):
+            logger.error("Exception retrieved from read queue: %s", task)
             self.is_stopped = True
             self._error = task
             raise task
@@ -251,8 +299,10 @@ async def read(self):
                 self._wakeup_producer.set()  # starts prefetching.
             return block
         except asyncio.CancelledError:
+            logger.debug("Read task was cancelled.")
             raise
         except Exception as e:
+            logger.error("Error awaiting read task: %s", e, exc_info=True)
             self.is_stopped = True
             self._error = e
             raise e
@@ -261,13 +311,16 @@ async def seek(self, new_offset):
         if new_offset == self.user_offset:
             return
 
+        logger.info("Seeking from %d to %d", self.user_offset, new_offset)
         self.user_offset = new_offset
         self.current_offset = new_offset
         await self._restart_producer()
 
     async def _async_fetch(self, start, end):
+        logger.debug("_async_fetch called for range %d - %d", start, end)
         if start != self.user_offset:
             # We seeked elsewhere, reset the current block
+            logger.debug("Seek detected in _async_fetch. Resetting block and seeking.")
             self._current_block = b""
             self._current_block_idx = 0
             await self.seek(start)
@@ -325,21 +378,30 @@ async def _async_fetch(self, start, end):
         else:
             out = b"".join(chunks)
 
+        logger.debug("Finished _async_fetch. Total bytes collected: %d", len(out))
         return out
 
     def _fetch(self, start: int | None, end: int | None) -> bytes:
         if start is None:
             start = 0
         if end is None:
             end = self.size
+
+        logger.debug("_fetch called for start=%s, end=%s", start, end)
+
         if start >= self.size or start >= end:
+            logger.debug(
+                "Invalid bounds in _fetch or EOF reached. Returning empty bytes."
+            )
             return b""
 
         with self._lock:
             if self._error:
+                logger.error("Cannot _fetch: instance has an active error state.")
                 raise self._error
 
             if self.is_stopped:
+                logger.error("Cannot _fetch: instance is closed/stopped.")
                 raise RuntimeError(
                     "The file instance has been closed. This can occur if a close operation"
                     "is executed concurrently while a read operation is still in progress."
@@ -348,11 +410,13 @@ def _fetch(self, start: int | None, end: int | None) -> bytes:
             try:
                 result = fsspec.asyn.sync(self.loop, self._async_fetch, start, end)
             except Exception as e:
+                logger.error("Exception during synchronous fetch: %s", e, exc_info=True)
                 self.is_stopped = True
                 self._error = e
                 raise
 
             if self.is_stopped:
+                logger.error("Instance stopped during fetch operation.")
                 raise RuntimeError(
                     "The file instance has been closed. This can occur if a close operation"
                     "is executed concurrently while a read operation is still in progress."
@@ -362,7 +426,9 @@ def _fetch(self, start: int | None, end: int | None) -> bytes:
 
     def close(self):
         """Clean shutdown. Cancels tasks and waits for them to abort."""
+        logger.info("Closing BackgroundPrefetcher")
         if self.is_stopped:
+            logger.debug("BackgroundPrefetcher already stopped")
             return
         self.is_stopped = True
         with self._lock:

gcsfs/tests/test_prefetcher.py

@@ -529,3 +529,21 @@ def test_read_runtime_error_on_stopped_empty():
     ):
         fsspec.asyn.sync(bp.loop, bp.read)
     bp.close()
+
+
+def test_init_invalid_max_prefetch_size():
+    with pytest.raises(
+        ValueError,
+        match=r"User max_prefetch_size \(50\) is smaller than blocksize \(100\)\.",
+    ):
+        BackgroundPrefetcher(
+            blocksize=100, fetcher=MockFetcher(b""), size=1000, max_prefetch_size=50
+        )
+
+
+def test_init_valid_max_prefetch_size_edge_case():
+    bp = BackgroundPrefetcher(
+        blocksize=100, fetcher=MockFetcher(b""), size=1000, max_prefetch_size=100
+    )
+    assert bp._user_max_prefetch_size == 100
+    bp.close()