TRootXTReq.cc

Go to the documentation of this file.

 
#include "Fireworks/Core/interface/TRootXTReq.h"
 
#include "TCondition.h"
#include "TThread.h"
#include "TMutex.h"
#include <TSysEvtHandler.h>
#include <TSystem.h>
#include <TTimer.h>
 
// Bloody root threads do not provide signal delivery.
#include <csignal>
 
// TRootXTReq
 
//______________________________________________________________________________
//
// Abstract base-class for delivering cross-thread requests into ROOT
// main thread.
// Sub-classes must implement the Act() method.
// Two methods are available to request execution:
//
// - ShootRequest()
//     Request execution and return immediately.
//     The request object is deleted in the main thread.
//
// - ShootRequestAndWait()
//   Request execution and wait until execution is finished.
//   This way results can be returned to the caller in data-members.
//   It is callers responsibility to delete the request object.
//   It can also be reused.
//
//
// Global queue and locks are implemented via static members of this
// class.
//
// Must be initialized from the main thread like this:
//   TRootXTReq::Bootstrap(TThread::SelfId());
 
TRootXTReq::lpXTReq_t TRootXTReq::sQueue;
pthread_t TRootXTReq::sRootThread = 0;
TMutex *TRootXTReq::sQueueMutex = nullptr;
TSignalHandler *TRootXTReq::sSigHandler = nullptr;
bool TRootXTReq::sSheduled = false;
 
//==============================================================================
 
TRootXTReq::TRootXTReq(const char *n) : m_return_condition(nullptr), mName(n) {}
 
TRootXTReq::~TRootXTReq() { delete m_return_condition; }
 
//------------------------------------------------------------------------------
 
void TRootXTReq::post_request() {
  TLockGuard _lck(sQueueMutex);
 
  sQueue.push_back(this);
 
  if (!sSheduled) {
    sSheduled = true;
    pthread_kill(sRootThread, SIGUSR1);
  }
}
 
void TRootXTReq::ShootRequest() {
  // Places request into the queue and requests execution in Rint thread.
  // It returns immediately after that, without waiting for execution.
  // The request is deleted after execution.
 
  if (m_return_condition) {
    delete m_return_condition;
    m_return_condition = nullptr;
  }
 
  post_request();
}
 
void TRootXTReq::ShootRequestAndWait() {
  // Places request into the queue, requests execution in Rint thread and
  // waits for the execution to be completed.
  // The request is not deleted after execution as it might carry return
  // value.
  // The same request can be reused several times.
 
  if (!m_return_condition)
    m_return_condition = new TCondition;
 
  m_return_condition->GetMutex()->Lock();
 
  post_request();
 
  m_return_condition->Wait();
  m_return_condition->GetMutex()->UnLock();
}
 
//==============================================================================
 
class RootSig2XTReqHandler : public TSignalHandler {
private:
  class XTReqTimer : public TTimer {
  public:
    XTReqTimer() : TTimer() {}
    ~XTReqTimer() override {}
 
    void FireAway() {
      Reset();
      gSystem->AddTimer(this);
    }
 
    Bool_t Notify() override {
      gSystem->RemoveTimer(this);
      TRootXTReq::ProcessQueue();
      return kTRUE;
    }
  };
 
  XTReqTimer mTimer;
 
public:
  RootSig2XTReqHandler() : TSignalHandler(kSigUser1), mTimer() { Add(); }
  ~RootSig2XTReqHandler() override {}
 
  Bool_t Notify() override {
    printf("Usr1 Woof Woof in Root thread! Starting Timer.\n");
    mTimer.FireAway();
    return kTRUE;
  }
};
 
//------------------------------------------------------------------------------
 
void TRootXTReq::Bootstrap(pthread_t root_thread) {
  static const TString _eh("TRootXTReq::Bootstrap ");
 
  if (sRootThread != 0)
    throw _eh + "Already initialized.";
 
  sRootThread = root_thread;
  sQueueMutex = new TMutex(kTRUE);
  sSigHandler = new RootSig2XTReqHandler;
}
 
void TRootXTReq::Shutdown() {
  static const TString _eh("TRootXTReq::Shutdown ");
 
  if (sRootThread == 0)
    throw _eh + "Have not beem initialized.";
 
  // Should lock and drain queue ... or sth.
 
  sRootThread = 0;
  delete sSigHandler;
  sSigHandler = nullptr;
  delete sQueueMutex;
  sQueueMutex = nullptr;
}
 
void TRootXTReq::ProcessQueue() {
  printf("Timer fired, processing queue.\n");
 
  while (true) {
    TRootXTReq *req = nullptr;
    {
      TLockGuard _lck(sQueueMutex);
 
      if (!sQueue.empty()) {
        req = sQueue.front();
        sQueue.pop_front();
      } else {
        sSheduled = false;
        break;
      }
    }
 
    req->Act();
 
    if (req->m_return_condition) {
      req->m_return_condition->GetMutex()->Lock();
      req->m_return_condition->Signal();
      req->m_return_condition->GetMutex()->UnLock();
    } else {
      delete req;
    }
  }
}