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

}  // namespace edm

#endif