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"
#include "FWCore/Utilities/interface/thread_safety_macros.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() {
    // Exception has to swallowed in order to avoid throwing from execute(). The user of SerialTaskQueue should handle exceptions within m_action().
    CMS_SA_ALLOW try { this->m_action(); } catch (...) {
    }
    return this->finishedTask();
  }
 
}  // namespace edm
 
#endif