Hi, When using tags for dask resources, I observe a strange behavior while using mapping. Let's consider the following flow:

map_fn = FunctionTask(lambda x: x + 1, tags=["dask-resource:GPU=1"])

with Flow('test') as flow:
    list = [1]
    map_fn.map(list)

While running such a flow on a dask cluster with 1 scheduler, and 1 worker (with

--resources GPU=1

), it turns out that the flow is blocked because there are 2 tasks that "reserved" the gpu resources (see attached pics from the dask dashboard). It looks like the map itself reserved 1 GPU, and then the map_fn map on element from numbers cannot be processed because dask wait for the resource to be released. Is that the expected behavior of the mapping with dask resources?

Hi @Romain! this is actually expected but I agree it is not ideal; at the moment, Prefect submits a copy of your task to the executor to determine whether it is ready to be mapped and if so, to actually submit the mapped work out via a

worker_client

, which results in 1 extra resource being used than you might otherwise expect. This is up for a refactor very very soon, in which case you will stop seeing this behavior!