dask / dask/distributed

Asynchronously executing custom graphs

Open
#3,400 4 comments 0 reactions 0 assignees View on GitHub
Dominant language
Python
Stars
1.7k
Forks
778
Avg merge
2h 50m
Merged PRs (30d)
3

Description

I'm trying to run multiple custom graphs via `client.get(...)` simultaneously but I'm having issues getting this to work.

The native `Futures` interface does not seem to support custom graphs but only executing individual functions.
I've also tried various permutations of the examples provided on the [Asynchronous Operation](https://distributed.dask.org/en/latest/asynchronous.html) section of the documentation (both in the `asyncio`and `torando` flavor), but to no avail.

The following code for instance yields an exception or cancels the tasks depending on where you apply the `asynchronous=True` argument:
```python
import asyncio
import time

from dask.distributed import LocalCluster, Client

def mywait(seconds: int):
print(f"I'm waiting for {seconds} seconds")
time.sleep(seconds)
return seconds

def passthrough(arg):
return arg

def run_asyncio():
cluster = LocalCluster(dashboard_address=22222, processes=True)
# client = Client(cluster)

graph_1 = {"wait 10": (mywait, 10), "end": (passthrough, "wait 10")}
graph_2 = {"wait 20": (mywait, 20), "end": (passthrough, "wait 20")}

async def _run_multiple_graphs(graphs):
client = await Client(cluster, asynchronous=True)
result = await asyncio.wait([client.get(graph, "end") for graph in graphs])
return result

output = asyncio.get_event_loop()\
.run_until_complete(_run_multiple_graphs([graph_1, graph_2]))
print(output)

if __name__ == "__main__":
run_asyncio()
```
As I will be starting computation of the graphs simultaneously a viable workaround is to just cleverly merge `graph_1` and `graph_2` into a single definition but I'm wondering whether that is the only solution.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.