TMCRootManager.cxx

Go to the documentation of this file.

//------------------------------------------------
// The Geant4 Virtual Monte Carlo package
// Copyright (C) 2013 - 2018 Ivana Hrivnacova
// All rights reserved.
//
// For the licensing terms see geant4_vmc/LICENSE.
// Contact: root-vmc@cern.ch
//-------------------------------------------------

 
#include "TMCRootManager.h"
#include "Riostream.h"
#include "TError.h"
#include "TMCAutoLock.h"
#include "TThread.h"
 
#include <atomic>
#include <cstdio>
#include <thread>
#include <vector>
 
namespace {
// Define mutexes per operation which modify shared data
TMCMutex createMutex = TMCMUTEX_INITIALIZER;
TMCMutex deleteMutex = TMCMUTEX_INITIALIZER;
 
// A global counter to assign numbers sequentially
std::atomic<int> global_thread_counter{0};
 
int get_clean_thread_id()
{
   // This variable is unique to each thread.
   // It initializes ONLY the first time this function is called on that thread.
   thread_local int my_id = global_thread_counter++;
   return my_id;
}
 
void threadWorker()
{
   // No arguments passed, but the thread can still get its 0, 1, 2 ID
   std::cout << "Thread " << std::this_thread::get_id() << " assigned itself Custom ID: " << get_clean_thread_id()
             << "\n";
}
 
} // namespace
 
//
// static data, methods
//
 
Int_t TMCRootManager::fgCounter = 0;
Bool_t TMCRootManager::fgDebug = false;
TMCThreadLocal TMCRootManager *TMCRootManager::fgInstance = 0;
 
//_____________________________________________________________________________
TMCRootManager *TMCRootManager::Instance()
{
 
   return fgInstance;
}
 
//
// ctors, dtor
//
 
//_____________________________________________________________________________
TMCRootManager::TMCRootManager(const char *projectName, TMCRootManager::FileMode fileMode, Int_t threadRank)
   : TMCRootManager(projectName, TMCRootManager::kTTree, fileMode, threadRank)
{
}
 
//_____________________________________________________________________________
TMCRootManager::TMCRootManager(const char *projectName, TMCRootManager::StorageMode storageMode,
                               TMCRootManager::FileMode fileMode, Int_t threadRank)
   : fStorageMode(storageMode)
{
   if (fgDebug)
      printf("TMCRootManager::TMCRootManager %p \n", this);
 
   // lock mutex
   TMCAutoLock lk(&createMutex);
 
   // Set Id
   fId = fgCounter;
 
   // Increment counter
   ++fgCounter;
 
   // singleton instance
   if (fgInstance) {
      Fatal("TMCRootManager", "Attempt to create two instances of singleton.");
      return;
   }
 
   fgInstance = this;
 
   // SingleThreaded or MT worker with TTree output or MT main thread
   // open file and create a tree
   OpenFile(projectName, fileMode, threadRank);
 
   // unlock mutex
   lk.unlock();
 
   if (fgDebug)
      printf("Done TMCRootManagerMT::TMCRootManagerMT %p \n", this);
}
 
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
//_____________________________________________________________________________
TMCRootManager::TMCRootManager(std::shared_ptr<RNTParaWriter> sharedWriter)
   : fStorageMode(TMCRootManager::kRNTuple), fParaWriter(std::move(sharedWriter))
{
   // Set Id
   fId = fgCounter;
 
   // Increment counter
   ++fgCounter;
 
   // singleton instance
   if (fgInstance) {
      Fatal("TMCRootManager", "Attempt to create two instances of singleton.");
      return;
   }
 
   fgInstance = this;
}
#endif
 
//_____________________________________________________________________________
TMCRootManager::~TMCRootManager()
{
 
   if (fgDebug)
      printf("TMCRootManager::~TMCRootManager %p \n", this);
 
   // lock mutex
   TMCAutoLock lk(&deleteMutex);
 
   if (fFile && !fIsClosed)
      fFile->Close();
   delete fFile;
 
   --fgCounter;
 
   // unlock mutex
   lk.unlock();
 
   if (fgDebug)
      printf("Done TMCRootManager::~TMCRootManager %p \n", this);
}
 
//
// privatemethods
//
 
