Use `concurrent.futures` in local scheduler by jakirkham · Pull Request #6322 · dask/dask

jakirkham · 2020-06-15T09:08:22Z

This adds support for concurrent.futures.Executors in the local scheduler. It also swaps out the existing synchronous, multithreaded, and multiprocessing execution engines for concurrent.futures.Executor-based implementations. Though legacy support for using the old multiprocessing.Pool execution engines remains. Both cases are covered in the threading and multiprocessing tests.

For the synchronous case, we have rolled our own SynchronousExecutor, which simply runs a function immediately when submit is called (storing the result or exception in a Future). This matches the current behavior we have for the synchronous case. It is also easier to implement/maintain than alternatives.

To provide support for the legacy multiprocessing.Pool execution engines, a MultiprocessingPoolExecutor is provided. Whenever a multiprocessing.Pool instance is encountered either in the threading or multiprocessing case, it is wrapped with a MultiprocessingPoolExecutor. The MultiprocessingPoolExecutor provides a very thin layer that does what is necessary to wrap apply_async with a submit call as well as store the result or exception in a Future.

The multithreading and multiprocessing cases are updated to just use the corresponding ThreadPoolExecutor and ProcessPoolExecutor. Docs are also updated to reflect this pattern. Also tests are updated to this usage model as well.

Internally the get_async function looks largely the same, but now takes an Executor.submit method (though any function that looks like it should also work). Passing an apply_async function to it no longer works. User looking to still do that should consider using MultiprocessingPoolExecutor or get_apply_async, which will help wrap calls to apply_async in a concurrent.futures compatible manner. Longer term users should consider migrating to concurrent.futures directly.

In addition to these changes, the local scheduler now batches several tasks per submit call. This is important for cases like multiprocessing where there can be a fair bit of overhead to setting up a process and communicate to/from it. Though this strategy is used in all cases. This results in comparable (if not slightly better) performance in all cases.

Tests added / passed
Passes black dask / flake8 dask

Closes #6220

mrocklin · 2020-06-15T13:45:34Z

Cool. It's nice to see how little had to change here.

jakirkham · 2020-06-15T17:16:11Z

Thanks! Was pleased as well. Didn't take long to do either.

There is still a bit of debugging to figure out some exception handling cases. I think Futures are winding up with the exceptions. Though the logic here with apply_async was to capture those exceptions for serialization. Not sure if we care about that if we switch to concurrent.futures. Thoughts? 🙂

mrocklin · 2020-06-15T17:34:16Z

Though the logic here with apply_async was to capture those exceptions for serialization. Not sure if we care about that if we switch to concurrent.futures. Thoughts?

We would want the same behavior as before. I think that apply_async/get have the same semantics as submit/result, so this shouldn't be hard to acheive I hope.

jakirkham · 2020-06-15T20:24:56Z

Are we wanting to still support ThreadPool and multiprocessing.Pool going forward or is it ok to drop them?

TomAugspurger · 2020-06-16T11:43:26Z

Are we wanting to still support ThreadPool and multiprocessing.Pool going forward or is it ok to drop them?

Is performance roughly the same between the old way and this PR? If so then I don't see a reason to keep the old way around.

jakirkham · 2020-06-16T18:59:46Z

I haven't compared yet.

Since concurrent.futures is designed to allow threads/processes to spin up during use and spin down when unused, the performance could differ on initial runs or workflows with large gaps between runs. Is this something we care about? If not, it is simple to benchmark in a loop with sufficient iterations.

Part of the reason I ask about legacy support is we do have some tests using pools instead, which would need to change somehow. Also in the docs, we mention configuring a pool. Though the docs do state, "some of the scheduler support other keyword arguments". So maybe this is not guaranteed to work? IDK how common this use case is the wild either. Users may just be choosing "threads" or "processes" instead.

Sorry for laboring on this point so much. Just trying to be transparent 🙂

mrocklin · 2020-06-16T19:23:37Z

Personally I'm comfortable breaking the pool argument. I recommend that we hold off of process differences between concurrent.futures and pools until we observe a meaningful difference.

…

