Core/src/TRootXTReq.cc

0001
0002 #include "Fireworks/Core/interface/TRootXTReq.h"
0003
0004 #include "TCondition.h"
0005 #include "TThread.h"
0006 #include "TMutex.h"
0007 #include <TSysEvtHandler.h>
0008 #include <TSystem.h>
0009 #include <TTimer.h>
0010
0011 // Bloody root threads do not provide signal delivery.
0012 #include <csignal>
0013
0014 // TRootXTReq
0015
0016 //______________________________________________________________________________
0017 //
0018 // Abstract base-class for delivering cross-thread requests into ROOT
0019 // main thread.
0020 // Sub-classes must implement the Act() method.
0021 // Two methods are available to request execution:
0022 //
0023 // - ShootRequest()
0024 //     Request execution and return immediately.
0025 //     The request object is deleted in the main thread.
0026 //
0027 // - ShootRequestAndWait()
0028 //   Request execution and wait until execution is finished.
0029 //   This way results can be returned to the caller in data-members.
0030 //   It is callers responsibility to delete the request object.
0031 //   It can also be reused.
0032 //
0033 //
0034 // Global queue and locks are implemented via static members of this
0035 // class.
0036 //
0037 // Must be initialized from the main thread like this:
0038 //   TRootXTReq::Bootstrap(TThread::SelfId());
0039
0040 TRootXTReq::lpXTReq_t TRootXTReq::sQueue;
0041 pthread_t TRootXTReq::sRootThread = 0;
0042 TMutex *TRootXTReq::sQueueMutex = nullptr;
0043 TSignalHandler *TRootXTReq::sSigHandler = nullptr;
0044 bool TRootXTReq::sSheduled = false;
0045
0046 //==============================================================================
0047
0048 TRootXTReq::TRootXTReq(const char *n) : m_return_condition(nullptr), mName(n) {}
0049
0050 TRootXTReq::~TRootXTReq() { delete m_return_condition; }
0051
0052 //------------------------------------------------------------------------------
0053
0054 void TRootXTReq::post_request() {
0055   TLockGuard _lck(sQueueMutex);
0056
0057   sQueue.push_back(this);
0058
0059   if (!sSheduled) {
0060     sSheduled = true;
0061     pthread_kill(sRootThread, SIGUSR1);
0062   }
0063 }
0064
0065 void TRootXTReq::ShootRequest() {
0066   // Places request into the queue and requests execution in Rint thread.
0067   // It returns immediately after that, without waiting for execution.
0068   // The request is deleted after execution.
0069
0070   if (m_return_condition) {
0071     delete m_return_condition;
0072     m_return_condition = nullptr;
0073   }
0074
0075   post_request();
0076 }
0077
0078 void TRootXTReq::ShootRequestAndWait() {
0079   // Places request into the queue, requests execution in Rint thread and
0080   // waits for the execution to be completed.
0081   // The request is not deleted after execution as it might carry return
0082   // value.
0083   // The same request can be reused several times.
0084
0085   if (!m_return_condition)
0086     m_return_condition = new TCondition;
0087
0088   m_return_condition->GetMutex()->Lock();
0089
0090   post_request();
0091
0092   m_return_condition->Wait();
0093   m_return_condition->GetMutex()->UnLock();
0094 }
0095
0096 //==============================================================================
0097
0098 class RootSig2XTReqHandler : public TSignalHandler {
0099 private:
0100   class XTReqTimer : public TTimer {
0101   public:
0102     XTReqTimer() : TTimer() {}
0103     ~XTReqTimer() override {}
0104
0105     void FireAway() {
0106       Reset();
0107       gSystem->AddTimer(this);
0108     }
0109
0110     Bool_t Notify() override {
0111       gSystem->RemoveTimer(this);
0112       TRootXTReq::ProcessQueue();
0113       return kTRUE;
0114     }
0115   };
0116
0117   XTReqTimer mTimer;
0118
0119 public:
0120   RootSig2XTReqHandler() : TSignalHandler(kSigUser1), mTimer() { Add(); }
0121   ~RootSig2XTReqHandler() override {}
0122
0123   Bool_t Notify() override {
0124     printf("Usr1 Woof Woof in Root thread! Starting Timer.\n");
0125     mTimer.FireAway();
0126     return kTRUE;
0127   }
0128 };
0129
0130 //------------------------------------------------------------------------------
0131
0132 void TRootXTReq::Bootstrap(pthread_t root_thread) {
0133   static const TString _eh("TRootXTReq::Bootstrap ");
0134
0135   if (sRootThread != 0)
0136     throw _eh + "Already initialized.";
0137
0138   sRootThread = root_thread;
0139   sQueueMutex = new TMutex(kTRUE);
0140   sSigHandler = new RootSig2XTReqHandler;
0141 }
0142
0143 void TRootXTReq::Shutdown() {
0144   static const TString _eh("TRootXTReq::Shutdown ");
0145
0146   if (sRootThread == 0)
0147     throw _eh + "Have not beem initialized.";
0148
0149   // Should lock and drain queue ... or sth.
0150
0151   sRootThread = 0;
0152   delete sSigHandler;
0153   sSigHandler = nullptr;
0154   delete sQueueMutex;
0155   sQueueMutex = nullptr;
0156 }
0157
0158 void TRootXTReq::ProcessQueue() {
0159   printf("Timer fired, processing queue.\n");
0160
0161   while (true) {
0162     TRootXTReq *req = nullptr;
0163     {
0164       TLockGuard _lck(sQueueMutex);
0165
0166       if (!sQueue.empty()) {
0167         req = sQueue.front();
0168         sQueue.pop_front();
0169       } else {
0170         sSheduled = false;
0171         break;
0172       }
0173     }
0174
0175     req->Act();
0176
0177     if (req->m_return_condition) {
0178       req->m_return_condition->GetMutex()->Lock();
0179       req->m_return_condition->Signal();
0180       req->m_return_condition->GetMutex()->UnLock();
0181     } else {
0182       delete req;
0183     }
0184   }
0185 }