//_____________________________________________________________________________
void TMCRootManager::OpenFile(const char *projectName, FileMode fileMode, Int_t threadRank)
{
   TString fileName(projectName);
   if (threadRank > 0) {
      Int_t threadId = get_clean_thread_id();
      fileName += "_";
      fileName += threadId;
   }
   fileName += ".root";
 
   TString treeTitle(projectName);
   treeTitle += " tree";
 
   switch (fileMode) {
   case TMCRootManager::kRead:
      fFile = new TFile(fileName);
      if (fStorageMode == kTTree)
         fTree = (TTree *)fFile->Get(projectName);
      break;
 
   case TMCRootManager::kWrite:
      if (fgDebug)
         printf("Going to create Root file \n");
      fFile = new TFile(fileName, "recreate");
      if (fgDebug)
         printf("Done: file %p \n", fFile);
 
      if (fgDebug)
         printf("Going to create TTree \n");
      if (fStorageMode == kTTree)
         fTree = new TTree(projectName, treeTitle);
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
      if (fStorageMode == kRNTuple) {
         fStorageName = projectName;
         fModel = RNTupleModel::Create();
      }
#endif
      if (fgDebug)
         printf("Done: TTree %p \n", fTree);
      ;
      ;
   }
}
 
//
// public methods
//
 
//_____________________________________________________________________________
void TMCRootManager::Register(const char *name, const char *className, void *objAddress)
{
 
   fFile->cd();
   if (!fTree->GetBranch(name))
      fTree->Branch(name, className, objAddress, 32000, 99);
   else
      fTree->GetBranch(name)->SetAddress(objAddress);
}
 
//_____________________________________________________________________________
void TMCRootManager::Register(const char *name, const char *className, const void *objAddress)
{
 
   Register(name, className, const_cast<void *>(objAddress));
}
 
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
//_____________________________________________________________________________
void TMCRootManager::CreateRNTuple(bool parallelMode, bool workerMode)
{
   if (fStorageMode == kRNTuple) {
      if (!parallelMode) {
         fWriter = RNTupleWriter::Append(std::move(fModel), fStorageName.c_str(), *fFile);
         fEntry = fWriter->GetModel().CreateBareEntry();
         for (auto nameAddress : fNameAddress) {
            fEntry->BindRawPtr(nameAddress.first, nameAddress.second);
         }
      } else {
         if (!workerMode) {
            fParaWriter = std::shared_ptr<RNTParaWriter>(
               RNTupleParallelWriter::Append(std::move(fModel), fStorageName.c_str(), *fFile).release());
         } else {
            fFillContext = fParaWriter->CreateFillContext();
            fEntry = fFillContext->CreateEntry();
            for (auto nameAddress : fNameAddress) {
               fEntry->BindRawPtr(nameAddress.first, nameAddress.second);
            }
         }
      }
   }
}
#endif
 
//_____________________________________________________________________________
void TMCRootManager::Fill()
{
   if (fStorageMode == kTTree) {
      fFile->cd();
      fTree->Fill();
   }
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
   else if (fStorageMode == kRNTuple) {
      if (!fParaWriter) {
         RNTupleFillStatus status;
         fWriter->FillNoFlush(*fEntry, status);
         if (status.ShouldFlushCluster()) {
            // If we are asked to flush, first try to do as much work as possible outside of the critical section:
            // FlushColumns() will flush column data and trigger compression, but not actually write to storage.
            // (A framework may of course also decide to flush more often.)
            fWriter->FlushColumns();
            {
               // FlushCluster() will flush data to the underlying TFile, so it requires synchronization.
               fWriter->FlushCluster();
            }
         }
      } else {
         fFillContext->Fill(*fEntry);
      }
   }
#endif
}
 
//_____________________________________________________________________________
void TMCRootManager::WriteAll()
{
   if (fStorageMode == kTTree) {
      fFile->cd();
      fFile->Write();
   }
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
   else if (fStorageMode == kRNTuple) {
      if (fParaWriter) {
         fEntry.reset();
         fFillContext.reset();
      } else {
         fModel.reset();
         fWriter.reset();
      }
   }
#endif
}
 
//_____________________________________________________________________________
void TMCRootManager::Close()
{
   if (fIsClosed) {
      Error("Close", "The file was already closed.");
      return;
   }
 
#if (ROOT_VERSION_CODE >= ROOT_VERSION(6, 38, 0))
   if (fParaWriter) {
      fParaWriter->CommitDataset();
   }
#endif
 
   fFile->cd();
   fFile->Close();
   fIsClosed = true;
}
 
//_____________________________________________________________________________
void TMCRootManager::WriteAndClose()
{
 
   WriteAll();
   Close();
}
 
//_____________________________________________________________________________
void TMCRootManager::ReadEvent(Int_t i)
{
 
   fTree->GetEntry(i);
}