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;
      }
   }
}