Implementation of the new TPool and ThreadPool classes

build/unix/makepchinput.py

@@ -424,7 +424,7 @@ def printModulesMessageOnScreen(selModules):
 def getExtraHeaders():
    """ Get extra headers which do not fall in other special categories
    """
-   extraHeaders=["ROOT/TSeq.h","ROOT/StringConv.h"]
+   extraHeaders=["ROOT/TSeq.h","ROOT/StringConv.h", "ThreadPool.h", "TPool.h"]
    code = "// Extra headers\n"
    for extraHeader in extraHeaders:
       code += '#include "%s"\n' %extraHeader
@@ -487,53 +487,3 @@ def makePCHInput():
 
 if __name__ == "__main__":
    makePCHInput()
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-

cmake/modules/SearchInstalledSoftware.cmake

@@ -1079,6 +1079,12 @@ if(imt)
   if(NOT builtin_tbb)
     message(STATUS "Looking for TBB")
     find_package(TBB)
+    if(TBB_FOUND)
+      if(${TBB_VERSION} VERSION_LESS 4.3)
+        message(STATUS "TBB version < 4.3. Switching on builtin_tbb option")
+        set(builtin_tbb ON CACHE BOOL "" FORCE)
+      endif()
+    endif()  
     if(NOT TBB_FOUND)
       message(STATUS "TBB not found. Switching on builtin_tbb option")
       set(builtin_tbb ON CACHE BOOL "" FORCE)

core/multiproc/CMakeLists.txt

@@ -4,7 +4,7 @@
 
 set(headers TMPClient.h MPSendRecv.h TProcPool.h TMPWorker.h TPoolWorker.h TPoolProcessor.h MPCode.h PoolUtils.h)
 
-set(sources TMPClient.cxx MPSendRecv.cxx TProcPool.cxx TMPWorker.cxx TPoolWorker.cxx TPoolProcessor.cxx PoolUtils.cxx)
+set(sources TMPClient.cxx MPSendRecv.cxx TProcPool.cxx TMPWorker.cxx TPoolWorker.cxx TPoolProcessor.cxx)
 
 ROOT_GENERATE_DICTIONARY(G__MultiProc ${headers} MODULE MultiProc LINKDEF LinkDef.h)
 

core/multiproc/Module.mk

@@ -25,8 +25,7 @@ MULTIPROCH      := $(MODDIRI)/TMPClient.h $(MODDIRI)/TProcPool.h \
 
 MULTIPROCS      := $(MODDIRS)/TMPClient.cxx $(MODDIRS)/TProcPool.cxx \
                 $(MODDIRS)/TMPWorker.cxx $(MODDIRS)/MPSendRecv.cxx \
-                $(MODDIRS)/TPoolWorker.cxx $(MODDIRS)/TPoolProcessor.cxx \
-                $(MODDIRS)/PoolUtils.cxx
+                $(MODDIRS)/TPoolWorker.cxx $(MODDIRS)/TPoolProcessor.cxx
 
 MULTIPROCO      := $(call stripsrc,$(MULTIPROCS:.cxx=.o))
 
@@ -83,4 +82,3 @@ distclean-$(MODNAME): clean-$(MODNAME)
 		@rm -f $(MULTIPROCDEP) $(MULTIPROCDS) $(MULTIPROCDH) $(MULTIPROCLIB) $(MULTIPROCMAP)
 
 distclean::     distclean-$(MODNAME)
-

core/multiproc/inc/PoolUtils.h

@@ -50,7 +50,46 @@ namespace PoolCode {
 ///
 //////////////////////////////////////////////////////////////////////////
 namespace PoolUtils {
-   TObject *ReduceObjects(const std::vector<TObject *> &objs);
+  //////////////////////////////////////////////////////////////////////////
+  /// Merge collection of TObjects.
+  /// This functor looks for an implementation of the Merge method
+  /// (e.g. TH1F::Merge) and calls it on the objects contained in objs.
+  /// If Merge is not found, a null pointer is returned.
+   template <class T>
+   class ReduceObjects{
+     public:
+     T operator()(const std::vector<T> &objs){
+      static_assert(std::is_constructible<TObject *, T>::value,
+                    "The argument should be a vector of pointers to TObject or derived classes");
+      if(objs.size() == 0)
+         return nullptr;
+
+      if(objs.size() == 1)
+         return objs[0];
+
+      //get first object from objs
+      auto obj = objs[0];
+      //get merge function
+      ROOT::MergeFunc_t merge = obj->IsA()->GetMerge();
+      if(!merge) {
+         std::cerr << "could not find merge method for the TObject\n. Aborting operation.";
+         return nullptr;
+      }
+
+      //put the rest of the objs in a list
+      TList mergelist;
+      unsigned NObjs = objs.size();
+      for(unsigned i=1; i<NObjs; ++i) //skip first object
+         mergelist.Add(objs[i]);
+
+      //call merge
+      merge(obj, &mergelist, nullptr);
+      mergelist.Delete();
+
+      //return result
+      return obj;
+    }
+  };
 }
 
 namespace ROOT {

core/multiproc/inc/TPoolProcessor.h

@@ -1,5 +1,5 @@
 /* @(#)root/multiproc:$Id$ */
-// Author: Enrico Guiraud September 2015
+// Author: Enrico Guiraud September 2015.
 
 /*************************************************************************
  * Copyright (C) 1995-2000, Rene Brun and Fons Rademakers.               *
@@ -227,7 +227,7 @@ void TPoolProcessor<F>::Process(unsigned code, MPCodeBufPair& msg)
       MPSend(GetSocket(), PoolCode::kProcError, reply.data());
       return;
    }
-   
+
    //execute function
    auto res = fProcFunc(reader);
 
@@ -238,12 +238,13 @@ void TPoolProcessor<F>::Process(unsigned code, MPCodeBufPair& msg)
    fProcessedEntries += finish - start;
 
    if(fCanReduce) {
-      fReducedResult = static_cast<decltype(fReducedResult)>(PoolUtils::ReduceObjects({res, fReducedResult})); //TODO try not to copy these into a vector, do everything by ref. std::vector<T&>?
+      PoolUtils::ReduceObjects<TObject *> redfunc;
+      fReducedResult = static_cast<decltype(fReducedResult)>(redfunc({res, fReducedResult})); //TODO try not to copy these into a vector, do everything by ref. std::vector<T&>?
    } else {
       fCanReduce = true;
       fReducedResult = res;
    }
-      
+
    if(fMaxNEntries == fProcessedEntries)
       //we are done forever
       MPSend(GetSocket(), PoolCode::kProcResult, fReducedResult);
@@ -285,7 +286,7 @@ TTree *TPoolProcessor<F>::RetrieveTree(TFile *fp)
             if (!strcmp(key->GetClassName(), "TTree") || !strcmp(key->GetClassName(), "TNtuple"))
                tree = static_cast<TTree*>(fp->Get(key->GetName()));
          }
-      }    
+      }
    } else {
       tree = static_cast<TTree*>(fp->Get(fTreeName.c_str()));
    }

core/multiproc/inc/TPoolWorker.h

@@ -43,7 +43,7 @@ class TPoolWorker : public TMPWorker {
    // we trust that decltype(redfunc(std::vector<decltype(func(args[0]))>)) == decltype(args[0])
    // TODO document somewhere that fReducedResult must have a default ctor
    TPoolWorker(F func, const std::vector<T> &args, R redfunc) :
-      TMPWorker(), fFunc(func), fArgs(std::move(args)), fRedFunc(redfunc),
+      TMPWorker(), fFunc(func), fArgs(args), fRedFunc(redfunc),
       fReducedResult(), fCanReduce(false)
    {}
    ~TPoolWorker() {}