Crash when using FromSpec with dask and threads

[acampove]

Check duplicate issues.

• Checked for duplicates

Description

I am seeing a crash when I try to use a Dask client to read files with FromSpec using threads. Processes seem to work.

Reproducer

import json
import ROOT
import numpy
import matplotlib.pyplot as plt

from ROOT             import RDF # type: ignore
from dask.distributed import Client, LocalCluster

NFILES  = 100
PATH    = 'files'
# -------------------------------------
def create_connection(nproc : int):
    cluster = LocalCluster(
        n_workers         =1, 
        threads_per_worker=nproc, 
        processes         =False, 
        memory_limit      ='2GiB')

    client  = Client(cluster)

    return client
# -------------------------------------
def _create_files(index : int):
    main     = f"{PATH}/main_{index:03}.root"
    friend   = f"{PATH}/frnd_{index:03}.root"
    
    df       = ROOT.RDataFrame(100000)
    df.Define("x", "gRandom->Rndm()").Snapshot("Events", main)
    
    df       = ROOT.RDataFrame(100000)
    df.Define("y", "gRandom->Rndm()").Snapshot("Events", friend)
# -------------------------------------
def _get_array(client) -> numpy.ndarray:
    main_files = [ f'{PATH}/main_{index:03}.root' for index in range(NFILES) ]
    frnd_files = [ f'{PATH}/frnd_{index:03}.root' for index in range(NFILES) ]
    
    spec = {
        "samples": {
            "main": {
                "trees": ["Events"],
                "files": main_files,
            }
        },
        "friends": {
            "friend": {
                "trees": ["Events"],
                "files": frnd_files, 
            }
        }
    }
    
    with open("spec.json", "w") as f:
        json.dump(spec, f, indent=4)
    
    df= RDF.Experimental.FromSpec("spec.json", executor = client)

    return df.AsNumpy(['x'])['x']
# -------------------------------------
def main():
    '''
    Entry point
    '''
    for index in range(NFILES):
        _create_files(index = index)

    client = create_connection(nproc = 5)
    arr_1  = _get_array(client)
    arr_2  = _get_array(client)

    plt.scatter(arr_1[:1000], arr_2[:1000])
    plt.savefig('plot.png')  
    plt.close('all')
# ----------------------
if __name__ == '__main__':
    main()
# -------------------------------------

ROOT version

6.38

Installation method

micromamba

Operating system

almalinux9

Additional context

The version using processes (which seems to work) could be a workaround for:

#20840

[vepadulano]

Dear @acampove ,

Thanks for your report!

Using distributed RDataFrame with Python threads, including Dask threads, doesn't seem to provide any advantage. Python threads are limited by the GIL anyway (at least for the foreseeable future). If one wants to use multithreading in RDataFrame, it is simply possible via ROOT.EnableImplicitMT() which will provide the best performance.

Following your issue, I will create a PR to explicitly forbid this mode of operation, I just need to understand how to detect in Dask that the user is trying to use threads vs processes.

[prakharsing7]

Hi @vepadulano, checked worker addresses from client.scheduler_info(). Thread workers run in the same process as the scheduler and get inproc:// addresses while process workers get tcp:// addresses

The workers and guard handles batch clusters (HTCondor, Slurm) where workers haven't registered yet at connection time, in that case we skip the check rather than raising a false positive.

For LocalCluster specifically there is also client.cluster.processes == False, but the scheduler_info() approach should work across all cluster types.

Let me know if the approach is correct, happy to open a PR with this guard added in DaskBackend.init if that's useful.

[vepadulano]

The linked PR was contributed by @JAGANNATHANJP, thanks!