How to protect against submitting too many jobs to a ThreadPoolExecutor?

user_1040915 0
Viewed 493

I recently came across the ThreadPoolExecutor class and have been using it in a toy project. ThreadPoolExecutor has a _work_queue field - which, when I submit more tasks than the number of workers assigned to the executor, starts filling up:

>>> from concurrent.futures import ThreadPoolExecutor
>>> from time import sleep
>>> def wait():
...   sleep(1000)
...
>>> tpe = ThreadPoolExecutor(max_workers=10)
>>> tpe._work_queue.qsize()
0
>>> for i in range(11):
...   tpe.submit(wait)
...
<Future at 0x10cf63908 state=running>
[...snip...]
<Future at 0x10d20b278 state=pending>
>>> tpe._work_queue.qsize()
1

I notice that the _work_queue has a method full(), which, presumably, indicates that the queue cannot take any more tasks. I would expect an exception to be thrown if I submit more tasks than the queue can hold - however, I don't see that behaviour referenced anywhere in the documentation, and I wasn't able to replicate it even after adding more than 100,000 tasks to my Executor.

Right now, I'm defending against this behaviour with:

for task_to_do in my_tasks:
  if tpe._work_queue.full():
    sleep(0.1)
  tpe.submit(task_to_do)

which feels hacky because of the reference to the "private" queue - I guess it would be more pythonic to do:

for task_to_do in my_tasks:
  task_added = False
  while not task_added:
    try:
      tpe.submit(task_to_do)
      task_added = True
    except SomeExceptionWhoseNameIDoNotKnowYet as e:
      pass

but, in order to do so, I need to know what kind of Exception would be thrown (or, I guess, just catch Exception)

concurrency python
1 Answers

override the submit method of ThreadPoolExecutor

class YourThreadPoolExecutor(ThreadPoolExecutor):
    def submit(self, fn, /, *args, **kwargs):
        if self._work_queue.qsize() > self._max_workers:
            raise Exception('Too many submits')
        with self._shutdown_lock, _global_shutdown_lock:
            if self._broken:
                raise BrokenThreadPool(self._broken)

            if self._shutdown:
                raise RuntimeError('cannot schedule new futures after shutdown')
            if _shutdown:
                raise RuntimeError('cannot schedule new futures after '
                                   'interpreter shutdown')

            f = _base.Future()
            w = _WorkItem(f, fn, args, kwargs)

            self._work_queue.put(w)
            self._adjust_thread_count()
            return f

Then always use YourThreadPoolExecutor instead of ThreadPoolExecutor

user_108996380
Related