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 "oneapi/tbb/task_group.h"
#include "oneapi/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(oneapi::tbb::task_group&, const T& iAction);

  private:
    class TaskBase {
      friend class SerialTaskQueue;

      oneapi::tbb::task_group* group() { return m_group; }
      virtual void execute() = 0;

    public:
      virtual ~TaskBase() = default;

    protected:
      explicit TaskBase(oneapi::tbb::task_group* iGroup) : m_group(iGroup) {}

    private:
      oneapi::tbb::task_group* m_group;
    };

    template <typename T>
    class QueuedTask : public TaskBase {
    public:
      QueuedTask(oneapi::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 --------------------------------
    oneapi::tbb::concurrent_queue<TaskBase*> m_tasks;
    std::atomic<bool> m_taskChosen;
    std::atomic<unsigned long> m_pauseCount;
  };

  template <typename T>
  void SerialTaskQueue::push(oneapi::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