Default and current client ignored in worker

[fjetter]

When calling dask.compute within a task, it is ignoring the default or current Client mechanism and always seems to use a Client that is managed by the Worker

from distributed import Client, LocalCluster, worker_client
from dask.base import get_scheduler


def launch_cluster(id):
    with LocalCluster(n_workers=1, processes=False, dashboard_address=":0") as cluster:
        cluster_client = cluster.get_client()
        # Even if one instantiates the client, it is not picked
        # cluster_client = Client(cluster, set_as_default=True)

        # It will not even pick the correct client if used with the `as_current` contextmanager
        # with cluster_client.as_current():
        compute_default = get_scheduler()
        assert compute_default is cluster_client.get, f"Dask compute selects wrong client. Cluster client {cluster_client} compute client {compute_default}"
        
with Client(n_workers=1, threads_per_worker=1) as c:
    f = c.submit(launch_cluster, 0)
    f.result()

dask.base.get_scheduler is calling into worker.get_client which always prefers the worker client mechanism if no address is provided.

This behavior results in the default client to be rejected, the current client not even considered such that always a new client is instantiated that connects to the scheduler the worker is connected to.

The expected behavior is obviously for dask.compute/get_scheduler to automatically pick the correct Client that is connected to the LocalCluster

[minhnguyenxuan60]

AssertionError: Dask compute selects wrong client. Cluster client <Client: 'inproc://192.168.40.124/15608/1' processes=1 threads=12, memory=15.92 GiB> compute client <bound method Client.get of <Client: 'tcp://127.0.0.1:51776' processes=1 threads=1, memory=15.92 GiB>>

The compute_default returns the correct Client that connected to the LocalCluster (1 threads, 1 process), so I'm a bit confused. Can you clarify more the goal here, and what is the expected behavior? May I ask are we changing the cluster.get_client() output or are we changing the dask.base.get_scheduler()? Thank you!

[fjetter]

The problem is that all APIs that are implicitly requiring a client object (e.g. dask.compute) will automatically pick the worker client without considering default or current client. This is a problem of precedence

The proper precedence should be

1. current
2. default
3. worker client

but instead it is picking worker client first. This is expressed by the assert statement. The example above show three different ways to instantiate or set a client and neither works

• cluster.get_client() is basically equivalent to the line below Client(cluster, set_as_default=True)
• Even if the default client mechanism (which is complicated) would fail us, it should be possible to explicitly set a client "as_current" which should basically override everything else.

High level we're interested in dask.base.get_scheduler returning the correct result. This is not happening because it is deferring to worker.get_client which prefers the worker client (which is pointing to the wrong cluster).

I suspect that fixing this is done by changing the precedence of worker.get_client