LimitedTaskQueue.h

Go to the documentation of this file.

#ifndef FWCore_Concurrency_LimitedTaskQueue_h
#define FWCore_Concurrency_LimitedTaskQueue_h
// -*- C++ -*-
//
// Package:     Concurrency
// Class  :     LimitedTaskQueue
// 
//
// Original Author:  Chris Jones
//         Created:  Thu Feb 21 11:14:39 CST 2013
// $Id$
//

// system include files
#include <atomic>
#include <vector>
#include <memory>

#include "FWCore/Concurrency/interface/SerialTaskQueue.h"

// user include files

// forward declarations
namespace edm {
   class LimitedTaskQueue
   {
   public:
     LimitedTaskQueue(unsigned int iLimit):
     m_queues{iLimit}
      {  }
      
     
      // ---------- member functions ---------------------------
     

      template<typename T>
      void push(const T& iAction);
      

      template<typename T>
      void pushAndWait(const T& iAction);
      
   private:
      LimitedTaskQueue(const LimitedTaskQueue&) = delete;
      const LimitedTaskQueue& operator=(const LimitedTaskQueue&) = delete;
     
      // ---------- member data --------------------------------
     std::vector<SerialTaskQueue> m_queues;
   };
   
   template<typename T>
   void LimitedTaskQueue::push(const T& iAction) {
     auto set_to_run = std::make_shared<std::atomic<bool>>(false);
     for(auto& q: m_queues) {
       q.push([set_to_run,iAction]() {
         bool expected = false;
         if(set_to_run->compare_exchange_strong(expected,true)) {
           iAction();
         }
       });
     }
   }
   
   template<typename T>
   void LimitedTaskQueue::pushAndWait(const T& iAction) {
      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]() {
         bool expected = false;
         if(set_to_run->compare_exchange_strong(expected,true)) {
           try {
             iAction();
           }catch(...) {}
           waitTask->decrement_ref_count();
         }
       });
     }
     waitTask->wait_for_all();
     tbb::task::destroy(*waitTask);
   }
   
}

#endif