voussoir/voussoirkit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Add lock around running_count increment, decrement.

Browse source 
I thought these assignments were atomic for int but they're not.
This commit is contained in:
voussoir 2021-11-20 19:36:29 -08:00
parent cca488bc7a
commit 305c6bfdf4
No known key found for this signature in database
GPG key ID: 5F7554F8C26DACCB
1 changed files with 10 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
voussoirkit/threadpool.py
View file

@ -92,11 +92,16 @@ class PooledThread:
return NO_MORE_JOBS return NO_MORE_JOBS
log.debug('%s is running job %s.', self, job) log.debug('%s is running job %s.', self, job)
with self.pool._running_count_lock:
self.pool._running_count += 1 self.pool._running_count += 1
try: try:
job.run() job.run()
except BaseException: except BaseException:
traceback.print_traceback() traceback.print_traceback()
with self.pool._running_count_lock:
self.pool._running_count -= 1 self.pool._running_count -= 1
def join(self) -> None: def join(self) -> None:
@ -163,8 +168,11 @@ class ThreadPool:
self._running_count = 0 self._running_count = 0
self._result_queue = None self._result_queue = None
self._pending_jobs = lazychain.LazyChain() self._pending_jobs = lazychain.LazyChain()
# Should be held while manipulating self._pending_jobs.
self._job_manager_lock = threading.Lock() self._job_manager_lock = threading.Lock()
# Should be held while manipulating self._running_count.
self._running_count_lock = threading.Lock()
self._size = size self._size = size
self._threads = [PooledThread(pool=self) for x in range(size)] self._threads = [PooledThread(pool=self) for x in range(size)]