#include <LimitedTaskQueue.h>

Classes
class	Resumer

Public Member Functions
unsigned int	concurrencyLimit () const

	LimitedTaskQueue (unsigned int iLimit)

template<typename T >
void	push (T &&iAction)
	asynchronously pushes functor iAction into queue More...

template<typename T >
void	pushAndPause (T &&iAction)
	asynchronously pushes functor iAction into queue then pause the queue and run iAction More...

template<typename T >
void	pushAndWait (T &&iAction)
	synchronously pushes functor iAction into queue More...

Private Member Functions
	LimitedTaskQueue (const LimitedTaskQueue &)=delete

const LimitedTaskQueue &	operator= (const LimitedTaskQueue &)=delete

Private Attributes
std::vector< SerialTaskQueue >	m_queues

Detailed Description

Definition at line 39 of file LimitedTaskQueue.h.

Constructor & Destructor Documentation

◆ LimitedTaskQueue() [1/2]

edm::LimitedTaskQueue::LimitedTaskQueue ( unsigned int iLimit )

inline

Definition at line 41 of file LimitedTaskQueue.h.

41 : m_queues{iLimit} {}

◆ LimitedTaskQueue() [2/2]

edm::LimitedTaskQueue::LimitedTaskQueue ( const LimitedTaskQueue & )

privatedelete

Member Function Documentation

◆ concurrencyLimit()

unsigned int edm::LimitedTaskQueue::concurrencyLimit ( ) const

inline

Definition at line 116 of file LimitedTaskQueue.h.

116 { return m_queues.size(); }

References m_queues.

Referenced by edm::limited::EDFilterBase::concurrencyLimit(), edm::limited::EDAnalyzerBase::concurrencyLimit(), edm::limited::EDProducerBase::concurrencyLimit(), and edm::limited::OutputModuleBase::concurrencyLimit().

◆ operator=()

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

privatedelete

◆ push()

template<typename T >

void LimitedTaskQueue::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 127 of file LimitedTaskQueue.h.

                                          {
     auto set_to_run = std::make_shared<std::atomic<bool>>(false);
     for (auto& q : m_queues) {
       q.push([set_to_run, iAction]() mutable {
         bool expected = false;
         if (set_to_run->compare_exchange_strong(expected, true)) {
           iAction();
         }
       });
     }
   }

References m_queues, and data-class-funcs::q.

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

◆ pushAndPause()

template<typename T >

void LimitedTaskQueue::pushAndPause ( T && iAction )

asynchronously pushes functor iAction into queue then pause the queue and run iAction

iAction must take as argument a copy of a LimitedTaskQueue::Resumer. To resume the queue let the last copy of the Resumer go out of scope, or call Resumer::resume(). Using this function will decrease the allowed concurrency limit by 1.

Definition at line 160 of file LimitedTaskQueue.h.

                                                  {
     auto set_to_run = std::make_shared<std::atomic<bool>>(false);
     for (auto& q : m_queues) {
       q.push([&q, set_to_run, iAction]() mutable {
         bool expected = false;
         if (set_to_run->compare_exchange_strong(expected, true)) {
           q.pause();
           iAction(Resumer(&q));
         }
       });
     }
   }

References m_queues, and data-class-funcs::q.

Referenced by edm::eventsetup::EventSetupRecordIOVQueue::startNewIOVAsync().

◆ pushAndWait()

template<typename T >

void LimitedTaskQueue::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 140 of file LimitedTaskQueue.h.

                                                 {
     tbb::empty_task* waitTask = new (tbb::task::allocate_root()) tbb::empty_task;
     waitTask->set_ref_count(2);
     auto set_to_run = std::make_shared<std::atomic<bool>>(false);
     for (auto& q : m_queues) {
       q.push([set_to_run, waitTask, iAction]() mutable {
         bool expected = false;
         if (set_to_run->compare_exchange_strong(expected, true)) {
           // Exception needs to be caught in order to decrease the waitTask reference count at the end. The user of SerialTaskQueue should handle exceptions within iAction.
           CMS_SA_ALLOW try { iAction(); } catch (...) {
           }
           waitTask->decrement_ref_count();
         }
       });
     }
     waitTask->wait_for_all();
     tbb::task::destroy(*waitTask);
   }