Hanging when using python multiprocessing along with dask.distributed

[YarShev]

What happened:

I encountered hanging when using python multiprocessing along with dask.distributed (nested parallelism);

What you expected to happen:

I wonder whether that should work.

Minimal Complete Verifiable Example:

if __name__ == '__main__':
    from distributed import Client
    client = Client()

    import pandas
    df = pandas.DataFrame([1])
    fut = client.scatter(df)

    def f(obj):
        local_df = client.gather(fut)
        local_df += obj

    from multiprocessing import Pool

    with Pool(5) as p:
        print(p.map(f, [1]))

Environment:

• Dask version: dask 2021.11.2 pyhd8ed1ab_0 conda-forge
• Python version: Python 3.8.12
• Operating System: Ubuntu 20.04.2
• Install method (conda, pip, source): conda

[YarShev]

cc @jsignell, @fjetter

[jsignell]

Ah sorry I missed this ping @YarShev. I suspect that this is not intended to work. Here is my understanding of what you are trying to do is apply a function to data in parallel using a matrix of inputs. I am wondering if client.map might be a simpler approach?

    from distributed import Client
    client = Client()

    import pandas
    df = pandas.DataFrame([1])
   
    def f(obj):
        return df + obj
    
    fut = client.map(f, [1])

From the docstring for client.map it seems like it'll use the worker's pool in this case anyways:

func : callable
    Callable to be scheduled for execution. If ``func`` returns a coroutine, it
    will be run on the main event loop of a worker. Otherwise ``func`` will be
    run in a worker's task executor pool (see ``Worker.executors`` for more
    information.)

[YarShev]

@jsignell, thanks for your answer! Yes, client.map would be a simpler approach. However, I am interested in joint using dask.dsitributed and python multiprocessing. The main question is that why this is not intended to work? Does python multiprocessing pool spawn new dask client processes?

[fjetter]

FWIW there has been an earlier report about this issue with some additional context #5334

[YarShev]

@fjetter, I took a look at the issue but I still don't quite get why that shouldn't work. Do you have any thoughts?

[fjetter]

Haven't found time to debug this, yet. My assumption is that the subprocesses create a new client instance and I am not entirely sure if it is possible for a client to fetch a future that belongs to another client. If that's the problem, recreating future instances with the proper keys, belonging to the new client instances would work around this.
That's just guessing at this time, though

[fjetter]

FWIW, If my analysis above is correct, the client should raise if it encounters an unknown future and not silently block; Or it should even automagically recreate a suitable future but I'd need to look into it further to make a clear assessment.

I assume you insistent on using the python multiprocessing API and it's not just about using dedicated processes, is it? You could also use client.map using a processpool executor and the tasks in there would then use a distributed executor (or whatever you choose)

[YarShev]

FWIW, If my analysis above is correct, the client should raise if it encounters an unknown future and not silently block; Or it should even automagically recreate a suitable future but I'd need to look into it further to make a clear assessment.

I would be grateful to you for a clean explanation. Looking forward to seeing your investigation of this question.

I assume you insistent on using the python multiprocessing API and it's not just about using dedicated processes, is it? You could also use client.map using a processpool executor and the tasks in there would then use a distributed executor (or whatever you choose)

I just wanted to achieve more parallelism using dask and multiprocessing if this is possible and reasonable.