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 (std::vector< std::shared_ptr< SerialTaskQueue >> iQueues)
	SerialTaskQueueChain (const SerialTaskQueueChain &)=delete
	SerialTaskQueueChain (SerialTaskQueueChain &&iOld)

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

Constructor & Destructor Documentation

edm::SerialTaskQueueChain::SerialTaskQueueChain ( )

inline

Definition at line 33 of file SerialTaskQueueChain.h.

Referenced by SerialTaskQueueChain().

{}

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

inlineexplicit

Definition at line 34 of file SerialTaskQueueChain.h.

References operator=(), and SerialTaskQueueChain().

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

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

delete

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

inline

Definition at line 39 of file SerialTaskQueueChain.h.

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

Member Function Documentation

template<typename T >

void SerialTaskQueueChain::actionToRun ( T &&  iAction )

private

Definition at line 138 of file SerialTaskQueueChain.h.

Referenced by passDownChain(), and push().

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

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

inline

Definition at line 70 of file SerialTaskQueueChain.h.

References m_queues.

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

{ return m_queues.size(); }

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

delete

Referenced by SerialTaskQueueChain().

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

inline

Definition at line 42 of file SerialTaskQueueChain.h.

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

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

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

inline

Definition at line 69 of file SerialTaskQueueChain.h.

References m_outstandingTasks.

{ return m_outstandingTasks; }

template<typename T >

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

private

Definition at line 124 of file SerialTaskQueueChain.h.

References actionToRun(), and m_queues.

Referenced by push().

                                                                                {
    //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); });
    }
  }

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

References actionToRun(), m_outstandingTasks, m_queues, and passDownChain().

Referenced by edm::EventProcessor::beginLumiAsync(), edm::EventProcessor::handleNextEventForStreamAsync(), operator=(), edm::InputProductResolver::prefetchAsync_(), edm::Worker::TaskQueueAdaptor::push(), pushAndWait(), and edm::ReducedProvenanceReader::ReducedProvenanceReader().

                                             {
    ++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); });
    }
  }

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

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

Referenced by operator=(), and edm::Worker::TaskQueueAdaptor::pushAndWait().

                                                    {
    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);
      try {
        iAction();
      } catch (...) {
        ptr = std::current_exception();
      }
    });

    waitTask->decrement_ref_count();
    waitTask->wait_for_all();

    if (ptr) {
      std::rethrow_exception(ptr);
    }
  }

Member Data Documentation

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

private

Definition at line 75 of file SerialTaskQueueChain.h.

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

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

private

Definition at line 74 of file SerialTaskQueueChain.h.

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