[df] Add another test for RDF and explicit multithreading

For the following scenario:

* Multiple threads, multiple different RDF computation graphs (one per thread).
* Some threads run a valid, working computation graph, others have some
  problems.
* Two classes of problems are used. The first is when the declaration
  of some code to JIT incurs in an error in the interpreter. The second is when
  there is an exception at runtime.
* All problems are hidden explicitly by a try-catch scope.
* Threads running a working computation graph should not be impacted by the
  threads that are running a flawed computation graph.

Author: vepadulano

Diff for: tree/dataframe/test/dataframe_concurrency.cxx

@@ -3,6 +3,8 @@
 #include <ROOT/RDF/RInterface.hxx>
 #include <ROOT/TThreadExecutor.hxx>
 #include <atomic>
+#include <condition_variable>
+#include <mutex>
 #include <numeric>
 #include <string>
 #include <thread>
@@ -274,3 +276,82 @@ TEST(RDFConcurrency, JITWithManyThreads)
 
    expect_vec_eq(results, expected);
 }
+
+TEST(RDFConcurrency, JITManyThreadsAndExceptions)
+{
+   ROOT::EnableThreadSafety();
+
+   std::condition_variable cv_1;
+   std::condition_variable cv_2;
+   std::condition_variable cv_3;
+   std::mutex m_1;
+   std::mutex m_2;
+   std::mutex m_3;
+   bool ready_1{false};
+   bool ready_2{false};
+   bool ready_3{false};
+
+   int res_work_2{};
+   int res_work_3{};
+
+   auto work_1 = [&]() {
+      std::lock_guard lk_1{m_1};
+      try {
+         ROOT::RDataFrame df{1};
+         auto df1 = df.Define("work_1_x", "throw std::runtime_error(\"Error in RDF!\"); return 42;");
+         auto v = df1.Sum<int>("work_1_x").GetValue();
+      } catch (const std::runtime_error &) {
+      }
+      ready_1 = true;
+      cv_1.notify_all();
+   };
+
+   auto work_2 = [&]() {
+      std::unique_lock lk_1(m_1);
+      std::lock_guard lk_2{m_2};
+      std::unique_lock lk_3{m_3};
+      ROOT::RDataFrame df{1};
+      auto df1 = df.Define("work_2_x", "42");
+      cv_1.wait(lk_1, [&ready_1] { return ready_1; });
+      auto df2 = df1.Define("work_2_y", "58");
+      cv_3.wait(lk_3, [&ready_3] { return ready_3; });
+      auto df3 = df2.Define("work_2_z", "work_2_x + work_2_y");
+      res_work_2 = df3.Sum<int>("work_2_z").GetValue();
+      ready_2 = true;
+      cv_2.notify_one();
+   };
+
+   auto work_3 = [&]() {
+      std::unique_lock lk_1(m_1);
+      std::unique_lock lk_2(m_2);
+      std::unique_lock lk_3(m_3);
+      ROOT::RDataFrame df{1};
+      auto df1 = df.Define("work_3_x", "11");
+      auto df2 = df1.Define("work_3_y", "work_3_x * 2");
+      cv_1.wait(lk_1, [&ready_1] { return ready_1; });
+      cv_2.wait(lk_2, [&ready_2] { return ready_2; });
+      auto df3 = df2.Define("work_3_z", "work_3_y + work_3_x");
+      cv_3.wait(lk_3, [&ready_3] { return ready_3; });
+      auto df4 = df3.Define("work_3_fin", "work_3_x + work_3_y + work_3_z");
+      res_work_3 = df4.Sum<int>("work_3_fin").GetValue();
+   };
+
+   auto work_4 = [&]() {
+      std::lock_guard lk_3{m_3};
+      try {
+         ROOT::RDataFrame df{1};
+         auto df1 = df.Define("x", "rndm");
+      } catch (const std::runtime_error &) {
+      }
+      ready_3 = true;
+      cv_3.notify_all();
+   };
+
+   std::array<std::thread, 4> threads{std::thread{work_1}, std::thread{work_2}, std::thread{work_3},
+                                      std::thread{work_4}};
+   for (auto &&t : threads)
+      t.join();
+
+   EXPECT_EQ(res_work_2, 100);
+   EXPECT_EQ(res_work_3, 66);
+}