Implementing a custom worker?

[naquad]

Hi all,

I have a complex use case where I need to integrate a Dask cluster with an existing legacy cluster of workers. These legacy workers consist of a large C++ codebase with their own protocol. My current idea is to create a custom Dask worker that would act as a proxy between the legacy workers and Dask.

I’ve been looking through worker.py and was surprised by the amount of code and the number of components involved. Am I approaching this the wrong way, or is implementing a custom Worker genuinely a high-effort task?

If there is a recommended pattern, extension point, or higher-level abstraction for this kind of integration, I’d really appreciate any pointers or examples.

Thank you.

[jacobtomlinson]

Implementing a custom worker implies you want to use the Dask scheduler for task scheduling still. If you already have a distributed codebase do you really need Dask to do the scheduling?

You haven't described how your workers communicate with each other or coordinate, but assuming they can connect to each other and perform work then you could just use Dask to bootstrap. You could submit your workers as tasks to the Dask workers.

[naquad]

I do need Dask to handle the scheduling: resource tracking, job lifecycle management, retries, in short: distributed computation orchestration.

The networking layer in the legacy system was built in a very residual manner when the time budget was nearly exhausted. It simply dispatches jobs as long as some worker is connected and assumes that some result will eventually be returned. Communication is done via a binary, size-prefixed Protocol Buffers payload over a TCP socket. The role of the proxy would be to translate Dask workloads into this Protocol Buffers protocol and back.

There are organizational constraints that require the existing deployment to remain in place (at least until Q3, ouch!), so replacing it outright would be the right approach but is unfortunately not an option at the moment. The primary issue is the network connectivity.

The new Dask-based implementation relies on the same C++ components as the legacy workers (both use the same shared library; the Dask version interacts with it via pybind11-based bindings). In practice, I’ve found the Dask implementation to be significantly more flexible, controllable, and operationally stable.

So, bottom line: Dask is required for scheduling and orchestration, and the legacy system needs to interoperate with it.

I understand this is not a typical use case, but necessity is the mother of invention and I have it.

[nazeeh111]

Given that Dask must own scheduling and the legacy deployment has to remain in place, I would start with ordinary Dask workers running a thin protocol adapter, rather than subclassing Distributed's full Worker class.

A Dask task on an adapter worker can encode the task as your size-prefixed protobuf message, send it to the legacy endpoint, wait for the matching response, and return the result (or raise on a remote failure). Dask then tracks that adapter task as a normal Future through queued, running, completed, or failed states.

For concurrency, advertise the legacy capacity as an abstract Dask resource on the adapter worker, for example resources={'legacy_slot': N}, and submit each bridge task with resources={'legacy_slot': 1}. The resource docs describe this scheduling model. Keep the advertised count aligned with the number of legacy slots that are actually available; Dask cannot infer the C++ pool's health from an arbitrary resource name.

If opening a TCP connection per task is too expensive, a WorkerPlugin can own a per-worker connection pool with setup/teardown hooks. The task still needs to wait for the remote result so Dask sees the real task lifetime.

One edge to design explicitly: if the legacy server accepts a job but the response is lost, Dask may see a failed worker/task without knowing whether the C++ job ran. A retry can duplicate side effects. Include a stable job ID and deduplication/status lookup in the protocol if possible; otherwise choose an explicit retry policy and reconcile ambiguous outcomes separately.

Because you called out networking as the primary issue, place the adapter where it has a route to the existing TCP service. If no Dask worker can reach that service, adding a Worker subclass alone will not solve the connectivity constraint.

This keeps the legacy cluster and its protocol intact while making Dask responsible for orchestration. A custom Worker subclass looks justified only if you need the legacy executors themselves to register as first-class Dask workers, rather than placing Dask adapters in front of them.