On Tue, Jun 16, 2020 at 12:00 PM jakirkham ***@***.***> wrote: I haven't compared yet. Since concurrent.futures is designed to allow threads/processes to spin up during use and spin down when unused, the performance could differ on initial runs or workflows with large gaps between runs. Is this something we care about? If not, it is simple to benchmark in a loop with sufficient iterations. Part of the reason I ask about legacy support is we do have some tests using pools instead, which would need to change somehow. Also in the docs, we mention configuring a pool <https://docs.dask.org/en/latest/scheduling.html#configuration>. Though the docs do state, "some of the scheduler support other keyword arguments". So maybe this is not guaranteed to work? IDK how common this use case is the wild either. Users may just be choosing "threads" or "processes" instead. Sorry for laboring on this point so much. Just trying to be transparent 🙂 — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#6322 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTDT434YFUQDPAD36IDRW66LDANCNFSM4N57EISQ> .

jakirkham · 2020-06-16T19:35:35Z

Personally I'm comfortable breaking the pool argument.

Thanks Matt 🙂

I recommend that we hold off of process differences between
concurrent.futures and pools until we observe a meaningful difference.

I didn't quite follow this bit. Can you please clarify?

mrocklin · 2020-06-16T19:40:33Z

I'm referring to this:

Since concurrent.futures is designed to allow threads/processes to spin

up during use and spin down when unused, the performance could differ on initial runs or workflows with large gaps between runs. Is this something we care about? If not, it is simple to benchmark in a loop with sufficient iterations. My recommendation is to try things out first. I'm guessing that an initial performance test won't take more than 30-60 minutes to run, and that seems like a better use of time than engaging many people in a technical discussion.

…

On Tue, Jun 16, 2020 at 12:35 PM jakirkham ***@***.***> wrote: Personally I'm comfortable breaking the pool argument. Thanks Matt 🙂 I recommend that we hold off of process differences between concurrent.futures and pools until we observe a meaningful difference. I didn't quite follow this bit. Can you please clarify? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#6322 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTCULJEATK3C6IEON5LRW7CRJANCNFSM4N57EISQ> .

jakirkham

Noticed some issues trying to query the num_workers with these concurrent.futures use case as shown below. This seems to be used by some logic in get_async to avoid oversubscription. Given concurrent.futures already has its own logic for avoiding oversubscription, am going to try removing our logic and the num_workers parameter and see if that works. Not expecting any issues. Just trying to document this for transparency.

jakirkham · 2020-06-20T02:36:09Z

So I think this is in a better state now. Remaining things...

A couple of order tests fail
Python 3.6 remains an odd case (may need to skip some tests there)

mrocklin · 2020-06-21T00:34:04Z

So I think this is in a better state now.

Yeah, this looks great! It's cool to see how cleanly this turned out.

A couple of order tests fail

Hrm, that's odd.

Python 3.6 remains an odd case (may need to skip some tests there)

We could also hold off until we drop Python 3.6, which may not be so long.

mrocklin · 2020-06-21T00:35:17Z

The next question I would have would be about performance. I'm curious how many trivial tasks per second we can push through this relative to the old method. I suspect that there are some benchmarks in dask-benchmarks for this, but if not this might also be a good opportunity to add them if you found that interesting.

jakirkham · 2020-06-22T18:12:52Z

A couple of order tests fail

Hrm, that's odd.

Yeah agreed. I'll try to look closer, but would certainly appreciate pointers if anything looks odd here.

Python 3.6 remains an odd case (may need to skip some tests there)

We could also hold off until we drop Python 3.6, which may not be so long.

For context, in Python 3.7 concurrent.futures added support for an mp_context (so multiprocessing.get_context(...) can be used) and initializer (so new processes can run an initialization function). We use both of these features today, which would not work out-of-the-box on Python 3.6.

What we have done is switch to using multiprocessing.set_start_method(...) on Python 3.6 only, which means we can only set the context once in that case. Though that is probably good enough in most practical cases where users just pick spawn or some flavor fork and stick with it (feel free to let me know examples if not). Though there are tests that change the multiprocessing method repeatedly in a session, which are then broken on Python 3.6. I'm thinking we just skip these there.

The initializer part was easier to address, which we have done by just wrapping user function calls to make sure initialization happens on Python 3.6. We can probably tweak this a bit if needed. Though it seems to get the right behavior.

My guess is these workarounds have captured the common use cases on Python 3.6. So it's now mostly a matter of updating the tests accordingly.

