.then execution of actions following a future's completion
Hello all, So, first time posting here. I’ve been bothered for a while about the lack of the ability to chain futures in python, such that the next future will execute upon the first’s completion. So I submitted a pr to do this. This would add the .then(self, fn) method to concurrent.futures.Future. Thoughts? -dancollins34 Github PR #5335 bugs.python.org <http://bugs.python.org/> issue #32672
I really don't want to distract Yury with this. Let's consider this (or something that addresses the same need) for 3.8. To be clear this is meant as a feature for concurrent.futures.Future, not for asyncio.Future. (It's a bit confusing since you also change asyncio.) Also to be honest I don't understand the use case *or* the semantics very well. You have some explaining to do... (Also, full links: https://bugs.python.org/issue32672; https://github.com/python/cpython/pull/5335) On Thu, Jan 25, 2018 at 8:38 PM, Daniel Collins <dancollins34@gmail.com> wrote:
-- --Guido van Rossum (python.org/~guido)
I have a simple way to solve this I believe. Why not just expose "_chain_future()" from asyncio/futures.py to the public, instead of copying and pasting parts of it? It already works, being used everywhere in the stdlib, it supports both asyncio and concurrent.futures, it's an easily testable external function (follows the design of asyncio for the better part), it's threadsafe right out of the box and it wouldn't require anything but removing a single underscore and adding documentation. (I always wondered why was it private anyway) It's like the function was meant to be public :-P -- Bar On Fri, Jan 26, 2018, 8:07 AM Guido van Rossum <guido@python.org> wrote:
Yeah, it would be better to use the chain_future call from async directly, but the problems are 1) it would make concurrent dependent on async and 2) if it were public, it would require users to instantiate futures, which they’re not supposed to do. -dancollins34 Sent from my iPhone
So, just going point by point: Yes, absolutely put this off for 3.8. I didn’t know the freeze was so close or I would have put the 3.8 tag on originally. Yes, absolutely it is only meant for concurrent.futures futures, it only changes async where async uses concurrent.futures futures. Here’s a more fleshed out description of the use case: Assume you have two functions. Function a(x: str)->AResult fetches an AResult object from a web resource, function b(y: AResult) performs some computationally heavy work on AResult. Assume you’re calling a 10 times with a threadpoolexecutor with 2 worker theads. If you were to schedule a as future using submit, and b as a callback, the executions would look like this: ExecutorThread: b*10 Worker1: a*5 Worker2: a*5 This only gets worse as more work (b) is scheduled as a callback for the result from a. Now you could resolve this by, instead of submitting b as a callback, submitting the following lambda: lambda x: executor.submit(b, x) But then you wouldn’t have easy access to this new future. You would have to build a lot of boilerplate code to collect that future into some external collection, and this would only get worse the deeper the nesting goes. With this syntax on the other hand, if you run a 10 times using submit, but then run a_fut.then(b) for each future, execution instead looks like this: ExecutorThread: Worker1: a*5 b*5 Worker2: a*5 b*5 You can also do additional depth easily. Suppose you want to run 3 c operations (processes the output of b) for each b operation. Then you could call this like b_fut = a_fut.then(b) for i in range(3): b_fut.then(c) And the execution would look like this: ExecutorThread: Worker1: a*5 b*5 c*15 Worker2: a*5 b*5 c*15 Which would be very difficult to do otherwise, and distributes the load across the workers, while having direct access to the outputs of the calls to c. -dancollins34 Sent from my iPhone
@Bar: I don't know about exposing _chain_future(). Certainly it's overkill for what the OP wants -- their PR only cares about chaining concurrent.future.Future. @Daniel: I present the following simpler solution -- it requires you to explicitly pass the executor, but since 'fn' is being submitted to an executor, EIBTI. def then(executor, future, fn): newf = concurrent.futures.Future() def callback(fut): f = executor.submit(fn, fut) try: newf.set_result(f.result()) except CancelledError: newf.cancel() except Exception as err: newf.set_exception(err) return executor.submit(callback) I can't quite follow your reasoning about worker threads (and did you realize that because of the GIL, Python doesn't actually use multiple cores?). But I suppose it doesn't matter whether I understand that -- your point is that you want the 'fn' function submitted to the executor, not run as a "done callback". And that's reasonable. But modifying so much code just so the Future can know which to executor it belongs so you can make then() a method seems overkill. On Fri, Jan 26, 2018 at 8:54 AM, Daniel Collins <dancollins34@gmail.com> wrote:
-- --Guido van Rossum (python.org/~guido)
@Guido As an aside, my understanding was libraries that fall back to c (Numpy as an example) release the GIL for load heavy operations. But I believe the explanation would hold in the general case if you replace thread with process using a ProcessPoolExecutor, that it would be good to be able to submit a callback function back to the executor. Sent from my iPhone
On Fri, Jan 26, 2018 at 9:20 AM, Daniel Collins <dancollins34@gmail.com> wrote:
Sure, but your explanation didn't mention any of that. And yes, good catch on the last line of my example. :-) Given that the solution is only a few lines -- perhaps it's enough to just add it as an example to the docs, rather than to add it as a new function to concurrent.futures? A doc change can be added to 3.7! -- --Guido van Rossum (python.org/~guido)
That’s very true. I’ll try to keep my terminology more in line with the implementation in the future. The only problem with that, is that the function utilizes methods that are marked in the documentation as exclusively to be called by the executor (set_result, instantiation of future objects) and it would be confusing if a few lines later, a “but you can use them for this” example was provided. -dancollins34 Sent from my iPhone
I really don't want to distract Yury with this. Let's consider this (or something that addresses the same need) for 3.8. To be clear this is meant as a feature for concurrent.futures.Future, not for asyncio.Future. (It's a bit confusing since you also change asyncio.) Also to be honest I don't understand the use case *or* the semantics very well. You have some explaining to do... (Also, full links: https://bugs.python.org/issue32672; https://github.com/python/cpython/pull/5335) On Thu, Jan 25, 2018 at 8:38 PM, Daniel Collins <dancollins34@gmail.com> wrote:
-- --Guido van Rossum (python.org/~guido)
I have a simple way to solve this I believe. Why not just expose "_chain_future()" from asyncio/futures.py to the public, instead of copying and pasting parts of it? It already works, being used everywhere in the stdlib, it supports both asyncio and concurrent.futures, it's an easily testable external function (follows the design of asyncio for the better part), it's threadsafe right out of the box and it wouldn't require anything but removing a single underscore and adding documentation. (I always wondered why was it private anyway) It's like the function was meant to be public :-P -- Bar On Fri, Jan 26, 2018, 8:07 AM Guido van Rossum <guido@python.org> wrote:
Yeah, it would be better to use the chain_future call from async directly, but the problems are 1) it would make concurrent dependent on async and 2) if it were public, it would require users to instantiate futures, which they’re not supposed to do. -dancollins34 Sent from my iPhone
So, just going point by point: Yes, absolutely put this off for 3.8. I didn’t know the freeze was so close or I would have put the 3.8 tag on originally. Yes, absolutely it is only meant for concurrent.futures futures, it only changes async where async uses concurrent.futures futures. Here’s a more fleshed out description of the use case: Assume you have two functions. Function a(x: str)->AResult fetches an AResult object from a web resource, function b(y: AResult) performs some computationally heavy work on AResult. Assume you’re calling a 10 times with a threadpoolexecutor with 2 worker theads. If you were to schedule a as future using submit, and b as a callback, the executions would look like this: ExecutorThread: b*10 Worker1: a*5 Worker2: a*5 This only gets worse as more work (b) is scheduled as a callback for the result from a. Now you could resolve this by, instead of submitting b as a callback, submitting the following lambda: lambda x: executor.submit(b, x) But then you wouldn’t have easy access to this new future. You would have to build a lot of boilerplate code to collect that future into some external collection, and this would only get worse the deeper the nesting goes. With this syntax on the other hand, if you run a 10 times using submit, but then run a_fut.then(b) for each future, execution instead looks like this: ExecutorThread: Worker1: a*5 b*5 Worker2: a*5 b*5 You can also do additional depth easily. Suppose you want to run 3 c operations (processes the output of b) for each b operation. Then you could call this like b_fut = a_fut.then(b) for i in range(3): b_fut.then(c) And the execution would look like this: ExecutorThread: Worker1: a*5 b*5 c*15 Worker2: a*5 b*5 c*15 Which would be very difficult to do otherwise, and distributes the load across the workers, while having direct access to the outputs of the calls to c. -dancollins34 Sent from my iPhone
@Bar: I don't know about exposing _chain_future(). Certainly it's overkill for what the OP wants -- their PR only cares about chaining concurrent.future.Future. @Daniel: I present the following simpler solution -- it requires you to explicitly pass the executor, but since 'fn' is being submitted to an executor, EIBTI. def then(executor, future, fn): newf = concurrent.futures.Future() def callback(fut): f = executor.submit(fn, fut) try: newf.set_result(f.result()) except CancelledError: newf.cancel() except Exception as err: newf.set_exception(err) return executor.submit(callback) I can't quite follow your reasoning about worker threads (and did you realize that because of the GIL, Python doesn't actually use multiple cores?). But I suppose it doesn't matter whether I understand that -- your point is that you want the 'fn' function submitted to the executor, not run as a "done callback". And that's reasonable. But modifying so much code just so the Future can know which to executor it belongs so you can make then() a method seems overkill. On Fri, Jan 26, 2018 at 8:54 AM, Daniel Collins <dancollins34@gmail.com> wrote:
-- --Guido van Rossum (python.org/~guido)
@Guido As an aside, my understanding was libraries that fall back to c (Numpy as an example) release the GIL for load heavy operations. But I believe the explanation would hold in the general case if you replace thread with process using a ProcessPoolExecutor, that it would be good to be able to submit a callback function back to the executor. Sent from my iPhone
On Fri, Jan 26, 2018 at 9:20 AM, Daniel Collins <dancollins34@gmail.com> wrote:
Sure, but your explanation didn't mention any of that. And yes, good catch on the last line of my example. :-) Given that the solution is only a few lines -- perhaps it's enough to just add it as an example to the docs, rather than to add it as a new function to concurrent.futures? A doc change can be added to 3.7! -- --Guido van Rossum (python.org/~guido)
That’s very true. I’ll try to keep my terminology more in line with the implementation in the future. The only problem with that, is that the function utilizes methods that are marked in the documentation as exclusively to be called by the executor (set_result, instantiation of future objects) and it would be confusing if a few lines later, a “but you can use them for this” example was provided. -dancollins34 Sent from my iPhone
participants (3)
-
Bar Harel -
Daniel Collins -
Guido van Rossum