@crusaderky yes, but seceding during the task is not successful completion and does not release resources. In my understanding, seceding is meant to free up the current worker and allow it to execute other tasks. Logically, the resources should be temporarily released as well.

See the example below: the jobs will never complete and the whole cluster will get stuck doing nothing. Is this a related issue or am I using resources incorrectly?

import time

import dask
from dask.distributed import (
    Client,
    LocalCluster,
    get_client,
    secede,
    rejoin,
    as_completed,
)


def subjob(arg):
    print("Starting subjob")

    time.sleep(5)

    print("Subjob done")
    return 0


def job(arg):
    print("Starting job")

    time.sleep(1)  # Wait until all jobs are submitted
    client = get_client()
    futs = [client.submit(subjob, i, resources={"CPU": 1}) for i in range(5)]

    secede()
    rets = client.gather(futs)
    rejoin()

    time.sleep(5)
    print("Job done")

    return 0


if __name__ == "__main__":
    with dask.config.set({"distributed.worker.resources.CPU": 8}):
        cluster = LocalCluster(n_workers=1, processes=False)

        client = Client(cluster)

        futures = [
            client.submit(
                job,
                i,
                resources={"CPU": 1},
            )
            for i in range(16)
        ]

        rets = client.gather(futures)

        print("This will never be reached")

Uh oh!

Seceded tasks don't release resources if they fail #6565

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions