SerialTaskQueue.h

Go to the documentation of this file.

#ifndef FWCore_Concurrency_SerialTaskQueue_h
#define FWCore_Concurrency_SerialTaskQueue_h
// -*- C++ -*-
//
// Package:     Concurrency
// Class  :     SerialTaskQueue
// 
//
// Original Author:  Chris Jones
//         Created:  Thu Feb 21 11:14:39 CST 2013
// $Id$
//

// system include files
#include <atomic>
#include <cassert>

#include "tbb/task.h"
#include "tbb/concurrent_queue.h"

// user include files

// forward declarations
namespace edm {
   class SerialTaskQueue
   {
   public:
      SerialTaskQueue():
      m_taskChosen(false),
      m_pauseCount{0}
      {  }
      
      SerialTaskQueue(SerialTaskQueue&& iOther):
        m_tasks(std::move(iOther.m_tasks)),
        m_taskChosen(iOther.m_taskChosen.exchange(false)),
        m_pauseCount(iOther.m_pauseCount.exchange(0))
      {
        assert(m_tasks.empty() and m_taskChosen == false);
      }
      ~SerialTaskQueue();
      
      // ---------- const member functions ---------------------

      bool isPaused() const { return m_pauseCount.load()!=0;}
      
      // ---------- member functions ---------------------------

      bool pause() {
         return 1 == ++m_pauseCount;
      }
      

      bool resume();
      

      template<typename T>
      void push(const T& iAction);
      

      template<typename T>
      void pushAndWait(const T& iAction);
      

      template<typename T>
      tbb::task* pushAndGetNextTaskToRun(const T& iAction);
      
   private:
      SerialTaskQueue(const SerialTaskQueue&) = delete;
      const SerialTaskQueue& operator=(const SerialTaskQueue&) = delete;
      
      class TaskBase : public tbb::task {
         friend class SerialTaskQueue;
         TaskBase(): m_queue(nullptr) {}
         
      protected:
         tbb::task* finishedTask();
      private:
         void setQueue(SerialTaskQueue* iQueue) { m_queue = iQueue;}
         
         SerialTaskQueue* m_queue;
      };
      
      template< typename T>
      class QueuedTask : public TaskBase {
      public:
         QueuedTask( const T& iAction):
         m_action(iAction) {}
         
      private:
         tbb::task* execute() override;
         
         T m_action;
      };
      
      friend class TaskBase;
      
      void pushTask(TaskBase*);
      tbb::task* pushAndGetNextTask(TaskBase*);
      tbb::task* finishedTask();
      //returns nullptr if a task is already being processed
      TaskBase* pickNextTask();
      
      void pushAndWait(tbb::empty_task* iWait,TaskBase*);
      
      
      // ---------- member data --------------------------------
      tbb::concurrent_queue<TaskBase*> m_tasks;
      std::atomic<bool> m_taskChosen;
      std::atomic<unsigned long> m_pauseCount;
   };
   
   template<typename T>
   void SerialTaskQueue::push(const T& iAction) {
      QueuedTask<T>* pTask{ new (tbb::task::allocate_root()) QueuedTask<T>{iAction} };
      pTask->setQueue(this);
      pushTask(pTask);
   }
   
   template<typename T>
   void SerialTaskQueue::pushAndWait(const T& iAction) {
      tbb::empty_task* waitTask = new (tbb::task::allocate_root()) tbb::empty_task;
      waitTask->set_ref_count(2);
      QueuedTask<T>* pTask{ new (waitTask->allocate_child()) QueuedTask<T>{iAction} };
      pTask->setQueue(this);
      pushAndWait(waitTask,pTask);
   }
   
   template<typename T>
   tbb::task* SerialTaskQueue::pushAndGetNextTaskToRun(const T& iAction) {
      QueuedTask<T>* pTask{ new (tbb::task::allocate_root()) QueuedTask<T>{iAction} };
      pTask->setQueue(this);
      return pushAndGetNextTask(pTask);
   }
   
   inline
   tbb::task*
   SerialTaskQueue::TaskBase::finishedTask() {return m_queue->finishedTask();}
   
   template <typename T>
   tbb::task*
   SerialTaskQueue::QueuedTask<T>::execute() {
      try {
         this->m_action();
      } catch(...) {}
      return this->finishedTask();
   }
   
}

#endif