Workers on private network, scheduler on a different network - how to make "gather" step work?

[kreczko]

Dear all,

I’ve been reading up the documentation and tried many things, but cannot make it work.

An oversimplified view of my problem can be seen here:

Generally, workers have only a private IP, WAN access and are on a separate network to the scheduler. They are spawned by a separate process, let’s ignore that for now.
The scheduler lives on a node with public IP and one open port to the outside world.

Looking at the Journey of a task, everything works well until Step7: Gather. Since the scheduler has no way to contact the workers directly, this step fails.

I’ve been looking at dask-gateway, and running multiple schedulers (one per cluster) but AFAIK, this also requires connectivity from gateway to scheduler, doesn’t it?

Finally, I had a look at plugins and came across the RabbitMQ example (distributed.dask.org/en/stable/plugins.html#rabbitmq-example).
Having a message queue on the open port could solve the issue if the scheduler could (for the gather step)

1. leave a message for the workers
2. the worker that has the result picks the msg up and pushes the result
3. the scheduler picks up the result from the msg queue

Before I go down that Rabbit hole, I wanted to make sure this is not a solved problem.
I might be simply missing the right keywords to look for or someone solved the problem in a different way.

I know of github.com/comp-dev-cms-ita/dask-remote-jobqueue, but would prefer not getting SSH involved.

Apologies for the improperly formatted links, but new users are only allowed 2

[guillaumeeb]

Hi @kreczko, welcome to Dask Discourse Forum!

@kreczko

Looking at the Journey of a task, everything works well until Step7: Gather. Since the scheduler has no way to contact the workers directly, this step fails.

To be honnest, I’m even surprised it goes this far! Do you happen to have some Error stacktrace to share?

@kreczko

I’ve been looking at dask-gateway, and running multiple schedulers (one per cluster) but AFAIK, this also requires connectivity from gateway to scheduler, doesn’t it?

Yes, I think so.

@kreczko

Before I go down that Rabbit hole, I wanted to make sure this is not a solved problem.

I’m not sure you can achieve what you want with a Workerplugin. The example you are referring to only allows to handle task status changes differently. I don’t know if there is a way of also handling data transfer differently through a WorkerPlugin. But this might be a solution.

Anyways, I also don’t think there is currently a solution to this. Scheduler and Worker are supposed to be able to talk to each other, and with the Client too.

I’m sorry for all the uncertainties in my answer, this is a complex topic, and I hope other with better knowledge can chime in.

[crusaderky]

Hello,
The scheduler needs to be able to contact the workers, and vice versa.
It is generally understood that they should be on the same LAN.
The only connection that is guaranteed to work through a firewall is client->scheduler.

everything works well until Step7: Gather

I too am surprised you got that far.

[kreczko]

Thank you for your replies!

@guillaumeeb

To be honnest, I’m even surprised it goes this far! Do you happen to have some Error stacktrace to share?

@crusaderky

I too am surprised you got that far.

Given that the network communication flow is a black box, anything is possible ;).
Once a connection is established from the worker side, the scheduler and workers can communicate in both directions (TCP). However, once the connection is dropped, the scheduler has no way to contact the workers.

Does anyone know how the communication in Dask works (in the form of an Alice ↔ Bob graph)?

I had a look at the gateway proxy code - in theory, if there is a way to keep the connections until the cluster is shut down, this issue can be fixed.

I’ve also started reading up on a related solution for a different system: the HTCondor Connection Broker:

This is instantiated by the collector to provide connection brokering
service. Daemons (called “targets”) register with the server and keep
a TCP connection to it to exchange messages with the server. Clients
send requests to the server to get a target daemon to do a reverse
connect to the client.

Short version: Provide a server with 1 open port and it will establish connections between nodes that cannot talk to each other otherwise. TCP magic.

[crusaderky]

Network comms in dask use a connection pool. When an actor initiates an RPC call, it picks from the pool. If there are free already-established connections in the pool, they will be reused - regardless of who initiated them. Otherwise, a new one will be added and, once the RPC call is done, it will be added to the pool and reused.
When I say “surprised you got this far” I mean that you were extremely lucky not to hit a race condition where more than one RPC call is necessary at the same time - thus causing the scheduler to initiate a second TCP connection to the worker - until you reach gather.