Using different interfaces for scheduler and workers

[kathoef]

I am currently trying to setup a jobqueue cluster in a HPC environment in which available network interfaces on the compute nodes (where Dask workers live) and the login nodes (where the scheduler lives) are not the same.

Available network interfaces

• on the compute nodes there are lo, ib0, but no ib1 and no eth{0,1}
• whereas on the login nodes there are lo, ib1, eth0, and eth1

Starting a jobqueue cluster with a basic workflow of

import dask_jobqueue
import dask.distributed as dask_distributed
jobqueue_cluster = dask_jobqueue.SLURMCluster(cores=6, memory='24GB',
                                              project='esmtst', queue='devel',
                                              interface='ib0')
client = dask_distributed.Client(jobqueue_cluster)
jobqueue_cluster.scale(jobs=1)

causes the following error

---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
<ipython-input-2-b93ee7f2ce8e> in <module>
----> 1 jobqueue_cluster = dask_jobqueue.SLURMCluster(cores=6, memory='24GB',
      2                                               project='esmtst', queue='devel',
      3                                               interface='ib0')

/p/project/cesmtst/hoeflich1/miniconda3/envs/Dask-jobqueue_v2020.02.10/lib/python3.8/site-packages/dask_jobqueue/core.py in __init__(self, n_workers, job_cls, loop, security, silence_logs, name, asynchronous, interface, host, protocol, dashboard_address, config_name, **kwargs)
    446             worker["group"] = ["-" + str(i) for i in range(kwargs["processes"])]
    447
--> 448         self._dummy_job  # trigger property to ensure that the job is valid
    449
    450         super().__init__(

/p/project/cesmtst/hoeflich1/miniconda3/envs/Dask-jobqueue_v2020.02.10/lib/python3.8/site-packages/dask_jobqueue/core.py in _dummy_job(self)
    473         except AttributeError:
    474             address = "tcp://<insert-scheduler-address-here>:8786"
--> 475         return self.job_cls(
    476             address or "tcp://<insert-scheduler-address-here>:8786",
    477             name="name",

/p/project/cesmtst/hoeflich1/miniconda3/envs/Dask-jobqueue_v2020.02.10/lib/python3.8/site-packages/dask_jobqueue/slurm.py in __init__(self, queue, project, walltime, job_cpu, job_mem, job_extra, config_name, *args, **kwargs)
     39             job_extra = dask.config.get("jobqueue.%s.job-extra" % config_name)
     40
---> 41         super().__init__(*args, config_name=config_name, **kwargs)
     42
     43         header_lines = []

/p/project/cesmtst/hoeflich1/miniconda3/envs/Dask-jobqueue_v2020.02.10/lib/python3.8/site-packages/dask_jobqueue/core.py in __init__(self, scheduler, name, cores, memory, processes, nanny, interface, death_timeout, local_directory, extra, env_extra, header_skip, log_directory, shebang, python, job_name, config_name, **kwargs)
    195         if interface:
    196             extra = extra + ["--interface", interface]
--> 197             kwargs.setdefault("host", get_ip_interface(interface))
    198         else:
    199             kwargs.setdefault("host", "")

/p/project/cesmtst/hoeflich1/miniconda3/envs/Dask-jobqueue_v2020.02.10/lib/python3.8/site-packages/distributed/utils.py in get_ip_interface(ifname)
    181     if ifname not in net_if_addrs:
    182         allowed_ifnames = list(net_if_addrs.keys())
--> 183         raise ValueError(
    184             "{!r} is not a valid network interface. "
    185             "Valid network interfaces are: {}".format(ifname, allowed_ifnames)

ValueError: 'ib0' is not a valid network interface. Valid network interfaces are: ['lo', 'ib1', 'eth0', 'eth1']

Workaround as suggested in #207 (comment) doesn't work, because Dask distributed doesn't like interface and host to be set at the same time.

[lesteve]

Workaround as suggested in #207 (comment) doesn't work, because Dask distributed doesn't like interface and host to be set at the same time.

Maybe it worked in the past for some combination of dask-jobqueue and distributed but not anymore.

I think the best fix would be to have a scheduler_options parameter as mentioned in #355 (comment).

[guillaumeeb]

@kathoef thanks for raising this issue. This is a major problem. If @hawk-sf is not up to it in #355, would you be willing to contribute a fix for this issue?

[kathoef]

I think the best fix would be to have a scheduler_options parameter as mentioned in #355 (comment).

Thank you for having pointed this out @lesteve! Last week (by digging through the code in Dask jobqueue and Dask distributed, and with the help of the system admins on login and compute node network locations) I was finally able to make it work. The workaround is to avoid the interface keyword and to set all IP addresses explicitly:

import dask_jobqueue, os
import dask.distributed as dask_distributed
jobqueue_cluster = dask_jobqueue.SLURMCluster(
    cores=6, memory='24GB',
    project='esmtst', queue='esm',
    # correctly passed down to scheduler dictionary
    host=os.environ['HOSTNAME'],
    # using Dask worker low-level options
    extra=['--host ${SLURMD_NODENAME}.ib.juwels.fzj.de'] 
)
client = dask_distributed.Client(jobqueue_cluster)
jobqueue_cluster.scale(jobs=1)

@kathoef thanks for raising this issue. This is a major problem. If @hawk-sf is not up to it in #355, would you be willing to contribute a fix for this issue?

Thank you @guillaumeeb and yes, I could probably do that. (Not before next week, though.)

[lesteve]

FYI in #384, I have a PR to allow to do that. You should be able to do something like this:

import dask_jobqueue, os
import dask.distributed as dask_distributed
jobqueue_cluster = dask_jobqueue.SLURMCluster(
    cores=6, memory='24GB',
    project='esmtst', queue='esm',
    # interface for the workers
    interface='ib0',
    # interface for the scheduler (not sure whether ib1 is the right interface for you)
    scheduler_options={'interface': 'ib1'}
)

client = dask_distributed.Client(jobqueue_cluster)
jobqueue_cluster.scale(jobs=1)

In any case @kathoef if you could double-check this works on your cluster that would be much appreciated. You need to install dask-jobqueue from the PR branch, for example pip install git+https://github.com/lesteve/dask-jobqueue@add-scheduler-options

[kathoef]

Yes, great! I will try this out tomorrow.

[lesteve]

@kathoef have you had the chance to try my snippet on your cluster? I would greatly appreciate if you took the time to do that. Or maybe you could ping someone else that is using dask-jobqueue on your cluster?

If you are busy with other things, I would completely understand, no worries at all! I'll try to remind you in a week or so.

[kathoef]

I am at it now, actually! And I'll report on it before I leave into the weekend. :)

[quasiben]

The ib0 interface is an Infiniband connection correct? Is this what you are ultimately after, using Infiniband ?

[lesteve]

I don't know @kathoef's cluster but I think the convention is that ib<something> is an Infiniband interface indeed.

In my experience, I have used ib<something> in dask-jobqueue just because there is this vague feeling that Infiniband is faster.

I have to admit, I haven't really tested whether that was the case and I don't have any good insights whether at least some things will be faster (e.g. data transfer between Dask scheduler and Dask workers). If you have any insights on this @quasiben that would be greatly appreciated!

[quasiben]

@lesteve I think to take advantage of Infiniband you would either need to write explicit c extensions for IB verbs/RDMA or use something like UCX/UCX-Py. I would not have expected that us interface=ib0 would work at all -- in fact, I would've expected this to fail. Did this at least work for you ?

[lesteve]

Yep interface='ib0' works for me and many others. For example ib0 is used in quite a few of real-world examples: https://jobqueue.dask.org/en/latest/configurations.html