Python Multiprocessing provides two high level classes that enable us to take advantage of a pool of workers to perform concurrent tasks: ThreadPoolExecutor and ProcessPoolExecutor.
I would suggest that a third is missing from the library: AsyncPoolExecutor.
The interface of an Executor is simple. It affords submit and map methods that takes a function and its arguments as its arguments and return one or many futures. A future allows for a result to be acquired as it becomes available or to throw an exception should the called function fail.
class AsyncPoolExecutor:
def __init__(self, max_workers=None):
self._max_workers = max_workers or 5
self._queue = asyncio.Queue()
self._futures = []
self._tasks = []
async def submit(self, func, *args, **kwargs):
future = asyncio.Future()
await self._queue.put((func, args, kwargs, future))
return future
async def map(self, func, iterables):
futures = []
for args in iterables:
future = await self.submit(func, *args)
futures.append(future)
return futures
async def __aenter__(self):
# Start worker tasks
for _ in range(self._max_workers):
task = asyncio.create_task(self._worker())
self._tasks.append(task)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.shutdown()
async def _worker(self):
while True:
try:
func, args, kwargs, future = await asyncio.wait_for(
self._queue.get(), timeout=1.0
)
result = await func(*args, **kwargs)
future.set_result(result)
except asyncio.TimeoutError:
break
except Exception as e:
future.set_exception(e)
async def shutdown(self):
# Cancel all remaining tasks
for task in self._tasks:
task.cancel()
await asyncio.gather(*self._tasks, return_exceptions=True)
Full source available as gist or test in repo.
I am an architect and this is an implementation, certainly neither exhaustive nor correct. But it is testable and can demonstrate the feature that I am suggesting. Map is just an iterative call to Submit. But we have a queue of work and a specified number of tasks to perform the work and we return concurrent.futures.Future instance(s). It is that set number of concurrent tasks that this pool provides.
Should you be using an external API with limits this could help.
Should you be calling a database with a set number of processes, this could help.
Iterating the results should provide individual results, but that is more correct and complete than needed to demonstrate my point.
You might notice that the Thread and Process Executors implement the Context Manager interface allowing us to write the following:
with ThreadPoolExecutor(max_workers=10) as executor:
futures = [executor.submit(worker, i) for i in range(100)]
for future in futures:
print(future.result())
Let us add that to our class as an Async Context Manager:
async def __aenter__(self):
# Initialize worker tasks
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.shutdown()
Here you can see a difference, our shutdown must be async. So I have disabled the pylint warning, that my shutdown differs from the Executors shutdown.
So now we come to the benefit of using this technique — performance.
The pytest I provide prepares our Thread and Process pools outside of the performance test, thus avoiding the creation and cleanup overheads. That is not true of the async test. The target of the tests is a tornado server — it merely returns a salutation including the requested resource:
@app.get("/api/greeting")
async def greeting(self):
self.write({"message": f"Hello, {self.request.arguments.get('name', [b'World'])[0].decode()}"})
So, running our server on port 8080, we use a fixture for our data:
@pytest.fixture
def data():
return [(i,) for i in range(100)]
And test our ThreadPoolExecutor:
def test_thread_pool(benchmark, data):
with ThreadPoolExecutor(max_workers=10) as executor:
benchmark(run_test, executor, data)
The ProcessPoolExecutor test is identical but for the construction of the pool. Our async test is near identical too, but we create the pool within the test using the Context Manager interface, as we have no global event loop:
async def test_async_pool(benchmark, data):
async with AsyncPoolExecutor(max_workers=10) as executor:
await benchmark(run_test, executor, data)
And the results:
Name (time in ms) | Min | Max | Mean | StdDev
test_a | 9.0464 | 38.7552 | 10.4006 | 3.0807
test_p | 10.6764 | 15.0337 | 11.9298 | 0.8462
test_t | 13.1372 | 44.6249 | 16.7564 | 5.2769
As I said, I am not correct or exhaustive, but I am intrigued. I have used tornado's HTTPClient and AsyncHTTPClient against a tornado server.
I assume but do not know, that that helps expose the efficiency of the library pools. I assume that the high max is the instantiation of the first event loop and threads. This is an IO operation and the very reason that asyncio was introduced to the standard library. It performs well.
Thanks for reading. If you have any thoughts or would like to improve on this, please feel free to comment or be in touch. The source to this madness is available here.