The next question I would have would be about performance.

It's certainly a fair question. However I want to hold off on this until we have something that has the behavior we expect. As it stands the order test failures worry me. So would not feel comfortable measuring performance until those are addressed. Otherwise we are merely how fast we can do the wrong thing 😉

mrocklin · 2020-06-22T18:26:41Z

It's certainly a fair question. However I want to hold off on this until

we have something that has the behavior we expect. As it stands the order test failures worry me. So would not feel comfortable measuring performance until those are addressed. Otherwise we are merely how fast we can do the wrong thing 😉 The counter-argument here is that if it's going to be slower then we're going to abandon this approach anyway. Assuming that someone is paying for your time they may want to get some signal that this is a worthwhile endeavor before making sure that everything is just right.

…

On Mon, Jun 22, 2020 at 11:13 AM jakirkham ***@***.***> wrote: A couple of order tests fail Hrm, that's odd. Yeah agreed. I'll try to look closer, but would certainly appreciate pointers if anything looks odd here. Python 3.6 remains an odd case (may need to skip some tests there) We could also hold off until we drop Python 3.6, which may not be so long. For context, in Python 3.7 concurrent.futures added support for an mp_context (so multiprocessing.get_context(...) can be used) and initializer (so new processes can run an initialization function). We use both of these features today, which would not work out-of-the-box on Python 3.6. What we have done is switch to using multiprocessing.set_start_method(...) on Python 3.6 only, which means we can only set the context once in that case. Though that is probably good enough in most practical cases where users just pick spawn or some flavor fork and stick with it (feel free to let me know examples if not). Though there are tests that change the multiprocessing method repeatedly in a session, which are then broken on Python 3.6. I'm thinking we just skip these there. The initializer part was easier to address, which we have done by just wrapping user function calls to make sure initialization happens on Python 3.6. We can probably tweak this a bit if needed. Though it seems to get the right behavior. My guess is these workarounds have captured the common use cases on Python 3.6. So it's now mostly a matter of updating the tests accordingly. The next question I would have would be about performance. It's certainly a fair question. However I want to hold off on this until we have something that has the behavior we expect. As it stands the order test failures worry me. So would not feel comfortable measuring performance until those are addressed. Otherwise we are merely how fast we can do the wrong thing 😉 — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#6322 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTASVSVXJ7X6WOGJEMDRX6NLFANCNFSM4N57EISQ> .

Provides a simple synchronous `Executor` for use with `concurrent.futures`. Simply executes any function once submitted and stores the result or exception in a `Future` returned to the user. This matches closely with our current implementation of synchronous execution. Also it is simpler to implement than say delaying execution until the `Future`'s `.result(...)` is queried. Besides as we already collect and deduplicate all work to do in the local scheduler before submitting it, there is no material difference between computing immediately or when the result is requested.

As any `Exception` (or subclass thereof) that is raised by the callback will be logged instead of allowed to propagate through, we don't get the excepted exception handling behavior with `concurrent.futures` when grabbing the result here. So pass the `Future` instead so that this is delayed until the `Future`'s result is later retrieved. Should allow the exception to be handled in the proper context.

We can just handle this with the `callback` directly. So drop the wrapper function. It was mostly for debugging anyways.

jakirkham · 2021-03-21T03:11:45Z

Have added some logic to make batching optional. Though users can configure this through an argument to get_async or the dask.config.

By default it is off for the SynchronousExecutor and the ThreadPoolExecutor. Those executors do well either way as there is not much overhead involved in setting up workers and submitting work. However for the MultiprocessingPoolExecutor batching has a pretty significant effect. So we have picked 6 as a default since it seems reasonable based on benchmarking, but is not so high that it would lead to issues.

Choose a small, but non-`1` value for batching with multiprocessing. It's helpful to have some batching here. Though `5` is a rough guess based on some benchmarking (and we may want to change that later).

As the `process` case uses a `chunksize=6`, this test will wind up putting all of the work in 1 job. To fix that, update the test to use a `chunksize=1` to ensure each task goes to a new worker.

jakirkham · 2021-03-22T23:21:34Z

There's a few other lingering places where we still use multiprocessing-based Pools just in the tests. Updating those with PR ( #7429 )

jakirkham · 2021-03-22T23:37:30Z

