From 48c0c323956090c9cc9240a070e765762a8f70a2 Mon Sep 17 00:00:00 2001 From: Peter Bell Date: Tue, 13 Apr 2021 20:12:31 +0100 Subject: [PATCH 1/2] Remove unnecesary sleep when submitting parallel tasks --- sphinx/util/parallel.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/sphinx/util/parallel.py b/sphinx/util/parallel.py index ed73ee4d6b9..9fa1e0c3eb4 100644 --- a/sphinx/util/parallel.py +++ b/sphinx/util/parallel.py @@ -100,12 +100,13 @@ def add_task(self, task_func: Callable, arg: Any = None, result_func: Callable = args=(psend, task_func, arg)) self._procs[tid] = proc self._precvsWaiting[tid] = precv - self._join_one() + _ = self._join_one() def join(self) -> None: try: while self._pworking: - self._join_one() + if not self._join_one(): + time.sleep(0.02) except Exception: # shutdown other child processes on failure self.terminate() @@ -119,7 +120,8 @@ def terminate(self) -> None: self._precvs.pop(tid) self._pworking -= 1 - def _join_one(self) -> None: + def _join_one(self) -> bool: + joined_any = False for tid, pipe in self._precvs.items(): if pipe.poll(): exc, logs, result = pipe.recv() @@ -131,15 +133,17 @@ def _join_one(self) -> None: self._procs[tid].join() self._precvs.pop(tid) self._pworking -= 1 + joined_any = True break - else: - time.sleep(0.02) + while self._precvsWaiting and self._pworking < self.nproc: newtid, newprecv = self._precvsWaiting.popitem() self._precvs[newtid] = newprecv self._procs[newtid].start() self._pworking += 1 + return joined_any + def make_chunks(arguments: Sequence[str], nproc: int, maxbatch: int = 10) -> List[Any]: # determine how many documents to read in one go From 33b8f4334f493f2202b1964c28983b0570869956 Mon Sep 17 00:00:00 2001 From: Peter Bell Date: Sat, 17 Apr 2021 13:17:59 +0100 Subject: [PATCH 2/2] Remove redundant assignment --- sphinx/util/parallel.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sphinx/util/parallel.py b/sphinx/util/parallel.py index 9fa1e0c3eb4..8d33d37d518 100644 --- a/sphinx/util/parallel.py +++ b/sphinx/util/parallel.py @@ -100,7 +100,7 @@ def add_task(self, task_func: Callable, arg: Any = None, result_func: Callable = args=(psend, task_func, arg)) self._procs[tid] = proc self._precvsWaiting[tid] = precv - _ = self._join_one() + self._join_one() def join(self) -> None: try: