edm::SerialTaskQueue Class Reference

#include <SerialTaskQueue.h>

Classes
class	QueuedTask
class	TaskBase

Public Member Functions
bool	isPaused () const
	Checks to see if the queue has been paused. More...
const SerialTaskQueue &	operator= (const SerialTaskQueue &)=delete
bool	pause ()
	Pauses processing of additional tasks from the queue. More...
template<typename T >
void	push (tbb::task_group &, const T &iAction)
	asynchronously pushes functor iAction into queue More...
bool	resume ()
	Resumes processing if the queue was paused. More...
	SerialTaskQueue ()
	SerialTaskQueue (SerialTaskQueue &&iOther)
	SerialTaskQueue (const SerialTaskQueue &)=delete
	~SerialTaskQueue ()

Private Member Functions
TaskBase *	finishedTask ()
TaskBase *	pickNextTask ()
TaskBase *	pushAndGetNextTask (TaskBase *)
void	pushTask (TaskBase *)
void	spawn (TaskBase &)

Private Attributes
std::atomic< unsigned long >	m_pauseCount
std::atomic< bool >	m_taskChosen
tbb::concurrent_queue< TaskBase * >	m_tasks

Friends
class	TaskBase

Detailed Description

Definition at line 67 of file SerialTaskQueue.h.

Constructor & Destructor Documentation

edm::SerialTaskQueue::SerialTaskQueue ( )

inline

Definition at line 69 of file SerialTaskQueue.h.

: m_taskChosen(false), m_pauseCount{0} {}

edm::SerialTaskQueue::SerialTaskQueue ( SerialTaskQueue &&  iOther )

inline

Definition at line 71 of file SerialTaskQueue.h.

References cms::cuda::assert(), m_taskChosen, and m_tasks.

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

edm::SerialTaskQueue::SerialTaskQueue ( const SerialTaskQueue &  )

delete

SerialTaskQueue::~SerialTaskQueue

(

)

Definition at line 27 of file SerialTaskQueue.cc.

References g, isPaused(), m_taskChosen, m_tasks, or, push(), and GlobalPosition_Frontier_DevDB_cff::tag.

                                  {
  //be certain all tasks have completed
  bool isEmpty = m_tasks.empty();
  bool isTaskChosen = m_taskChosen;
  if ((not isEmpty and not isPaused()) or isTaskChosen) {
    tbb::task_group g;
    g.run([&g, this]() {
      tbb::task::suspend(
          [&g, this](tbb::task::suspend_point tag) { push(g, [tag]() { tbb::task::resume(tag); }); });  //suspend
    });                                                                                                 //group run
    g.wait();
  }
}

Member Function Documentation

SerialTaskQueue::TaskBase * SerialTaskQueue::finishedTask ( )

private

Definition at line 85 of file SerialTaskQueue.cc.

References m_taskChosen, and pickNextTask().

Referenced by spawn().

                                                       {
  m_taskChosen.store(false);
  return pickNextTask();
}

bool edm::SerialTaskQueue::isPaused ( ) const

inline

Checks to see if the queue has been paused.

Returns

true if the queue is paused

See Also

pause(), resume()

Definition at line 87 of file SerialTaskQueue.h.

References m_pauseCount.

Referenced by ~SerialTaskQueue().

{ return m_pauseCount.load() != 0; }

const SerialTaskQueue& edm::SerialTaskQueue::operator= ( const SerialTaskQueue &  )

delete

bool edm::SerialTaskQueue::pause ( )

inline

Pauses processing of additional tasks from the queue.

Any task already running will not be paused however once that running task finishes no further tasks will be started. Multiple calls to pause() are allowed, however each call to pause() must be balanced by a call to resume().

Returns

false if queue was already paused.

See Also

resume(), isPaused()

Definition at line 99 of file SerialTaskQueue.h.

References m_pauseCount.

Referenced by edm::LimitedTaskQueue::Resumer::Resumer().

{ return 1 == ++m_pauseCount; }

SerialTaskQueue::TaskBase * SerialTaskQueue::pickNextTask ( )

private

Definition at line 90 of file SerialTaskQueue.cc.

References LIKELY, m_pauseCount, m_taskChosen, m_tasks, and submitPVValidationJobs::t.

