From 5413187c2db6f13679654c80439a6f5300b5eabf Mon Sep 17 00:00:00 2001
From: karajan1001 <mishanyo1001@gmail.com>
Date: Fri, 14 Oct 2022 20:51:33 +0800
Subject: [PATCH] Give lock acquiring more time in concurrency situation.

fix: #8417

1. Give the executor initialization and result collection more time for
   acquiring locks.
---
 dvc/repo/experiments/queue/base.py | 18 ++++++++++++++++--
 1 file changed, 16 insertions(+), 2 deletions(-)

diff --git a/dvc/repo/experiments/queue/base.py b/dvc/repo/experiments/queue/base.py
index 92605e916..e7739d284 100644
--- a/dvc/repo/experiments/queue/base.py
+++ b/dvc/repo/experiments/queue/base.py
@@ -18,11 +18,12 @@ from typing import (
     Union,
 )
 
-from funcy import cached_property
+from funcy import cached_property, retry
 
 from dvc.dependency import ParamsDependency
 from dvc.env import DVCLIVE_RESUME
 from dvc.exceptions import DvcException
+from dvc.lock import LockError
 from dvc.ui import ui
 
 from ..exceptions import CheckpointExistsError, ExperimentExistsError
@@ -560,6 +561,7 @@ class BaseStashQueue(ABC):
         self.scm.add(list(params.keys()))
 
     @staticmethod
+    @retry(180, errors=LockError, timeout=1)
     @scm_locked
     def init_executor(
         exp: "Experiments",
@@ -601,8 +603,9 @@ class BaseStashQueue(ABC):
         )
 
     @staticmethod
+    @retry(180, errors=LockError, timeout=1)
     @scm_locked
-    def collect_executor(
+    def collect_git(
         exp: "Experiments",
         executor: BaseExecutor,
         exec_result: ExecutorResult,
@@ -626,6 +629,17 @@ class BaseStashQueue(ABC):
                 logger.debug("Collected experiment '%s'.", exp_rev[:7])
                 results[exp_rev] = exec_result.exp_hash
 
+        return results
+
+    @classmethod
+    def collect_executor(
+        cls,
+        exp: "Experiments",
+        executor: BaseExecutor,
+        exec_result: ExecutorResult,
+    ) -> Dict[str, str]:
+        results = cls.collect_git(exp, executor, exec_result)
+
         if exec_result.ref_info is not None:
             executor.collect_cache(exp.repo, exec_result.ref_info)