Skip to content

Instantly share code, notes, and snippets.

@sthagen
Forked from noxdafox/max_queue_size_pool.py
Created January 17, 2022 08:47
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save sthagen/73a60d57b6881b5905b9eba32e181841 to your computer and use it in GitHub Desktop.
Save sthagen/73a60d57b6881b5905b9eba32e181841 to your computer and use it in GitHub Desktop.
This code snippet shows how to wrap a concurrent.futures.Executor class to provide a limited queue size.
from threading import BoundedSemaphore
from concurrent.futures import ProcessPoolExecutor
class MaxQueuePool:
"""This Class wraps a concurrent.futures.Executor
limiting the size of its task queue.
If `max_queue_size` tasks are submitted, the next call to submit will block
until a previously submitted one is completed.
"""
def __init__(self, executor, max_queue_size, max_workers=None):
self.pool = executor(max_workers=max_workers)
self.pool_queue = BoundedSemaphore(max_queue_size)
def submit(self, function, *args, **kwargs):
"""Submits a new task to the pool, blocks if Pool queue is full."""
self.pool_queue.acquire()
future = self.pool.submit(function, *args, **kwargs)
future.add_done_callback(self.pool_queue_callback)
return future
def pool_queue_callback(self, _):
"""Called once task is done, releases one queue slot."""
self.pool_queue.release()
if __name__ == '__main__':
pool = MaxQueuePool(ProcessPoolExecutor, 8)
f = pool.submit(print, "Hello World!")
f.result()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment