Request to ensure event processing order in parallel RDF execution

[acampove]

Check duplicate issues.

• Checked for duplicates

Description

It seems that AsNumpy (and prossibly other functions) process data in random and inconsistent orders when using multithreading.

Reproducer

from ROOT import EnableImplicitMT, RDataFrame

EnableImplicitMT(10)

rdf=RDataFrame(100_000)
rdf=rdf.Define('a', 'rdfentry_')
arr_dat = rdf.AsNumpy(['a'])['a']

print(arr_dat[:10])
print(arr_dat[10:])

The code above shows me:

(rx_data) [acampove@thinkbook enableMT]$ python test.py
[65000 65001 65002 65003 65004 65005 65006 65007 65008 65009]
[65010 65011 65012 ... 99997 99998 99999]
(rx_data) [acampove@thinkbook enableMT]$
(rx_data) [acampove@thinkbook enableMT]$
(rx_data) [acampove@thinkbook enableMT]$ python test.py
[90000 90001 90002 90003 90004 90005 90006 90007 90008 90009]
[90010 90011 90012 ... 24997 24998 24999]

I expect to see the first 10and the last 10 digits of the 0-100K sequence every time

ROOT version

---

| Welcome to ROOT 6.32.10 https://root.cern |
| (c) 1995-2024, The ROOT Team; conception: R. Brun, F. Rademakers |
| Built for linuxx8664gcc on Feb 12 2025, 01:47:45 |
| From tags/6-32-10@6-32-10 |
| With |
| Try '.help'/'.?', '.demo', '.license', '.credits', '.quit'/'.q' |

Installation method

micromamba

Operating system

Almalinux 9

Additional context

This is a very dangerous bug and needs to be fixed ASAP.

[dpiparo]

Thanks for reaching out. If I understand correctly, you see the order of the output events scrambled when going MT. Please correct me if I am wrong.

RDF relies on a multithreaded runtime for parallelism. As such, the order in which the ranges of events read (or generated) is processed is intrinsically random and not reproducible. This is not a feature of RDF, but in general of task parallelised applications, such as data processing frameworks of HEP experiments.

[guitargeek]

This is not a feature of RDF, but in general of task parallelised applications, such as data processing frameworks of HEP experiments.

Doesn't this depend on how you aggregate the data, i.e. the "action" in RDataFrame jargon?

If you accumulate to histograms, like with Histo1D(), I agree it doesn't matter. But if your action doesn't aggregate, like Snapshot(), Take() or AsNumpy(), then I see how not preserving the event order can lead to danger. For example, if you use RDataFrame to derive a new auxiliary dataset that you then want to match with the original one. That happened to me also a few times in the past, and it always took some time before I remembered to disable IMT. But there is no warning about this whatsoever.

Maybe as a hotfix, we can print a warning when IMT is used with a non-aggregating action? Or generally not use IMT for these actions?

[acampove]

Thanks for reaching out. If I understand correctly, you see the order of the output events scrambled when going MT. Please correct me if I am wrong.

RDF relies on a multithreaded runtime for parallelism. As such, the order in which the ranges of events read (or generated) is processed is intrinsically random and not reproducible. This is not a feature of RDF, but in general of task parallelised applications, such as data processing frameworks of HEP experiments.

Dear @dpiparo

Thanks for your reply. Yes, the scrambling of entries is the problem because it is unexpected/counterintuitive. Adding counterintuitive features to the code will make the code do things that the user does not intend to. The user will get the wrong outputs and will need to either spend the whole day trying to figure out why the plots do not make sense or, in the worst case scenario, the user will never find out and the plots will end up in Nature with wrong data.

The user does not know (or care) about the details of multithreading. For the user (PhD student) multithreading is a magical switch that makes the computer run faster, but otherwise it works exactly the same as not using multithreading.

[ferdymercury]

Related discussion: https://root-forum.cern.ch/t/read-data-from-rdataframe-with-implicit-multithreading-is-scrambled/56890/5?u=ferhue

Maybe doxygen docu could add an explicit note abouy this feature

[acampove]

Maybe as a hotfix, we can print a warning when IMT is used with a non-aggregating action? Or generally not use IMT for these actions?

Dear @guitargeek

What about

from ROOT import Multiprocessing

Multiprocessing(workers=10)

This would just start multiple processes, each doing 1/10th of the file and then putting all the parts together at the end, as it is mentioned here. Or quoting @vepadulano

Just to give you a hint at an obstacle towards this goal, the implicit multithreading capabilities of RDataFrame that you are praising in your comment are simply not achievable in Python today. Parallel libraries, including this one you pointed at, do not parallelise via threads but via processes. This has an inherent overhead (also described in the documentation of that library, but that is just an example and everything in Python works the same way). So overcoming this limitation of the Python language is unfortunately something out of scope for the ROOT project. That is why we offer the performance of native threads via the C++ processing. But, as you describe, it comes at the cost of not offering 100% of the Python features, due to this important Python limitation.

with this we might get added some pandas style apply functionality, which is missing in RDataFrame. At the end, the user could decide if he/she wants the fast mulithreading approach or the user friendly an safer multiprocessing approach.

[acampove]

Related discussion: https://root-forum.cern.ch/t/read-data-from-rdataframe-with-implicit-multithreading-is-scrambled/56890/5?u=ferhue

Maybe doxygen docu could add an explicit note abouy this feature

No, no one reads documentation, I would just fix the design and make it safer rather than document dangerous design.

[guitargeek]

Dear @guitargeek

What about

from ROOT import Multiprocessing

Multiprocessing(workers=10)

This would just start multiple processes, each doing 1/10th of the file and then putting all the parts together at the end, as it is mentioned here.

That will have to be discussed with the RDataFrame developers @vepadulano @martamaja10. I was just chiming in to support your case, which I think is very important.

[dpiparo]

is unexpected/counterintuitive.

Dear Angel,

I understand you and, once again, thank you for reaching out with this report.

It is widespread knowledge that, without significant losses of runtime performance, multithreading intrinsically implies program execution to be non-deterministic. Unfortunately, this is also true for data reductions such as histogramming, due to summing of floating point numbers.

Said that, it might be possible that the multithreading model on which RDF relies becomes better than anything HEP has seen so far, it might just not be very easy, especially not being ready to give away any performance.

You seem to care about the problem a lot: would you have actionable suggestions which could lead to improvements?

PS
You seem to hint at advantages of a multiprocess approach. Having a pool of workers, that in parallel execute tasks, impacts determinism, being those workers processes or threads - unless, once again, some ordering is imposed, at the price of performance.

[acampove]

Dear @dpiparo

You seem to care about the problem a lot: would you have actionable suggestions which could lead to improvements?

My feedback would be to implement a multiprocessing approach (like pandarallel) that would be safer, intuitive and would allow the use of python functions directly, no numba jitting, even if that lowers the performance. The user would later figure out what approach to use, the multithreading or the multiprocessing one.

I am surprised that there hasn't been more complains about this. Think of it, any piece of code that uses AsNumpy is potentially buggy. There are millions of lines of analysis code out there and the user has to always make sure to check that they are not using any of these dangerous features together with multithreading.

I very much doubt the user will be diligent enough to read the documentation and find about this and then check if they are using this combination in the code. A warning might truly help, and not even that has been done.

Cheers.

[vepadulano]

Dear @acampove ,

Thanks for the feedback you provided to the ROOT project in this issue!

Let me step in with an attempt to clarify certain aspects discussed so far and then get to the point of proposing how to mover forwards:

1. Data-chunk-based parallel processing, which is the core and soul of modern HEP data analysis, means that different tasks process different sets of entries/events. The order is by the very nature of the concept of data-chunk-based parallelism, not guaranteed. This has nothing to do with ROOT, C++, multithreading, multiprocessing, Python or RDF. The same concept applies to any piece of software written in any language. It also applies to pandarallel, since you bring that package as an example. I haven't used this tool in particular, but if it just so happens that you used it and you see that the paralell chunks of work result in an ordered output, it is simply because the pandarallel developer decided to do extra work under the hood to ensure that ordering. This, as others have mentioned, is just a choice, which comes at a cost.
2. The fact that RDF, today, does not offer the possibility of performant execution of non-performant Python code is not due per se to the fact that the scheduling for parallel processing is done via multiple threads rather than via multiple processes. That is more of an issue with Python-C++ language interoperability (or better yet dynamically-interpreted vs statically-compiled languages).

The issue you bring up is not completely unknown to the project. In fact, it is an extension of an existing issue #12190 and it is related to other user reports we've seen also on the forum. As such, it is valued and it is important that we do something about it. We can think of ways to ensure that the order of event processing is restored after the execution of the computation graph, which would effectively give you the desired result. As soon as something is done about it, I will make sure to keep you informed. Thanks again for your collaboration!