Add `Executor.filter` to concurrent.futures

[cool-RR]

BPO	24195
Nosy	@rhettinger, @pfmoore, @brianquinlan, @ncoghlan, @pitrou, @cool-RR
Files	1.patch issue24195.stoneleaf.01.patch 2.patch filter_example.py

^{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 = None
created_at = <Date 2015-05-14.21:26:41.957>
labels = ['type-feature', 'docs']
title = 'Add `Executor.filter` to concurrent.futures'
updated_at = <Date 2019-05-06.20:06:01.565>
user = 'https://github.com/cool-RR'

bugs.python.org fields:

activity = <Date 2019-05-06.20:06:01.565>
actor = 'bquinlan'
assignee = 'docs@python'
closed = False
closed_date = None
closer = None
components = ['Documentation']
creation = <Date 2015-05-14.21:26:41.957>
creator = 'cool-RR'
dependencies = []
files = ['39373', '39377', '39381', '39417']
hgrepos = []
issue_num = 24195
keywords = ['patch']
message_count = 32.0
messages = ['243217', '243218', '243221', '243240', '243242', '243254', '243256', '243260', '243272', '243274', '243276', '243281', '243296', '243298', '243388', '243390', '243436', '243463', '243493', '243871', '243886', '243888', '243890', '243891', '244914', '245004', '245016', '245017', '245025', '245026', '246029', '341635']
nosy_count = 9.0
nosy_names = ['rhettinger', 'paul.moore', 'bquinlan', 'ncoghlan', 'pitrou', 'jnoller', 'cool-RR', 'docs@python', 'sbt']
pr_nums = []
priority = 'normal'
resolution = None
stage = None
status = 'open'
superseder = None
type = 'enhancement'
url = 'https://bugs.python.org/issue24195'
versions = ['Python 3.6']

[cool-RR]

Executor.filter is to filter what Executor.map is to map.

See Python-ideas thread:

https://groups.google.com/forum/#!topic/python-ideas/EBOC5YCWPyo

Patch attached. I don't know how to run the Python test suite (I'm guessing it involves building Python and I don't know how to do that.) Please tell me whether the tests I wrote pass.

I'm guessing that other than that, the main thing missing in my patch is documentation. If there's agreement that this feature is desirable and the implementation is good, I'll be happy to write the docs. Let me know.

[cool-RR]

Also, I notice Python 3.5 feature freeze is a bit over a week away, and I hope we can get that in so it could go in Python 3.5. (Assuming it goes in at all.)

[pfmoore]

You should add docs for the new method, as well.

[ethanfurman]

Updated the tests (had to use real defs, not lambdas, and the expected results for filter_exception weren't right).

Tests pass.

Get some docs written! :)

(More reviews would also be good. ;)

[brianquinlan]

This feature seems unnecessary to me but couldn't the implementation be simplified to work in terms of map? i.e. (pseudocode):

def filter(self, fn, iterable, timeout=None):
  l = list(iterable)
  return (item for (item, keep) in zip(l, self.map(fn, l, timeout)) if keep)

[cool-RR]

Patch with documentation attached. (I don't know how to concatenate patches, so 2.patch contains only the documentation, while 1.patch has the implementation and the tests (but Ethan's patch is better.))

Brian, regarding your simpler implementation based on Executor.map: It looks good to me, but I'm not sure if it passes the tests. If it does then I don't mind if that's the implementation that would be used.

[pfmoore]

Just as a note - to test a pure Pthon patch like this, you can apply the patch to your installed Python 3.4 installation, and just run the test using that. There should be no need to build your own Python.

python C:\Apps\Python34\Lib\test\test_concurrent_futures.py
test_cancel (main.FutureTests) ... ok
test_cancelled (main.FutureTests) ... ok
test_done (main.FutureTests) ... ok
test_done_callback_already_cancelled (main.FutureTests) ... ok
test_done_callback_already_failed (main.FutureTests) ... ok
test_done_callback_already_successful (main.FutureTests) ... ok
[... etc]

[pitrou]

This feature looks unnecessary to me as well. Adding features has a non-zero cost in maintenance.

[ethanfurman]

Short History:
=============

(Ram Rachum)
What do you think about adding a method: Executor.filter?
I was using something like this:

    my_things = [thing for thing in things if some_condition(thing)]

But the problem was that some_condition took a long time to run waiting on I/O, which is a great candidate for parallelizing with ThreadPoolExecutor. I made it work using Executor.map and some improvizing, but it would be nicer if I could do:

    with concurrent.futures.ThreadPoolExecutor(100) as executor:
        my_things = executor.filter(some_condition, things)

And have the condition run in parallel on all the threads.

(Nick Coughlan)
I think this is sufficiently tricky to get right that it's worth adding filter() as a parallel to the existing map() API.

[pitrou]

(Nick Coughlan)

Nick Coghlan isn't currently on this issue. Unless you were talking about another, separate Nick Coughlan that I don't know of? :)

[pitrou]

But to answer your argument:

I think this is sufficiently tricky to get right that it's worth
adding filter() as a parallel to the existing map() API.

"Tricky to get right" is not a good criterion. The question is whether it's useful or not. Only the OP has had that need AFAIK.

[ethanfurman]

That comment was from an email by Nick, not to Nick. But now I've added him. ;)