some comments to the SmartQueue

yapapi/executor/_smartq.py

@@ -27,7 +27,16 @@
 Item = TypeVar("Item")
 
 
-class Handle(Generic[Item], object):
+class Handle(
+    Generic[Item], object
+):  # Handling? QueueItem? ConsumedItem? ConsumerItem? ConsumerBinding?
+    """
+    Binding between a queue item and a specific consumer.
+
+    Additionally it keeps track of the previously used consumers of the given item
+    to prevent them from being assigned to this item again.
+    """
+
     __slots__ = ("_data", "_prev_consumers", "_consumer")
 
     def __init__(self, data: Item, *, consumer: Optional["Consumer[Item]"] = None):
@@ -52,9 +61,17 @@ def data(self) -> Item:
 
 
 class SmartQueue(Generic[Item], object):
-    def __init__(self, items: Iterable[Item], *, retry_cnt: int = 2):
+    def __init__(self, items: Iterable[Item]):
+        """
+        :param items: the items to be iterated over
+        :param retry_cnt:
+        """
         self._items: Iterator[Item] = peekable(items)
+
+        """the items scheduled for reassignment to another consumer"""
         self._rescheduled_items: Set[Handle[Item]] = set()
+
+        """the items currently assigned to consumers"""
         self._in_progress: Set[Handle[Item]] = set()
 
         # Synchronization primitives
@@ -72,7 +89,7 @@ def has_unassigned_items(self) -> bool:
         An item is _unassigned_ if it's new (hasn't been retrieved yet by any consumer)
         or it has been rescheduled and is not in progress.
 
-        A queue has unassigned items iff `get()` will immediately return some item,
+        A queue has unassigned items iff `get()` immediately returns some item,
         without waiting for an item that is currently "in progress" to be rescheduled.
         """
         return self.has_new_items() or bool(self._rescheduled_items)
@@ -126,13 +143,15 @@ async def mark_done(self, handle: Handle[Item]) -> None:
             )
 
     async def reschedule(self, handle: Handle[Item]) -> None:
+        """free the item for reassignment to another consumer"""
         assert handle in self._in_progress, "handle is not in progress"
         async with self._lock:
             self._in_progress.remove(handle)
             self._rescheduled_items.add(handle)
             self._new_items.notify_all()
 
     async def reschedule_all(self, consumer: "Consumer[Item]"):
+        """make all items currently assigned to the consumer available for reassignment"""
         async with self._lock:
             handles = [handle for handle in self._in_progress if handle.consumer == consumer]
             for handle in handles:
@@ -149,6 +168,7 @@ def stats(self) -> Dict:
         }
 
     async def wait_until_done(self) -> None:
+        """wait until all items in the queue are processed"""
         async with self._lock:
             while self.__has_data():
                 await self._eof.wait()
@@ -160,6 +180,11 @@ class Consumer(
     AsyncIterable[Handle[Item]],
     ContextManager["Consumer[Item]"],
 ):
+    """
+    Provides an interface to asynchronously iterate over items in the given queue
+    while cooperating with other consumers attached to this queue.
+    """
+
     def __init__(self, queue: SmartQueue[Item]):
         self._queue = queue
         self._fetched: Optional[Handle[Item]] = None
@@ -178,6 +203,7 @@ def __exit__(
 
     @property
     def last_item(self) -> Optional[Item]:
+        """shouldn't it be called `current_item` ?"""
         return self._fetched.data if self._fetched else None
 
     async def __anext__(self) -> Handle[Item]: