EIC code displayed by LXR master/​include/​root/​ROOT/​TThreadedObject.hxx	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2025-01-18 10:10:51

 // @(#)root/thread:$Id$
 // Author: Danilo Piparo, CERN  11/2/2016
 
 /*************************************************************************
  * Copyright (C) 1995-2018, Rene Brun and Fons Rademakers.               *
  * All rights reserved.                                                  *
  *                                                                       *
  * For the licensing terms see $ROOTSYS/LICENSE.                         *
  * For the list of contributors see $ROOTSYS/README/CREDITS.             *
  *************************************************************************/
 
 #ifndef ROOT_TThreadedObject
 #define ROOT_TThreadedObject
 
 #include "ROOT/TSpinMutex.hxx"
 #include "TDirectory.h"
 #include "TError.h"
 #include "TList.h"
 #include "TROOT.h"
 
 
 #include <algorithm>
 #include <exception>
 #include <deque>
 #include <functional>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <sstream>
 #include <string>
 #include <thread>
 #include <vector>
 
 class TH1;
 
 namespace ROOT {
 
    /**
     * \class ROOT::TNumSlots
     * \brief Defines the number of threads in some of ROOT's interfaces.
     */
    struct TNumSlots {
       unsigned int fVal; // number of slots
       friend bool operator==(TNumSlots lhs, TNumSlots rhs) { return lhs.fVal == rhs.fVal; }
       friend bool operator!=(TNumSlots lhs, TNumSlots rhs) { return lhs.fVal != rhs.fVal; }
    };
 
    namespace Internal {
 
       namespace TThreadedObjectUtils {
 
 
          template<typename T, bool ISHISTO = std::is_base_of<TH1,T>::value>
          struct Detacher{
             static T* Detach(T* obj) {
                return obj;
             }
          };
 
          template<typename T>
          struct Detacher<T, true>{
             static T* Detach(T* obj) {
                obj->SetDirectory(nullptr);
                obj->ResetBit(kMustCleanup);
                return obj;
             }
          };
 
          /// Return a copy of the object or a "Clone" if the copy constructor is not implemented.
          template<class T, bool isCopyConstructible = std::is_copy_constructible<T>::value>
          struct Cloner {
             static T *Clone(const T *obj, TDirectory* d = nullptr) {
                T* clone;
                if (d){
                   TDirectory::TContext ctxt(d);
                   clone = new T(*obj);
                } else {
                   clone = new T(*obj);
                }
                return Detacher<T>::Detach(clone);
             }
          };
 
          template<class T>
          struct Cloner<T, false> {
             static T *Clone(const T *obj, TDirectory* d = nullptr) {
                T* clone;
                if (d){
                   TDirectory::TContext ctxt(d);
                   clone = (T*)obj->Clone();
                } else {
                   clone = (T*)obj->Clone();
                }
                return clone;
             }
          };
 
          template <class T, bool ISHISTO = std::is_base_of<TH1, T>::value>
          struct DirCreator {
             static TDirectory *Create()
             {
                static unsigned dirCounter = 0;
                const std::string dirName = "__TThreaded_dir_" + std::to_string(dirCounter++) + "_";
                return gROOT->mkdir(dirName.c_str());
             }
          };
 
          template <class T>
          struct DirCreator<T, true> {
             static TDirectory *Create() { return nullptr; }
          };
 
       } // End of namespace TThreadedObjectUtils
    } // End of namespace Internal
 
    namespace TThreadedObjectUtils {
 
       template<class T>
       using MergeFunctionType = std::function<void(std::shared_ptr<T>, std::vector<std::shared_ptr<T>>&)>;
 
       /// Merge TObjects
       template<class T>
       void MergeTObjects(std::shared_ptr<T> target, std::vector<std::shared_ptr<T>> &objs)
       {
          if (!target) return;
          TList objTList;
          // Cannot do better than this
          for (auto obj : objs) {
             if (obj && obj != target) objTList.Add(obj.get());
          }
          target->Merge(&objTList);
       }
    } // end of namespace TThreadedObjectUtils
 
    /**
     * \class ROOT::TThreadedObject
     * \brief A wrapper to make object instances thread private, lazily.
     * \tparam T Class of the object to be made thread private (e.g. TH1F)
     * \ingroup Parallelism
     *
     * A wrapper which makes objects thread private. The methods of the underlying
     * object can be invoked via the arrow operator. The object is created in
     * a specific thread lazily, i.e. upon invocation of one of its methods.
     * The correct object pointer from within a particular thread can be accessed
     * with the overloaded arrow operator or with the Get method.
     * In case an elaborate thread management is in place, e.g. in presence of
     * stream of operations or "processing slots", it is also possible to
     * manually select the correct object pointer explicitly.
     */
    template<class T>
    class TThreadedObject {
    public:
       /// The initial number of empty processing slots that a TThreadedObject is constructed with by default.
       /// Deprecated: TThreadedObject grows as more slots are required.
       static constexpr const TNumSlots fgMaxSlots{64};
 
       TThreadedObject(const TThreadedObject&) = delete;
 
       /// Construct the TThreadedObject with initSlots empty slots and the "model" of the thread private objects.
       /// \param initSlots Set the initial number of slots of the TThreadedObject.
       /// \tparam ARGS Arguments' class type of the constructor of T
       /// \param args variadic arguments
       ///
       /// This form of the constructor is useful to manually pre-set the content of a given number of slots
       /// when used in combination with TThreadedObject::SetAtSlot().
       template <class... ARGS>
       TThreadedObject(TNumSlots initSlots, ARGS &&... args) : fIsMerged(false)
       {
          const auto nSlots = initSlots.fVal;
          fObjPointers.resize(nSlots);
 
          // create at least one directory (we need it for fModel), plus others as needed by the size of fObjPointers
          fDirectories.emplace_back(Internal::TThreadedObjectUtils::DirCreator<T>::Create());
          for (auto i = 1u; i < nSlots; ++i)
             fDirectories.emplace_back(Internal::TThreadedObjectUtils::DirCreator<T>::Create());
 
          TDirectory::TContext ctxt(fDirectories[0]);
          fModel.reset(Internal::TThreadedObjectUtils::Detacher<T>::Detach(new T(std::forward<ARGS>(args)...)));
       }
 
       /// Construct the TThreadedObject and the "model" of the thread private objects.
       /// \tparam ARGS Arguments of the constructor of T
       template<class ...ARGS>
       TThreadedObject(ARGS&&... args) : TThreadedObject(fgMaxSlots, args...) { }
 
       /// Return the number of currently available slot.
       ///
       /// The method is safe to call concurrently to other TThreadedObject methods.
       /// Note that slots could be available but contain no data (i.e. a nullptr) if
       /// they have not been used yet.
       unsigned GetNSlots() const
       {
          std::lock_guard<ROOT::TSpinMutex> lg(fSpinMutex);
          return fObjPointers.size();
       }
 
       /// Access a particular processing slot.
       ///
       /// This method is thread-safe as long as concurrent calls request different slots (i.e. pass a different
       /// argument) and no thread accesses slot `i` via the arrow operator, so mixing usage of GetAtSlot
       /// with usage of the arrow operator can be dangerous.
       std::shared_ptr<T> GetAtSlot(unsigned i)
       {
          std::size_t nAvailableSlots;
          {
             // fObjPointers can grow due to a concurrent operation on this TThreadedObject, need to lock
             std::lock_guard<ROOT::TSpinMutex> lg(fSpinMutex);
             nAvailableSlots = fObjPointers.size();
          }
 
          if (i >= nAvailableSlots) {
             Warning("TThreadedObject::GetAtSlot", "This slot does not exist.");
             return nullptr;
          }
 
          auto &objPointer = fObjPointers[i];
          if (!objPointer)
             objPointer.reset(Internal::TThreadedObjectUtils::Cloner<T>::Clone(fModel.get(), fDirectories[i]));
          return objPointer;
       }
 
       /// Set the value of a particular slot.
       ///
       /// This method is thread-safe as long as concurrent calls access different slots (i.e. pass a different
       /// argument) and no thread accesses slot `i` via the arrow operator, so mixing usage of SetAtSlot
       /// with usage of the arrow operator can be dangerous.
       void SetAtSlot(unsigned i, std::shared_ptr<T> v)
       {
          std::size_t nAvailableSlots;
          {
             // fObjPointers can grow due to a concurrent operation on this TThreadedObject, need to lock
             std::lock_guard<ROOT::TSpinMutex> lg(fSpinMutex);
             nAvailableSlots = fObjPointers.size();
          }
 
          if (i >= nAvailableSlots) {
             Warning("TThreadedObject::SetAtSlot", "This slot does not exist, doing nothing.");
             return;
          }
 
          fObjPointers[i] = v;
       }
 
       /// Access a particular slot which corresponds to a single thread.
       /// This is in general faster than the GetAtSlot method but it is
       /// responsibility of the caller to make sure that the slot exists
       /// and to check that the contained object is initialized (and not
       /// a nullptr).
       std::shared_ptr<T> GetAtSlotUnchecked(unsigned i) const
       {
          return fObjPointers[i];
       }
 
       /// Access a particular slot which corresponds to a single thread.
       /// This overload is faster than the GetAtSlotUnchecked method but
       /// the caller is responsible to make sure that the slot exists, to
       /// check that the contained object is initialized and that the returned
       /// pointer will not outlive the TThreadedObject that returned it, which
       /// maintains ownership of the actual object.
       T* GetAtSlotRaw(unsigned i) const
       {
          return fObjPointers[i].get();
       }
 
       /// Access the pointer corresponding to the current slot. This method is
       /// not adequate for being called inside tight loops as it implies a
       /// lookup in a mapping between the threadIDs and the slot indices.
       /// A good practice consists in copying the pointer onto the stack and
       /// proceed with the loop as shown in this work item (psudo-code) which
       /// will be sent to different threads:
       /// ~~~{.cpp}
       /// auto workItem = [](){
       ///    auto objPtr = tthreadedObject.Get();
       ///    for (auto i : ROOT::TSeqI(1000)) {
       ///       // tthreadedObject->FastMethod(i); // don't do this! Inefficient!
       ///       objPtr->FastMethod(i);
       ///    }
       /// }
       /// ~~~
       std::shared_ptr<T> Get()
       {
          return GetAtSlot(GetThisSlotNumber());
       }
 
       /// Access the wrapped object and allow to call its methods.
       T *operator->()
       {
          return Get().get();
       }
 
       /// Merge all the thread private objects. Can be called once: it does not
       /// create any new object but destroys the present bookkeping collapsing
       /// all objects into the one at slot 0.
       std::shared_ptr<T> Merge(TThreadedObjectUtils::MergeFunctionType<T> mergeFunction = TThreadedObjectUtils::MergeTObjects<T>)
       {
          // We do not return if we already merged.
          if (fIsMerged) {
             Warning("TThreadedObject::Merge", "This object was already merged. Returning the previous result.");
             return fObjPointers[0];
          }
          // need to convert to std::vector because historically mergeFunction requires a vector
          auto vecOfObjPtrs = std::vector<std::shared_ptr<T>>(fObjPointers.begin(), fObjPointers.end());
          mergeFunction(fObjPointers[0], vecOfObjPtrs);
          fIsMerged = true;
          return fObjPointers[0];
       }
 
       /// Merge all the thread private objects. Can be called many times. It
       /// does create a new instance of class T to represent the "Sum" object.
       /// This method is not thread safe: correct or acceptable behaviours
       /// depend on the nature of T and of the merging function.
       std::unique_ptr<T> SnapshotMerge(TThreadedObjectUtils::MergeFunctionType<T> mergeFunction = TThreadedObjectUtils::MergeTObjects<T>)
       {
          if (fIsMerged) {
             Warning("TThreadedObject::SnapshotMerge", "This object was already merged. Returning the previous result.");
             return std::unique_ptr<T>(Internal::TThreadedObjectUtils::Cloner<T>::Clone(fObjPointers[0].get()));
          }
          auto targetPtr = Internal::TThreadedObjectUtils::Cloner<T>::Clone(fModel.get());
          std::shared_ptr<T> targetPtrShared(targetPtr, [](T *) {});
          // need to convert to std::vector because historically mergeFunction requires a vector
          auto vecOfObjPtrs = std::vector<std::shared_ptr<T>>(fObjPointers.begin(), fObjPointers.end());
          mergeFunction(targetPtrShared, vecOfObjPtrs);
          return std::unique_ptr<T>(targetPtr);
       }
 
    private:
       std::unique_ptr<T> fModel;                         ///< Use to store a "model" of the object
       // std::deque's guarantee that references to the elements are not invalidated when appending new slots
       std::deque<std::shared_ptr<T>> fObjPointers;       ///< An object pointer per slot
       // If the object is a histogram, we also create dummy directories that the histogram associates with
       // so we do not pollute gDirectory
       std::deque<TDirectory*> fDirectories;              ///< A TDirectory per slot
       std::map<std::thread::id, unsigned> fThrIDSlotMap; ///< A mapping between the thread IDs and the slots
       mutable ROOT::TSpinMutex fSpinMutex;               ///< Protects concurrent access to fThrIDSlotMap, fObjPointers
       bool fIsMerged : 1;                                ///< Remember if the objects have been merged already
 
       /// Get the slot number for this threadID, make a slot if needed
       unsigned GetThisSlotNumber()
       {
          const auto thisThreadID = std::this_thread::get_id();
          std::lock_guard<ROOT::TSpinMutex> lg(fSpinMutex);
          const auto thisSlotNumIt = fThrIDSlotMap.find(thisThreadID);
          if (thisSlotNumIt != fThrIDSlotMap.end())
             return thisSlotNumIt->second;
          const auto newIndex = fThrIDSlotMap.size();
          fThrIDSlotMap[thisThreadID] = newIndex;
          R__ASSERT(newIndex <= fObjPointers.size() && "This should never happen, we should create new slots as needed");
          if (newIndex == fObjPointers.size()) {
             fDirectories.emplace_back(Internal::TThreadedObjectUtils::DirCreator<T>::Create());
             fObjPointers.emplace_back(nullptr);
          }
          return newIndex;
       }
    };
 
    template<class T>
    constexpr const TNumSlots TThreadedObject<T>::fgMaxSlots;
 
 } // End ROOT namespace
 
 ////////////////////////////////////////////////////////////////////////////////
 /// Print a TThreadedObject at the prompt:
 
 namespace cling {
    template<class T>
    std::string printValue(ROOT::TThreadedObject<T> *val)
    {
       auto model = ((std::unique_ptr<T>*)(val))->get();
       std::ostringstream ret;
       ret << "A wrapper to make object instances thread private, lazily. "
           << "The model which is replicated is " << printValue(model);
       return ret.str();
    }
 }
 
 
 #endif

This page was automatically generated by the 2.3.7 LXR engine. The LXR team