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_group.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 SerialTaskQueue&) = delete;
    const SerialTaskQueue& operator=(const SerialTaskQueue&) = delete;
 
    ~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(tbb::task_group&, const T& iAction);
 
  private:
    class TaskBase {
      friend class SerialTaskQueue;
 
      tbb::task_group* group() { return m_group; }
      virtual void execute() = 0;
 
    public:
      virtual ~TaskBase() = default;
 
    protected:
      explicit TaskBase(tbb::task_group* iGroup) : m_group(iGroup) {}
 
    private:
      tbb::task_group* m_group;
    };
 
    template <typename T>
    class QueuedTask : public TaskBase {
    public:
      QueuedTask(tbb::task_group& iGroup, const T& iAction) : TaskBase(&iGroup), m_action(iAction) {}
 
    private:
      void execute() final;
 
      T m_action;
    };
 
    friend class TaskBase;
 
    void pushTask(TaskBase*);
    TaskBase* pushAndGetNextTask(TaskBase*);
    TaskBase* finishedTask();
    //returns nullptr if a task is already being processed
    TaskBase* pickNextTask();
 
    void spawn(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(tbb::task_group& iGroup, const T& iAction) {
    QueuedTask<T>* pTask{new QueuedTask<T>{iGroup, iAction}};
    pushTask(pTask);
  }
 
  template <typename T>
  void 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 (...) {
    }
  }
 
}  // namespace edm
 
#endif