00001 // -*- C++ -*- 00002 // 00003 // Package: Concurrency 00004 // Class : SerialTaskQueue 00005 // 00006 // Implementation: 00007 // [Notes on implementation] 00008 // 00009 // Original Author: Chris Jones 00010 // Created: Thu Feb 21 11:31:52 CST 2013 00011 // $Id: SerialTaskQueue.cc,v 1.1 2013/02/21 22:14:11 chrjones Exp $ 00012 // 00013 00014 // system include files 00015 00016 // user include files 00017 #include "FWCore/Concurrency/interface/SerialTaskQueue.h" 00018 00019 #include "FWCore/Utilities/interface/Likely.h" 00020 00021 using namespace edm; 00022 00023 // 00024 // member functions 00025 // 00026 bool 00027 SerialTaskQueue::resume() { 00028 if(0==--m_pauseCount) { 00029 tbb::task* t = pickNextTask(); 00030 if(0 != t) { 00031 tbb::task::spawn(*t); 00032 } 00033 return true; 00034 } 00035 return false; 00036 } 00037 00038 void 00039 SerialTaskQueue::pushTask(TaskBase* iTask) { 00040 tbb::task* t = pushAndGetNextTask(iTask); 00041 if(0!=t) { 00042 tbb::task::spawn(*t); 00043 } 00044 } 00045 00046 tbb::task* 00047 SerialTaskQueue::pushAndGetNextTask(TaskBase* iTask) { 00048 tbb::task* returnValue{0}; 00049 if likely(0!=iTask) { 00050 m_tasks.push(iTask); 00051 returnValue = pickNextTask(); 00052 } 00053 return returnValue; 00054 } 00055 00056 00057 tbb::task* 00058 SerialTaskQueue::finishedTask() { 00059 m_taskChosen.clear(); 00060 return pickNextTask(); 00061 } 00062 00063 SerialTaskQueue::TaskBase* 00064 SerialTaskQueue::pickNextTask() { 00065 00066 if likely(0 == m_pauseCount and not m_taskChosen.test_and_set()) { 00067 TaskBase* t=0; 00068 if likely(m_tasks.try_pop(t)) { 00069 return t; 00070 } 00071 //no task was actually pulled 00072 m_taskChosen.clear(); 00073 00074 //was a new entry added after we called 'try_pop' but before we did the clear? 00075 if(not m_tasks.empty() and not m_taskChosen.test_and_set()) { 00076 TaskBase* t=0; 00077 if(m_tasks.try_pop(t)) { 00078 return t; 00079 } 00080 //no task was still pulled since a different thread beat us to it 00081 m_taskChosen.clear(); 00082 00083 } 00084 } 00085 return 0; 00086 } 00087 00088 void SerialTaskQueue::pushAndWait(tbb::empty_task* iWait, TaskBase* iTask) { 00089 auto nextTask = pushAndGetNextTask(iTask); 00090 if likely(nullptr != nextTask) { 00091 if likely(nextTask == iTask) { 00092 //spawn and wait for all requires the task to have its parent set 00093 iWait->spawn_and_wait_for_all(*nextTask); 00094 } else { 00095 tbb::task::spawn(*nextTask); 00096 iWait->wait_for_all(); 00097 } 00098 } else { 00099 //a task must already be running in this queue 00100 iWait->wait_for_all(); 00101 } 00102 tbb::task::destroy(*iWait); 00103 } 00104 00105 00106 // 00107 // const member functions 00108 // 00109 00110 // 00111 // static member functions 00112 //