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 (tbb::task_group &iGroup, T &&iAction)
	asynchronously 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, tbb::task_group &iGroup, 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

edm::SerialTaskQueueChain::SerialTaskQueueChain ( )

inline

Definition at line 34 of file SerialTaskQueueChain.h.

{}

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

inlineexplicit

Definition at line 35 of file SerialTaskQueueChain.h.

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

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

delete

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

template<typename T >

void SerialTaskQueueChain::actionToRun ( T &&  iAction )

private

Definition at line 101 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 60 of file SerialTaskQueueChain.h.

References m_queues.

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

{ return m_queues.size(); }

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

delete

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

inline

Definition at line 43 of file SerialTaskQueueChain.h.

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

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

References m_outstandingTasks.

{ return m_outstandingTasks; }

template<typename T >

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

private

Definition at line 86 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(iGroup, [this, iAction]() mutable { this->actionToRun(iAction); });
    } else {
      auto nextQueue = iQueueIndex + 1;
      m_queues[iQueueIndex]->push(
          iGroup, [this, nextQueue, &iGroup, iAction]() mutable { this->passDownChain(nextQueue, iGroup, iAction); });
    }
  }

template<typename T >

void SerialTaskQueueChain::push	(	tbb::task_group &	iGroup,
		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 75 of file SerialTaskQueueChain.h.

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

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

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

Member Data Documentation

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

private

Definition at line 65 of file SerialTaskQueueChain.h.

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

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

private

Definition at line 64 of file SerialTaskQueueChain.h.

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