Referenced by finishedTask(), pushAndGetNextTask(), and resume().

                                                       {
  bool expect = false;
  if LIKELY (0 == m_pauseCount and m_taskChosen.compare_exchange_strong(expect, true)) {
    TaskBase* t = nullptr;
    if LIKELY (m_tasks.try_pop(t)) {
      return t;
    }
    //no task was actually pulled
    m_taskChosen.store(false);

    //was a new entry added after we called 'try_pop' but before we did the clear?
    expect = false;
    if (not m_tasks.empty() and m_taskChosen.compare_exchange_strong(expect, true)) {
      t = nullptr;
      if (m_tasks.try_pop(t)) {
        return t;
      }
      //no task was still pulled since a different thread beat us to it
      m_taskChosen.store(false);
    }
  }
  return nullptr;
}

template<typename T >

void SerialTaskQueue::push	(	tbb::task_group &	iGroup,
		const T &	iAction
	)

asynchronously pushes functor iAction into queue

The function will return immediately and iAction will either process concurrently with the calling thread or wait until the protected resource becomes available or until a CPU becomes available.

Parameters

[in]

iAction

Must be a functor that takes no arguments and return no values.

Definition at line 167 of file SerialTaskQueue.h.

References pushTask().

Referenced by edm::EventProcessor::beginLumiAsync(), evf::GlobalEvFOutputEventWriter::doOutputEventAsync(), edm::eventsetup::ESSourceDataProxyBase::prefetchAsyncImpl(), and ~SerialTaskQueue().

                                                                    {
    QueuedTask<T>* pTask{new QueuedTask<T>{iGroup, iAction}};
    pushTask(pTask);
  }

SerialTaskQueue::TaskBase * SerialTaskQueue::pushAndGetNextTask ( TaskBase *  iTask )

private

Definition at line 76 of file SerialTaskQueue.cc.

References LIKELY, m_tasks, and pickNextTask().

Referenced by pushTask().

                                                                            {
  TaskBase* returnValue{nullptr};
  if LIKELY (nullptr != iTask) {
    m_tasks.push(iTask);
    returnValue = pickNextTask();
  }
  return returnValue;
}

void SerialTaskQueue::pushTask ( TaskBase *  iTask )

private

Definition at line 69 of file SerialTaskQueue.cc.

References pushAndGetNextTask(), spawn(), and submitPVValidationJobs::t.

Referenced by push().

                                              {
  auto t = pushAndGetNextTask(iTask);
  if (nullptr != t) {
    spawn(*t);
  }
}

bool SerialTaskQueue::resume

(

)

Resumes processing if the queue was paused.

Multiple calls to resume() are allowed if there were multiple calls to pause(). Only when we reach as many resume() calls as pause() calls will the queue restart.

Returns

true if the call really restarts the queue

See Also

pause(), isPaused()

Definition at line 58 of file SerialTaskQueue.cc.

References m_pauseCount, pickNextTask(), spawn(), and submitPVValidationJobs::t.

Referenced by edm::EventProcessor::globalEndLumiAsync(), edm::LimitedTaskQueue::Resumer::operator=(), and edm::Worker::RunModuleTask< T >::EnableQueueGuard::~EnableQueueGuard().

                             {
  if (0 == --m_pauseCount) {
    auto t = pickNextTask();
    if (nullptr != t) {
      spawn(*t);
    }
    return true;
  }
  return false;
}

void SerialTaskQueue::spawn ( TaskBase &  iTask )

private

Definition at line 41 of file SerialTaskQueue.cc.

References edm::SerialTaskQueue::TaskBase::execute(), finishedTask(), g, edm::SerialTaskQueue::TaskBase::group(), and submitPVValidationJobs::t.

Referenced by pushTask(), and resume().

                                           {
  auto pTask = &iTask;
  iTask.group()->run([pTask, this]() {
    TaskBase* t = pTask;
    auto g = pTask->group();
    do {
      t->execute();
      delete t;
      t = finishedTask();
      if (t and t->group() != g) {
        spawn(*t);
        t = nullptr;
      }
    } while (t != nullptr);
  });
}

Friends And Related Function Documentation

friend class TaskBase

friend

Definition at line 150 of file SerialTaskQueue.h.

Member Data Documentation

std::atomic<unsigned long> edm::SerialTaskQueue::m_pauseCount

private

Definition at line 163 of file SerialTaskQueue.h.

Referenced by isPaused(), pause(), pickNextTask(), and resume().

std::atomic<bool> edm::SerialTaskQueue::m_taskChosen

private

Definition at line 162 of file SerialTaskQueue.h.

Referenced by finishedTask(), pickNextTask(), SerialTaskQueue(), and ~SerialTaskQueue().

tbb::concurrent_queue<TaskBase*> edm::SerialTaskQueue::m_tasks

private

Definition at line 161 of file SerialTaskQueue.h.

Referenced by pickNextTask(), pushAndGetNextTask(), SerialTaskQueue(), and ~SerialTaskQueue().