Materializing a future in a worker while the respective data is located in other worker

[YarShev]

Hi there,

I have the following question. Imagine we have two Dask worker processes and two sets of data (for example, two pandas DataFrames), which are located in one of the workers. Suppose we submit a long-running task into the worker, which manages both the DataFrames, whereas we submit another task into other worker, which requires the data from the first worker to execute the task.

from distributed.client import Client
import pandas

client = Client()
df1 = pandas.DataFrame([1, 2, 3])
df2 = pandas.DataFrame([4, 5, 6])
f1 = client.scatter(df1) # suppose we scattered the data into the worker 1
f2 = client.scatter(df2) # suppose we scattered the data into the worker 1 as well

def foo1(df1):
    # long-running stuff with df1

def foo2(df2):
    # some stuff with df2

f3 = client.submit(foo1, f1)
f4 = client.submit(foo2, f2)

My question is that will the second worker wait for completion of the first task in the first worker in order for it to submit required data into the second worker?

Thanks in advance!

[YarShev]

cc @jsignell, @martindurant, @fjetter

[martindurant]

You would expect the two tasks to run simultaneously, which may require copying of data between workers. Dask doesn't like copying data, but it also doesn't like idle CPUs, so it has heuristics in place to decide when to copy and when not. The dashboard will show you what is going on.

[martindurant]

The estimate is based on other similar tasks, if they have been processed. If not, a standin hard-coded value is used for the first time a task of a given type is run.

[YarShev]

The estimate is based on other similar tasks, if they have been processed.

What tasks are supposed to be similar? What criteria are used for determining that? For example, let's say foo1 and foo2 have been processed, are they supposed to be similar?

If not, a standin hard-coded value is used for the first time a task of a given type is run.

Let's say we have a user defined function (imagine it is long running but we don't know that), what a standin hard-coded value is used for the first time of the task?

[YarShev]

@martindurant, just a friendly reminder on my questions above.

[martindurant]

"smiliar" I think means that their dask key start with the same prefix. In many places you can control this value exactly - but for delayed() the standard behaviour uses the name of the function being delayed. In other cases (like the many tasks of a single dataframe operation), all the tasks of that op will automatically have the same prefix.

what a standin hard-coded value is used for the first time of the task

You are very unlikely to need to need to know about this, since tasks are initially assigned in a round-robin fashion amongst the workers, where they start empty. So your function will very likely run at least once. However, you can configure the value yourself by setting

• "distributed.scheduler.default-task-durations" (specific known tasks)
• "distributed.scheduler.unknown-task-duration" (everything else, 500ms)

See distributed.scheduler.SchedulerState.get_task_duration and distributed.stealing.WorkStealing.steal_time_ratio for implementation.

[YarShev]

@martindurant, thanks a lot! Your first statement is clear to me now. I'll take a closer look at your second statement.