edm::SerialTaskQueueChain Class Reference

#include <SerialTaskQueueChain.h>

Public Member Functions
std::size_t	numberOfQueues () const
SerialTaskQueueChain &	operator= (const SerialTaskQueueChain &)=delete
SerialTaskQueueChain &	operator= (SerialTaskQueueChain &&iOld)
unsigned long	outstandingTasks () const
template<typename T >
void	push (T &&iAction)
	asynchronously pushes functor iAction into queue More...
template<typename T >
void	pushAndWait (T &&iAction)
	synchronously pushes functor iAction into queue More...
	SerialTaskQueueChain ()
	SerialTaskQueueChain (const SerialTaskQueueChain &)=delete
	SerialTaskQueueChain (SerialTaskQueueChain &&iOld)
	SerialTaskQueueChain (std::vector< std::shared_ptr< SerialTaskQueue >> iQueues)

Private Member Functions
template<typename T >
void	actionToRun (T &&iAction)
template<typename T >
void	passDownChain (unsigned int iIndex, T &&iAction)

Private Attributes
std::atomic< unsigned long >	m_outstandingTasks {0}
std::vector< std::shared_ptr< SerialTaskQueue > >	m_queues

Detailed Description

Definition at line 32 of file SerialTaskQueueChain.h.

Constructor & Destructor Documentation

SerialTaskQueueChain() [1/4]

edm::SerialTaskQueueChain::SerialTaskQueueChain ( )

inline

Definition at line 34 of file SerialTaskQueueChain.h.

{}

Referenced by actionToRun().

SerialTaskQueueChain() [2/4]

edm::SerialTaskQueueChain::SerialTaskQueueChain ( std::vector< std::shared_ptr< SerialTaskQueue >>  iQueues )

inlineexplicit

Definition at line 35 of file SerialTaskQueueChain.h.

        : m_queues(std::move(iQueues)) {}

SerialTaskQueueChain() [3/4]

edm::SerialTaskQueueChain::SerialTaskQueueChain ( const SerialTaskQueueChain &  )

delete

SerialTaskQueueChain() [4/4]

edm::SerialTaskQueueChain::SerialTaskQueueChain ( SerialTaskQueueChain &&  iOld )

inline

Definition at line 40 of file SerialTaskQueueChain.h.

        : m_queues(std::move(iOld.m_queues)), m_outstandingTasks{iOld.m_outstandingTasks.load()} {}

Member Function Documentation

actionToRun()

template<typename T >

void SerialTaskQueueChain::actionToRun ( T &&  iAction )

private

Definition at line 138 of file SerialTaskQueueChain.h.

                                                    {
    //even if an exception happens we will resume the queues.
    using Queues = std::vector<std::shared_ptr<SerialTaskQueue>>;
    auto sentryAction = [](SerialTaskQueueChain* iChain) {
      auto& vec = iChain->m_queues;
      for (auto it = vec.rbegin() + 1; it != vec.rend(); ++it) {
        (*it)->resume();
      }
      --(iChain->m_outstandingTasks);
    };
 
    std::unique_ptr<SerialTaskQueueChain, decltype(sentryAction)> sentry(this, sentryAction);
    iAction();
  }

References SerialTaskQueueChain().

Referenced by passDownChain(), and push().

numberOfQueues()

std::size_t edm::SerialTaskQueueChain::numberOfQueues ( ) const

inline

Definition at line 71 of file SerialTaskQueueChain.h.

{ return m_queues.size(); }

References m_queues.

Referenced by edm::SharedResourcesAcquirer::numberOfResources().

operator=() [1/2]

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

delete

operator=() [2/2]

SerialTaskQueueChain& edm::SerialTaskQueueChain::operator= ( SerialTaskQueueChain &&  iOld )

inline

Definition at line 43 of file SerialTaskQueueChain.h.

                                                                 {
      m_queues = std::move(iOld.m_queues);
      m_outstandingTasks.store(iOld.m_outstandingTasks.load());
      return *this;
    }

References m_outstandingTasks, m_queues, and eostools::move().

