Support for multiple protocols or heterogenous protocols

[jacobtomlinson]

I have a use case where the client<=>scheduler communication needs to be proxied, so we use the websocket protocol.

However, doing so means that scheduler<=>worker and worker<=>worker communication must also use websockets as we assume consistency.

This results in performance issues on systems with high-performance networking like Infiniband or NVLink. We usually use UCX for worker<=>worker communication to get the best out of the hardware.

I wanted to start a discussion around a solution for this. The two things that spring to mind are:

• Add support for multiple protocols simultaneously.
• Allow for one protocol for client<=>scheduler and a different one for worker<=>worker.

[pentschev]

It seems to me like supporting different protocols for client<=>scheduler and worker<=>worker would be a lot easier. I'm not familiar with internals of websocket, but it looks like we would need a UCX-based implementation (and any other potential future communication frameworks that are supported) of it to make it work. This new layer would very likely incur in additional overhead that wouldn't really add any extra benefit for worker<=>worker scenario, or at least I can't think of any. Given we're talking about high-performance communications, I would any day rather avoid introducing additional overhead.

[mrocklin]

This is already done I think. Here is a test

@gen_test()
async def test_multiple_listeners():
    with captured_logger(logging.getLogger("distributed.scheduler")) as log:
        async with Scheduler(dashboard_address=":0", protocol=["inproc", "tcp"]) as s:
            async with Worker(s.listeners[0].contact_address) as a:
                async with Worker(s.listeners[1].contact_address) as b:
                    assert a.address.startswith("inproc")
                    assert a.scheduler.address.startswith("inproc")
                    assert b.address.startswith("tcp")
                    assert b.scheduler.address.startswith("tcp")

                    async with Client(s.address, asynchronous=True) as c:
                        futures = c.map(inc, range(20))
                        await wait(futures)

                        # Force inter-worker communication both ways
                        await c.submit(sum, futures, workers=[a.address])
                        await c.submit(len, futures, workers=[b.address])

    log = log.getvalue()
    assert re.search(r"Scheduler at:\s*tcp://", log)
    assert re.search(r"Scheduler at:\s*inproc://", log)

[jacobtomlinson]

Oooh awesome, I didn't know protocol takes a list. I think there would be a need to specify which ports the different protocols listen on and set this from the dask-scheduler CLI. But that's a pretty superficial change if the core work has already been done.

[mrocklin]

Yeah, I did this shortly after the websocket work, probably for the same reasons that you're looking into it now.

As a heads-up, websockets work, but didn't feel bulletproof.

[mrocklin]

I think there would be a need to specify which ports the different protocols listen on and set this from the dask-scheduler CLI.

You'e probably already figured this out by now, but many arguments now accept lists. Also, for complex options I've mostly switched to using dask-spec directly. It's nice to cut out the CLI layer and just focus on the Scheduler class options.