concurrent.futures.ProcessPoolExecutor.map() doesn't batch function arguments by chunks

[tbrink]

BPO	11271
Nosy	@brianquinlan, @pitrou, @MojoVampire
Files	map_comparison.py: Demonstration and workaround new_processpoolexecutor.py: Improved the workaround map_chunksize.patch: Adds chunksize parameter to ProcessPoolExecutor.map. map_chunksize_with_test.patch test_mult.py: benchmark script map_chunksize_with_docs.patch: Updated patch with doc changes. map_chunksize_docs_update.patch: More tests, small change to how we "yield from" the result chunks

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = <Date 2014-10-04.18:28:06.450>
created_at = <Date 2011-02-21.15:04:48.909>
labels = ['library', 'performance']
title = "concurrent.futures.ProcessPoolExecutor.map() doesn't batch function arguments by chunks"
updated_at = <Date 2014-10-05.00:28:50.713>
user = 'https://bugs.python.org/tbrink'

bugs.python.org fields:

activity = <Date 2014-10-05.00:28:50.713>
actor = 'dan.oreilly'
assignee = 'none'
closed = True
closed_date = <Date 2014-10-04.18:28:06.450>
closer = 'pitrou'
components = ['Library (Lib)']
creation = <Date 2011-02-21.15:04:48.909>
creator = 'tbrink'
dependencies = []
files = ['20825', '20826', '36067', '36184', '36185', '36306', '36352']
hgrepos = []
issue_num = 11271
keywords = ['patch']
message_count = 19.0
messages = ['128963', '128970', '137351', '137359', '137692', '155114', '223802', '223803', '223855', '223966', '224369', '224372', '224451', '225045', '225143', '225214', '228472', '228473', '228514']
nosy_count = 8.0
nosy_names = ['bquinlan', 'pitrou', 'neologix', 'tbrink', 'python-dev', 'sbt', 'josh.r', 'dan.oreilly']
pr_nums = []
priority = 'normal'
resolution = 'fixed'
stage = 'resolved'
status = 'closed'
superseder = None
type = 'performance'
url = 'https://bugs.python.org/issue11271'
versions = ['Python 3.5']

[tbrink]

I tested the new concurrent.futures.ProcessPoolExecutor.map() in 3.2 with the is_prime() function from the documentation example. This was significantly slower than using multiprocessing.Pool.map(). Quick look at the source showed that multiprocessing sends the iterable in chunks to the worker process while futures sends always only one entry of the iterable to the worker.

Functions like is_prime() which finish relatively fast make the communication overhead (at least I guess that is the culprit) very big in comparison.

Attached is a file which demonstrates the problem and a quick workaround. The workaround uses the chunk idea from multiprocessing. The problem is that it requires the iterables passed to map() to have a length and be indexable with a slice. I believe this limitation could be worked around.

[tbrink]

Playing around a bit I wrote the attached implementation which works with all iterables.

[brianquinlan]

On my crappy computer, ProcessPoolExecutor.map adds <3ms of added execution time per call using your test framework.

What is your use case where that is too much?

That being said, I don't have any objections to making improvements.

If you want to pursue this, could you attach a working map_comprison.py?

[tbrink]

I can confirm an overhead of 2 ms to 3 ms using a relatively recent Intel Core i5 CPU.

I (personally) believe these 3 ms to be a pretty big overhead on modern computers and I also believe that it would be relatively simple to reduce it.

I don't have much time at the moment but I will try to produce a complete proof of concept patch for the futures module in the next weeks. I'd be happy to get some comments when it is finished.

[pitrou]

Using your test script fixed (on Python 3.3), I get the following numbers:

Starting multiproc...done in 2.1014609336853027 s.
Starting futures...done in 20.209479093551636 s.
Starting futures "fixed"...done in 2.026125907897949 s.

So there's a 0.2ms overhead per remote function call here (20/(100100000-100000000)).

Can't your chunks() function use itertools.islice()?

Also, the chunksize can't be anything else than 1 by default, since your approach is increasing latency of returning results.

[brianquinlan]

I'm closing this since tbrink didn't respond to pitrou's comments.

[danoreilly]

I'm seeing an even larger difference between multiprocessing.Pool and ProcessPoolExecutor on my machine, with Python 3.4:

Starting multiproc...done in 2.160644769668579 s.
Starting futures...done in 67.953957319259644 s.
Starting futures "fixed"...done in 2.134932041168213 s.

I've updated the initial patch to address the comments Antoine made; the chunksize now defaults to 1, and itertools is used to chunk the input iterables, rather than building a list. Attached is an updated benchmark script:

Starting multiproc...done in 2.2295100688934326 s.
Starting futures...done in 68.5991039276123 s.
Starting futures "fixed" (no chunking)...done in 69.18992304801941 s.
Starting futures "fixed" (with chunking)...done in 2.352942705154419 s.

The new implementation of map has essentially identical performance to the original with chunksize=1, but it performs much better with a larger chunksize provided.

[danoreilly]

Here's a patch that adds the new map implementation from the benchmark script to concurrent.futures.process.

[danoreilly]

I've added new versions of the patch/demonstration that ensures we actually submit all the futures before trying to yield from the returned iterator. Previously we were just returning a generator object when map was called, without actually submitting any futures.

[pitrou]

Thank you for posting this, I'm reopening the issue.

[pitrou]

Thank you for the patch. I've posted some review comments. Two further remarks:

• this makes the ProcessPool API slightly different from the ThreadPool one. I wonder if this is acceptable or not. Thoughts?

• the patch would need unit tests for the additional functionality

[danoreilly]

re: Diverging ThreadPoolExecutor and ProcessPoolExecutor APIs. I thought about this as well. It would of course be possible to make ThreadPoolExecutor's API match, but it would serve no useful purpose that I can think of. I guess we could make the ThreadPoolExecutor API accept the chunksize argument just so the APIs match, but then not actually use it (and document that it's not used, of course). That would make it easier to switch between the two APIs, at least.

On the other hand, the two APIs are slightly different already: ThreadPoolExecutor requires max_workers be supplied, while ProcessPoolExecutor will default to multiprocessing.cpu_count(). So this wouldn't completely be without precedent.

Anyway, I will review your comments on the patch, add unit tests, and re-submit. Thanks!

[danoreilly]

I've attached an updated patch based on your review comments; there's now a unit test with a non-default chunksize, the chunking algorithm is more readable, and the same code path is used no matter what chunksize is provided.

I've also updated the test script to match the implementation changes.

[danoreilly]

Here's an updated patch that adds documentation and Antoine's requested code changes.

[pitrou]

Thanks for the update! This looks basically good, I'll wait a bit to see if other people have comments, otherwise I'll commit.

[danoreilly]

A couple of small updates based on comments from Charles-François Natali:

• Use itertools.chain.from_iterable to "yield from" the result chunks instead of a for loop.

• Add more tests with varying chunksizes.

[python-dev]

New changeset f87c2c4f03da by Antoine Pitrou in branch 'default':
Issue bpo-11271: concurrent.futures.Executor.map() now takes a *chunksize*
https://hg.python.org/cpython/rev/f87c2c4f03da

[pitrou]

Sorry for the delay. I simplified the patch a tiny bit ("yield from" wasn't needed, it was sufficient to result the itertools.chain.from_iterable() result), and committed it. Thank you!

[danoreilly]

Hey, my first committed patch :) Thanks for helping to push this through, Antoine!