@dask/maintenance it would be good to get some thoughts on this when someone has a chance 🙂

jakirkham · 2021-03-24T19:42:23Z

Planning on merging EOD Friday if no comments

Now that the local scheduler uses `concurrent.futures`, change this test to use a `concurrent.futures` based `Executor` as well. While the `multiprocessing.Pool` would still work, this is the recommended way to go in the future.

jcrist · 2021-03-27T03:48:43Z

Sorry I'm late to seeing this. Does this break the old pool argument, or will manually configuring a pool still work? We make use of this within prefect to handle cancellation in a more robust way, just want to know if I need to change this for the next dask release or not.

jakirkham · 2021-03-27T04:13:09Z

No worries. Figured people might stop by later and am generally happy to deal with any issues that come up 🙂

Yep this was designed to be backwards compatible. So setting a pool works. Also both multiprocessing.Pool based objects and newer concurrent.futures.Executor based objects work with pool. Internally we wrap multiprocessing.Pool objects to look like concurrent.futures.Executors. Additionally we supply a utility function for wrapping apply_async methods or functions to work here as a submit function, which is now used. Hope that helps

Please let us know if you have any other questions

jakirkham · 2021-04-02T19:18:06Z

+        Defaults to 5 as some batching is helpful.
+        If -1, will be computed to evenly divide ready work across workers.
    """
+    chunksize = chunksize or config.get("chunksize", 6)


cc @jcrist (as this may also be relevant to Prefect as it sounds like the Multiprocessing Pool is used)

Opened issue ( #7510 ) about potentially refining this and documenting it better

Thanks for flagging this @jakirkham -- it made it a lot easier to track down where this came from when users started encountering a "lack of parallelism" with trivial flows. I'll be maintaining the Dask executors in Prefect now and would appreciate similar pings in the future if you remember :)

The main debug of this issue can be seen in #7583. I went through the entire series of steps for this where I started with no knowledge of how this worked to now coming up with ideas for how this could perform better 😉 ### The Issues The issues fixed here were introduced in #6322. They can be tested by monitoring/profiling the memory usage from this example: ```python from dask.diagnostics import ResourceProfiler, Profiler, CacheProfiler, visualize import dask.array as da from datetime import datetime if __name__ == "__main__": with Profiler() as prof, ResourceProfiler(dt=0.5) as rprof, CacheProfiler() as cprof: d = da.random.random((5000, 250000), chunks=(1, -1)).sum() d.compute(optimize_graph=False) #visualize( # [rprof, prof, cprof], # file_path=f'/tmp/profile_{datetime.utcnow():%Y%m%d_%H%M%S}.html', # show=False # ) print(max(x.mem for x in rprof.results)) ``` If you uncomment the visualize call then you can see the full profile output for it (see the related issue for screenshots). Ever since #6322 the memory for this example is huge at ~10GB max. Prior to that PR the max memory was ~220MB. ### Fixes This PR mostly reverts us to the behavior before the concurrent.futures PR, but keeps the concurrent.futures API.

mrocklin reviewed Jun 15, 2020

View reviewed changes

Comment thread dask/local.py Outdated

jakirkham commented Jun 19, 2020

View reviewed changes

Comment thread dask/threaded.py

Comment thread dask/multiprocessing.py

jakirkham commented Jun 20, 2020

View reviewed changes

Comment thread dask/tests/test_order.py Outdated

jakirkham force-pushed the use_concurrent_futures branch from a1fac00 to bcc318c Compare June 20, 2020 23:58

tomwhite mentioned this pull request Jun 24, 2020

A concurrent.futures.Executor scheduling interface #6220

Closed

Base automatically changed from master to main March 8, 2021 20:19

jakirkham added 10 commits March 9, 2021 11:35

Create apply_async for concurrent.futures

4946e17

Use synchronous executor internally

9dd09e6

Use ThreadPoolExecutor for multithreading

05a1241

Use ProcessPoolExecutor for multiprocessing

3b854d9

Fix exception handling tests

e4f30ea

Use wrapper function to convert Future to result

a9a4fb1

Handle Futures in queue_get

355a909

Drop callback_wrapper

e52bbbd

We can just handle this with the `callback` directly. So drop the wrapper function. It was mostly for debugging anyways.

Document parameter that was previously missing

ae06e3d

jakirkham force-pushed the use_concurrent_futures branch from ab4a3d9 to 8810902 Compare March 21, 2021 03:27

Provide chunksize for the multiprocessing case

cc61931

Choose a small, but non-`1` value for batching with multiprocessing. It's helpful to have some batching here. Though `5` is a rough guess based on some benchmarking (and we may want to change that later).

jakirkham force-pushed the use_concurrent_futures branch from 8810902 to cc61931 Compare March 21, 2021 03:39

jakirkham added 6 commits March 22, 2021 11:19

Merge dask/main into jakirkham/use_concurrent_futures

63e792f

Merge dask/main into jakirkham/use_concurrent_futures

2c0a488

Use chunksize=1 in test_num_workers_config

9cc54cd

As the `process` case uses a `chunksize=6`, this test will wind up putting all of the work in 1 job. To fix that, update the test to use a `chunksize=1` to ensure each task goes to a new worker.

Update single machine scheduler docs

cd62922

Update shared memory docs

e99e043

Update scheduling docs

3117c50

jakirkham changed the title ~~Use concurrent.futures for scheduling~~ Use concurrent.futures in local scheduler Mar 23, 2021

jakirkham added 4 commits March 26, 2021 11:47

Use concurrent.futures in test_read_text

1778b09

Now that the local scheduler uses `concurrent.futures`, change this test to use a `concurrent.futures` based `Executor` as well. While the `multiprocessing.Pool` would still work, this is the recommended way to go in the future.

test-hdfs

51de9bd

Merge dask/main into jakirkham/use_concurrent_futures

8ec0173

test-hdfs

a595d86

jakirkham merged commit ad0e5d1 into dask:main Mar 27, 2021

jakirkham deleted the use_concurrent_futures branch March 27, 2021 03:37

jakirkham commented Apr 2, 2021

View reviewed changes

clarkzinzow mentioned this pull request Apr 3, 2021

[dask-on-ray] Update Dask-on-Ray scheduler to be compatible with concurrent.futures refactor. ray-project/ray#15094

Closed

crusaderky mentioned this pull request Apr 21, 2021

Regression: poor memory management in threads and sync schedulers #7583

Closed

mgsnuno mentioned this pull request May 19, 2021

LocalDaskExecutor multiprocessing regression due to dask-core==2021.4.0 PrefectHQ/prefect#4537

Closed

pnuu mentioned this pull request Aug 13, 2021

Memory usage has increased drastically pytroll/satpy#1791

Closed

djhoese mentioned this pull request Aug 14, 2021

Fix threaded scheduler memory backpressure regression #8040

Merged

3 tasks

Uh oh!

Uh oh!

Conversation

jakirkham commented Jun 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

mrocklin commented Jun 15, 2020

Uh oh!

jakirkham commented Jun 15, 2020

Uh oh!

mrocklin commented Jun 15, 2020

Uh oh!

jakirkham commented Jun 15, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

TomAugspurger commented Jun 16, 2020

Uh oh!

jakirkham commented Jun 16, 2020

Uh oh!

mrocklin commented Jun 16, 2020 via email

Uh oh!

jakirkham commented Jun 16, 2020

Uh oh!

mrocklin commented Jun 16, 2020 via email

Uh oh!

jakirkham left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

jakirkham commented Jun 20, 2020

Uh oh!

mrocklin commented Jun 21, 2020

Uh oh!

mrocklin commented Jun 21, 2020

Uh oh!

jakirkham commented Jun 22, 2020

Uh oh!

mrocklin commented Jun 22, 2020 via email

Uh oh!

jakirkham commented Mar 21, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jakirkham commented Mar 22, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jakirkham commented Mar 22, 2021

Uh oh!

jakirkham commented Mar 24, 2021

Uh oh!

jcrist commented Mar 27, 2021

Uh oh!

jakirkham commented Mar 27, 2021

Uh oh!

jakirkham Apr 2, 2021

Choose a reason for hiding this comment

Uh oh!

zanieb Jun 1, 2021

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jakirkham commented Jun 15, 2020 •

edited

Loading

jakirkham commented Jun 15, 2020 •

edited

Loading

jakirkham commented Mar 21, 2021 •

edited

Loading

jakirkham commented Mar 22, 2021 •

edited

Loading