|
| 1 | +import asyncio |
| 2 | +from collections import defaultdict, deque |
| 3 | +import time |
| 4 | +from typing import Deque, Dict, Optional, Set, TypeVar, Any |
| 5 | +from contextvars import ContextVar |
| 6 | + |
| 7 | +from loguru import logger |
| 8 | + |
| 9 | +from .abstract_queue import AbstractQueue |
| 10 | + |
| 11 | +T = TypeVar("T") |
| 12 | +MaybeStr = str | None |
| 13 | + |
| 14 | +_NO_GROUP = object() |
| 15 | +_current_group: ContextVar[Any] = ContextVar("current_group", default=_NO_GROUP) |
| 16 | + |
| 17 | + |
| 18 | +class GroupQueue(AbstractQueue[T]): |
| 19 | + """Queue with exclusive processing per group.""" |
| 20 | + |
| 21 | + def __init__( |
| 22 | + self, |
| 23 | + group_key: MaybeStr = None, |
| 24 | + name: MaybeStr = None, |
| 25 | + lock_timeout: float = 300, |
| 26 | + ): |
| 27 | + super().__init__(name) |
| 28 | + self.group_key = group_key |
| 29 | + self._queues: Dict[MaybeStr, Deque[T]] = defaultdict(deque) |
| 30 | + self._locked: Set[MaybeStr] = set() |
| 31 | + self._queue_not_empty = asyncio.Condition() |
| 32 | + self.lock_timeout = lock_timeout |
| 33 | + self._lock_timestamps: Dict[MaybeStr, float] = {} |
| 34 | + self._timeout_task: Optional[asyncio.Task[None]] = None |
| 35 | + |
| 36 | + async def _background_timeout_check(self) -> None: |
| 37 | + """Periodically release locks that have timed out.""" |
| 38 | + while True: |
| 39 | + try: |
| 40 | + await asyncio.sleep(self.lock_timeout / 4) |
| 41 | + async with self._queue_not_empty: |
| 42 | + await self._release_expired_locks() |
| 43 | + except asyncio.CancelledError: |
| 44 | + break |
| 45 | + |
| 46 | + def _extract_group_key(self, item: T) -> MaybeStr: |
| 47 | + """Extract the group key from an item.""" |
| 48 | + if self.group_key is None: |
| 49 | + return None |
| 50 | + if not hasattr(item, self.group_key): |
| 51 | + raise ValueError( |
| 52 | + f"Item {item!r} lacks attribute '{self.group_key}' required for grouping" |
| 53 | + ) |
| 54 | + return getattr(item, self.group_key) |
| 55 | + |
| 56 | + async def put(self, item: T) -> None: |
| 57 | + """Add item to its group's queue.""" |
| 58 | + group_key = self._extract_group_key(item) |
| 59 | + async with self._queue_not_empty: |
| 60 | + self._queues[group_key].append(item) |
| 61 | + self._queue_not_empty.notify_all() |
| 62 | + |
| 63 | + async def _release_expired_locks(self) -> None: |
| 64 | + """Release locks that have exceeded the timeout.""" |
| 65 | + now = time.time() |
| 66 | + expired_groups = [] |
| 67 | + |
| 68 | + for group, timestamp in list(self._lock_timestamps.items()): |
| 69 | + if now - timestamp > self.lock_timeout: |
| 70 | + expired_groups.append(group) |
| 71 | + logger.warning(f"Releasing expired lock for group {group}") |
| 72 | + self._locked.discard(group) |
| 73 | + del self._lock_timestamps[group] |
| 74 | + |
| 75 | + if expired_groups: |
| 76 | + self._queue_not_empty.notify_all() |
| 77 | + |
| 78 | + async def get(self) -> T: |
| 79 | + """Get the next item from an unlocked group, locking that group.""" |
| 80 | + if self._timeout_task is None or self._timeout_task.done(): |
| 81 | + self._timeout_task = asyncio.create_task(self._background_timeout_check()) |
| 82 | + |
| 83 | + async with self._queue_not_empty: |
| 84 | + while True: |
| 85 | + await self._release_expired_locks() |
| 86 | + |
| 87 | + for group, queue in self._queues.items(): |
| 88 | + if queue and group not in self._locked: |
| 89 | + self._locked.add(group) |
| 90 | + self._lock_timestamps[group] = time.time() |
| 91 | + _current_group.set(group) |
| 92 | + return queue[0] |
| 93 | + |
| 94 | + await self._queue_not_empty.wait() |
| 95 | + |
| 96 | + async def commit(self) -> None: |
| 97 | + """Remove the current item and unlock its group.""" |
| 98 | + group = _current_group.get() |
| 99 | + if group is _NO_GROUP: |
| 100 | + logger.warning("commit() called without active get()") |
| 101 | + return |
| 102 | + |
| 103 | + async with self._queue_not_empty: |
| 104 | + queue = self._queues.get(group) |
| 105 | + if queue: |
| 106 | + queue.popleft() |
| 107 | + if not queue: |
| 108 | + del self._queues[group] |
| 109 | + |
| 110 | + self._locked.discard(group) |
| 111 | + self._lock_timestamps.pop(group, None) |
| 112 | + _current_group.set(_NO_GROUP) |
| 113 | + self._queue_not_empty.notify_all() |
| 114 | + |
| 115 | + async def teardown(self) -> None: |
| 116 | + """Wait until all queues are empty and no groups are locked.""" |
| 117 | + async with self._queue_not_empty: |
| 118 | + while any(self._queues.values()) or self._locked: |
| 119 | + await self._queue_not_empty.wait() |
| 120 | + |
| 121 | + if self._timeout_task and not self._timeout_task.done(): |
| 122 | + self._timeout_task.cancel() |
| 123 | + try: |
| 124 | + await self._timeout_task |
| 125 | + except asyncio.CancelledError: |
| 126 | + pass |
| 127 | + |
| 128 | + async def size(self) -> int: |
| 129 | + """Return total number of items across all groups.""" |
| 130 | + async with self._queue_not_empty: |
| 131 | + return sum(len(queue) for queue in self._queues.values()) |
| 132 | + |
| 133 | + async def force_unlock_all(self) -> None: |
| 134 | + """Force unlock all groups.""" |
| 135 | + async with self._queue_not_empty: |
| 136 | + self._locked.clear() |
| 137 | + self._lock_timestamps.clear() |
| 138 | + self._queue_not_empty.notify_all() |
0 commit comments