From 16bb31a284616e12490eb530d72a18109e082e9f Mon Sep 17 00:00:00 2001 From: karajan1001 Date: Mon, 13 Jun 2022 15:43:58 +0800 Subject: [PATCH] queue worker shutdown: disable restore fix: #7884 The complete task restored back to the queue causing a task duplication. Might related to a bug from the celery celery/celery#5663 Follow what in https://github.com/celery/celery/issues/5663#issuecomment-734871162 Disabled the restore_at_shutdown --- dvc/repo/experiments/queue/local.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/dvc/repo/experiments/queue/local.py b/dvc/repo/experiments/queue/local.py index 2bed25ae0..83d8a7152 100644 --- a/dvc/repo/experiments/queue/local.py +++ b/dvc/repo/experiments/queue/local.py @@ -71,6 +71,11 @@ class LocalCeleryQueue(BaseStashQueue): @cached_property def celery(self) -> "FSApp": + from kombu.transport.filesystem import Channel # type: ignore + + # related to https://github.com/iterative/dvc-task/issues/61 + Channel.QoS.restore_at_shutdown = False + from dvc_task.app import FSApp app = FSApp( @@ -181,7 +186,7 @@ class LocalCeleryQueue(BaseStashQueue): for msg, entry in self._iter_processed(): task_id = msg.headers["id"] - result = AsyncResult(task_id) + result: AsyncResult = AsyncResult(task_id) if not result.ready(): yield _TaskEntry(task_id, entry) @@ -190,7 +195,7 @@ class LocalCeleryQueue(BaseStashQueue): for msg, entry in self._iter_processed(): task_id = msg.headers["id"] - result = AsyncResult(task_id) + result: AsyncResult = AsyncResult(task_id) if result.ready(): yield _TaskEntry(task_id, entry) @@ -230,7 +235,7 @@ class LocalCeleryQueue(BaseStashQueue): if entry.stash_rev == active_entry.stash_rev: logger.debug("Waiting for exp task '%s' to complete", task_id) try: - result = AsyncResult(task_id) + result: AsyncResult = AsyncResult(task_id) result.get(timeout=timeout) except _CeleryTimeout as exc: raise DvcException(