Mercurial > piecrust2

comparison piecrust/workerpool.py @ 1008:09c3d415d9e5

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
bake: Make `batch-size` option work again.
author Ludovic Chabant <ludovic@chabant.com>
date Sat, 25 Nov 2017 23:01:48 -0800
parents 1857dbd4580f
children c4cf3cfe2726
comparison
equal deleted inserted replaced
1007:09dc0240f08a 1008:09c3d415d9e5
176 176
177 result_list = [] 177 result_list = []
178 178
179 for td in task_data_list: 179 for td in task_data_list:
180 try: 180 try:
181 res = w.process(td) 181 worker_res = w.process(td)
182 result_list.append((td, res, True)) 182 result_list.append((td, worker_res, True))
183 except Exception as e: 183 except Exception as e:
184 logger.debug( 184 logger.debug(
185 "Error processing job, sending exception to main process:") 185 "Error processing job, sending exception to main process:")
186 logger.debug(traceback.format_exc()) 186 logger.debug(traceback.format_exc())
187 we = _get_worker_exception_data(wid) 187 we = _get_worker_exception_data(wid)
188 res = (td, we, False) 188 error_res = (td, we, False)
189 result_list.append((td, res, False)) 189 result_list.append((td, error_res, False))
190 190
191 res = (task_type, wid, result_list) 191 res = (task_type, wid, result_list)
192 put_start_time = time.perf_counter() 192 put_start_time = time.perf_counter()
193 put(res) 193 put(res)
194 time_in_put += (time.perf_counter() - put_start_time) 194 time_in_put += (time.perf_counter() - put_start_time)
382 self._event.set() 382 self._event.set()
383 return False 383 return False
384 384
385 return True 385 return True
386 386
387 def _onTaskDone(self): 387 def _onTaskDone(self, done_count):
388 with self._lock_jobs_left: 388 with self._lock_jobs_left:
389 left = self._jobs_left - 1 389 left = self._jobs_left - done_count
390 self._jobs_left = left 390 self._jobs_left = left
391 391
392 if left == 0: 392 if left == 0:
393 self._event.set() 393 self._event.set()
394 394
429 "Worker %d failed to process a job:" % wid) 429 "Worker %d failed to process a job:" % wid)
430 logger.error(data) 430 logger.error(data)
431 except Exception as ex: 431 except Exception as ex:
432 logger.exception(ex) 432 logger.exception(ex)
433 433
434 if task_type == TASK_JOB: 434 if task_type == TASK_JOB or task_type == TASK_JOB_BATCH:
435 pool._onTaskDone() 435 pool._onTaskDone(len(res_data_list))
436 436
437 437
438 class _ReportHandler: 438 class _ReportHandler:
439 def __init__(self, worker_count): 439 def __init__(self, worker_count):
440 self.reports = [None] * worker_count 440 self.reports = [None] * worker_count