outstandingTasks()

unsigned long edm::SerialTaskQueueChain::outstandingTasks ( ) const

inline

Definition at line 70 of file SerialTaskQueueChain.h.

{ return m_outstandingTasks; }

References m_outstandingTasks.

passDownChain()

template<typename T >

void SerialTaskQueueChain::passDownChain ( unsigned int  iIndex, T &&  iAction  )

private

Definition at line 124 of file SerialTaskQueueChain.h.

                                                                                {
    //Have to be sure the queue associated to this running task
    // does not attempt to start another task
    m_queues[iQueueIndex - 1]->pause();
    //is this the last queue?
    if (iQueueIndex + 1 == m_queues.size()) {
      m_queues[iQueueIndex]->push([this, iAction]() mutable { this->actionToRun(iAction); });
    } else {
      auto nextQueue = iQueueIndex + 1;
      m_queues[iQueueIndex]->push([this, nextQueue, iAction]() mutable { this->passDownChain(nextQueue, iAction); });
    }
  }

References actionToRun(), and m_queues.

Referenced by push().

push()

template<typename T >

void SerialTaskQueueChain::push

(

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 86 of file SerialTaskQueueChain.h.

                                             {
    ++m_outstandingTasks;
    if (m_queues.size() == 1) {
      m_queues[0]->push([this, iAction]() mutable { this->actionToRun(iAction); });
    } else {
      assert(!m_queues.empty());
      m_queues[0]->push([this, iAction]() mutable { this->passDownChain(1, iAction); });
    }
  }

References actionToRun(), cms::cuda::assert(), m_outstandingTasks, m_queues, and passDownChain().

Referenced by edm::EventProcessor::beginLumiAsync(), edm::EventProcessor::handleNextEventForStreamAsync(), edm::Worker::TaskQueueAdaptor::push(), and pushAndWait().

pushAndWait()

template<typename T >

void SerialTaskQueueChain::pushAndWait

(

T &&

iAction

)

synchronously pushes functor iAction into queue

The function will wait until iAction has completed before returning. If another task is already running on the queue, the system is allowed to find another TBB task to execute while waiting for the iAction to finish. In that way the core is not idled while waiting.

Parameters

[in]

iAction

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

Definition at line 97 of file SerialTaskQueueChain.h.

                                                    {
    auto destry = [](tbb::task* iTask) { tbb::task::destroy(*iTask); };
 
    std::unique_ptr<tbb::task, decltype(destry)> waitTask(new (tbb::task::allocate_root()) tbb::empty_task, destry);
    waitTask->set_ref_count(3);
 
    std::exception_ptr ptr;
    auto waitTaskPtr = waitTask.get();
    push([waitTaskPtr, iAction, &ptr]() {
      //must wait until exception ptr would be set
      auto dec = [](tbb::task* iTask) { iTask->decrement_ref_count(); };
      std::unique_ptr<tbb::task, decltype(dec)> sentry(waitTaskPtr, dec);
      // Caught exception is rethrown further below.
      CMS_SA_ALLOW try { iAction(); } catch (...) {
        ptr = std::current_exception();
      }
    });
 
    waitTask->decrement_ref_count();
    waitTask->wait_for_all();
 
    if (ptr) {
      std::rethrow_exception(ptr);
    }
  }

References CMS_SA_ALLOW, TauDecayModes::dec, push(), and TrackValidation_cff::task.

Referenced by edm::Worker::TaskQueueAdaptor::pushAndWait().

Member Data Documentation

m_outstandingTasks

std::atomic<unsigned long> edm::SerialTaskQueueChain::m_outstandingTasks {0}

private

Definition at line 76 of file SerialTaskQueueChain.h.

Referenced by operator=(), outstandingTasks(), and push().

m_queues

std::vector<std::shared_ptr<SerialTaskQueue> > edm::SerialTaskQueueChain::m_queues

private

Definition at line 75 of file SerialTaskQueueChain.h.

Referenced by numberOfQueues(), operator=(), passDownChain(), and push().