diff --git a/deployment/aliyun/binance_lob_archiver.py b/deployment/aliyun/binance_lob_archiver.py index 3a61d6631..7486ecf8e 100644 --- a/deployment/aliyun/binance_lob_archiver.py +++ b/deployment/aliyun/binance_lob_archiver.py @@ -693,16 +693,27 @@ def run_process_watchdog() -> None: async def cancel_tasks_bounded(tasks: tuple[asyncio.Task, ...]) -> None: if not tasks: return - for task in tasks: - task.cancel() + cancel_requested = {task: task.cancel() for task in tasks} _, pending = await asyncio.wait( tasks, timeout=TASK_CANCEL_TIMEOUT_SECONDS ) if pending: - LOG.error( - "task cancellation timed out pending=%s", - len(pending), - ) + for task in pending: + stack = task.get_stack(limit=1) + location = ( + f"{stack[-1].f_code.co_name}:{stack[-1].f_lineno}" + if stack + else "unknown" + ) + LOG.error( + "task cancellation timed out name=%s coro=%s " + "cancel_requested=%s cancelling=%s location=%s", + task.get_name(), + task.get_coro().__qualname__, + cancel_requested[task], + task.cancelling(), + location, + ) raise TaskCancellationStuck( f"task cancellation timed out pending={len(pending)}" ) @@ -893,10 +904,15 @@ async def run_session( queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_BUFFERED_DIFFS) failure: BaseException | None = None receivers = [ - asyncio.create_task(receive_url(url, queue, stop)) for url in stream_urls() + asyncio.create_task( + receive_url(url, queue, stop), name=f"receiver-{index}" + ) + for index, url in enumerate(stream_urls()) ] snapshot_limiter = SnapshotRateLimiter(SNAPSHOT_REQUESTS_PER_SECOND) - snapshotter = asyncio.create_task(produce_snapshots(queue, snapshot_limiter)) + snapshotter = asyncio.create_task( + produce_snapshots(queue, snapshot_limiter), name="snapshotter" + ) tasks = [*receivers, snapshotter] resync_tasks: dict[str, asyncio.Task] = {} LOG.info( @@ -990,7 +1006,8 @@ async def run_session( resync_tasks[gap.symbol] = asyncio.create_task( produce_snapshot( gap.symbol.lower(), queue, snapshot_limiter - ) + ), + name=f"resync-{gap.symbol}", ) continue @@ -1028,6 +1045,11 @@ async def run_session( failure = error raise finally: + LOG.info( + "cancelling session tasks queue_size=%s queue_maxsize=%s", + queue.qsize(), + queue.maxsize, + ) await cancel_tasks_bounded(tuple([*tasks, *resync_tasks.values()])) archive_only = isinstance(failure, SequenceGap) drain_gap: SequenceGap | None = None diff --git a/deployment/aliyun/test_binance_lob_archiver.py b/deployment/aliyun/test_binance_lob_archiver.py index 58ecacab2..6be9b9817 100644 --- a/deployment/aliyun/test_binance_lob_archiver.py +++ b/deployment/aliyun/test_binance_lob_archiver.py @@ -342,15 +342,21 @@ async def stubborn(): except ARCHIVER.asyncio.CancelledError: await release.wait() - task = ARCHIVER.asyncio.create_task(stubborn()) + task = ARCHIVER.asyncio.create_task(stubborn(), name="stubborn-test") await ARCHIVER.asyncio.sleep(0) with ( patch.object(ARCHIVER, "TASK_CANCEL_TIMEOUT_SECONDS", 0.01), - self.assertLogs("binance-lob-archiver", level="ERROR"), + self.assertLogs("binance-lob-archiver", level="ERROR") as logs, ): with self.assertRaises(ARCHIVER.TaskCancellationStuck): await ARCHIVER.cancel_tasks_bounded((task,)) + self.assertIn("name=stubborn-test", logs.output[0]) + self.assertIn("coro=RuntimeContractTests", logs.output[0]) + self.assertIn("cancel_requested=True", logs.output[0]) + self.assertIn("cancelling=1", logs.output[0]) + self.assertIn("location=stubborn:", logs.output[0]) + release.set() await task