Concurrency/src/WaitingTaskWithArenaHolder.cc

0001 // -*- C++ -*-
0002 //
0003 // Package:     Concurrency
0004 // Class  :     WaitingTaskWithArenaHolder
0005 //
0006 // Original Author:  W. David Dagenhart
0007 //         Created:  6 December 2017
0008
0009 #include "FWCore/Concurrency/interface/WaitingTaskWithArenaHolder.h"
0010 #include "FWCore/Concurrency/interface/WaitingTask.h"
0011 #include "FWCore/Concurrency/interface/WaitingTaskHolder.h"
0012 #include "FWCore/Utilities/interface/Likely.h"
0013
0014 namespace edm {
0015
0016   WaitingTaskWithArenaHolder::WaitingTaskWithArenaHolder() : m_task(nullptr) {}
0017
0018   // Note that the arena will be the one containing the thread
0019   // that runs this constructor. This is the arena where you
0020   // eventually intend for the task to be spawned.
0021   WaitingTaskWithArenaHolder::WaitingTaskWithArenaHolder(oneapi::tbb::task_group& iGroup, WaitingTask* iTask)
0022       : m_task(iTask),
0023         m_group(&iGroup),
0024         m_arena(std::make_shared<oneapi::tbb::task_arena>(oneapi::tbb::task_arena::attach())) {
0025     m_task->increment_ref_count();
0026   }
0027
0028   WaitingTaskWithArenaHolder::WaitingTaskWithArenaHolder(WaitingTaskHolder&& iTask)
0029       : m_task(iTask.release_no_decrement()),
0030         m_group(iTask.group()),
0031         m_arena(std::make_shared<oneapi::tbb::task_arena>(oneapi::tbb::task_arena::attach())) {}
0032
0033   WaitingTaskWithArenaHolder::~WaitingTaskWithArenaHolder() {
0034     if (m_task) {
0035       doneWaiting(std::exception_ptr{});
0036     }
0037   }
0038
0039   WaitingTaskWithArenaHolder::WaitingTaskWithArenaHolder(WaitingTaskWithArenaHolder const& iHolder)
0040       : m_task(iHolder.m_task), m_group(iHolder.m_group), m_arena(iHolder.m_arena) {
0041     if (LIKELY(m_task != nullptr)) {
0042       m_task->increment_ref_count();
0043     }
0044   }
0045
0046   WaitingTaskWithArenaHolder::WaitingTaskWithArenaHolder(WaitingTaskWithArenaHolder&& iOther)
0047       : m_task(iOther.m_task), m_group(iOther.m_group), m_arena(std::move(iOther.m_arena)) {
0048     iOther.m_task = nullptr;
0049   }
0050
0051   WaitingTaskWithArenaHolder& WaitingTaskWithArenaHolder::operator=(const WaitingTaskWithArenaHolder& iRHS) {
0052     WaitingTaskWithArenaHolder tmp(iRHS);
0053     std::swap(m_task, tmp.m_task);
0054     std::swap(m_group, tmp.m_group);
0055     std::swap(m_arena, tmp.m_arena);
0056     return *this;
0057   }
0058
0059   WaitingTaskWithArenaHolder& WaitingTaskWithArenaHolder::operator=(WaitingTaskWithArenaHolder&& iRHS) {
0060     WaitingTaskWithArenaHolder tmp(std::move(iRHS));
0061     std::swap(m_task, tmp.m_task);
0062     std::swap(m_group, tmp.m_group);
0063     std::swap(m_arena, tmp.m_arena);
0064     return *this;
0065   }
0066
0067   // This spawns the task. The arena is needed to get the task spawned
0068   // into the correct arena of threads. Use of the arena allows doneWaiting
0069   // to be called from a thread outside the arena of threads that will manage
0070   // the task. doneWaiting can be called from a non-TBB thread.
0071   void WaitingTaskWithArenaHolder::doneWaiting(std::exception_ptr iExcept) {
0072     if (iExcept) {
0073       m_task->dependentTaskFailed(iExcept);
0074     }
0075     //enqueue can run the task before we finish
0076     // doneWaiting and some other thread might
0077     // try to reuse this object. Resetting
0078     // before enqueue avoids problems
0079     auto task = m_task;
0080     m_task = nullptr;
0081     if (0 == task->decrement_ref_count()) {
0082       // The enqueue call will cause a worker thread to be created in
0083       // the arena if there is not one already.
0084       m_arena->enqueue([task = task, group = m_group]() {
0085         group->run([task]() {
0086           TaskSentry s(task);
0087           task->execute();
0088         });
0089       });
0090     }
0091   }
0092
0093   void WaitingTaskWithArenaHolder::presetTaskAsFailed(std::exception_ptr iExcept) noexcept {
0094     if (iExcept) {
0095       m_task->dependentTaskFailed(iExcept);
0096     }
0097   }
0098
0099   // This next function is useful if you know from the context that
0100   // m_arena (which is set when the  constructor was executes) is the
0101   // same arena in which you want to execute the doneWaiting function.
0102   // It allows an optimization which avoids the enqueue step in the
0103   // doneWaiting function.
0104   //
0105   // Be warned though that in general this function cannot be used.
0106   // Spawning a task outside the correct arena could create a new separate
0107   // arena with its own extra TBB worker threads if this function is used
0108   // in an inappropriate context (and silently such that you might not notice
0109   // the problem quickly).
0110
0111   WaitingTaskHolder WaitingTaskWithArenaHolder::makeWaitingTaskHolderAndRelease() {
0112     WaitingTaskHolder holder(*m_group, m_task);
0113     m_task->decrement_ref_count();
0114     m_task = nullptr;
0115     return holder;
0116   }
0117
0118   bool WaitingTaskWithArenaHolder::taskHasFailed() const noexcept { return static_cast<bool>(m_task->exceptionPtr()); }
0119
0120   bool WaitingTaskWithArenaHolder::hasTask() const noexcept { return m_task != nullptr; }
0121
0122 }  // namespace edm