tasklets.hpp
Go to the documentation of this file.
1// -*- mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
2// vi: set et ts=4 sw=4 sts=4:
3/*
4 This file is part of the Open Porous Media project (OPM).
5
6 OPM is free software: you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation, either version 2 of the License, or
9 (at your option) any later version.
10
11 OPM is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
15
16 You should have received a copy of the GNU General Public License
17 along with OPM. If not, see <http://www.gnu.org/licenses/>.
18
19 Consult the COPYING file in the top-level source directory of this
20 module for the precise wording of the license and the list of
21 copyright holders.
22*/
27#ifndef OPM_TASKLETS_HPP
28#define OPM_TASKLETS_HPP
29
30#include <atomic>
31#include <condition_variable>
32#include <memory>
33#include <mutex>
34#include <queue>
35#include <thread>
36
37namespace Opm {
38
45{
46public:
47 explicit TaskletInterface(int refCount = 1)
48 : referenceCount_(refCount)
49 {}
50
51 virtual ~TaskletInterface() {}
52 virtual void run() = 0;
53 virtual bool isEndMarker () const { return false; }
54
56 { -- referenceCount_; }
57
58 int referenceCount() const
59 { return referenceCount_; }
60
61private:
62 int referenceCount_;
63};
64
69template <class Fn>
71{
72public:
74 FunctionRunnerTasklet(int numInvocations, const Fn& fn)
75 : TaskletInterface(numInvocations)
76 , fn_(fn)
77 {}
78
79 void run() override
80 { fn_(); }
81
82private:
83 const Fn& fn_;
84};
85
93{
95 class BarrierTasklet : public TaskletInterface
96 {
97 public:
98 explicit BarrierTasklet(unsigned numWorkers);
99
100 void run() override;
101
102 void wait();
103
104 private:
105 unsigned numWorkers_;
106 unsigned numWaiting_;
107
108 std::condition_variable barrierCondition_;
109 std::mutex barrierMutex_;
110 };
111
114 class TerminateThreadTasklet : public TaskletInterface
115 {
116 public:
117 void run() override
118 {}
119
120 bool isEndMarker() const override
121 { return true; }
122 };
123
124public:
125 // prohibit copying of tasklet runners
126 TaskletRunner(const TaskletRunner&) = delete;
127
134 explicit TaskletRunner(unsigned numWorkers);
135
144
145 bool failure() const;
146
152 int workerThreadIndex() const;
153
158 { return threads_.size(); }
159
165 void dispatch(std::shared_ptr<TaskletInterface> tasklet);
166
170 template <class Fn>
171 std::shared_ptr<FunctionRunnerTasklet<Fn> > dispatchFunction(Fn &fn, int numInvocations = 1)
172 {
173 using Tasklet = FunctionRunnerTasklet<Fn>;
174 auto tasklet = std::make_shared<Tasklet>(numInvocations, fn);
175 this->dispatch(tasklet);
176 return tasklet;
177 }
178
182 void barrier();
183
184private:
185 // Atomic flag that is set to failure if any of the tasklets run by the TaskletRunner fails.
186 // This flag is checked before new tasklets run or get dispatched and in case it is true, the
187 // thread execution will be stopped / no new tasklets will be started and the program will abort.
188 // To set the flag and load the flag, we use std::memory_order_relaxed.
189 // Atomic operations tagged memory_order_relaxed are not synchronization operations; they do not
190 // impose an order among concurrent memory accesses. They guarantee atomicity and modification order
191 // consistency. This is the right choice for the setting here, since it is enough to broadcast failure
192 // before new run or get dispatched.
193 std::atomic<bool> failureFlag_ = false;
194
195protected:
196 // main function of the worker thread
197 static void startWorkerThread_(TaskletRunner* taskletRunner, int workerThreadIndex);
198
200 void run_();
201
202 std::vector<std::unique_ptr<std::thread> > threads_;
203 std::queue<std::shared_ptr<TaskletInterface> > taskletQueue_;
205 std::condition_variable workAvailableCondition_;
206
207 static thread_local TaskletRunner* taskletRunner_;
208 static thread_local int workerThreadIndex_;
209};
210
211} // end namespace Opm
212
213#endif // OPM_TASKLETS_HPP
A simple tasklet that runs a function that returns void and does not take any arguments a given numbe...
Definition: tasklets.hpp:71
FunctionRunnerTasklet(int numInvocations, const Fn &fn)
Definition: tasklets.hpp:74
FunctionRunnerTasklet(const FunctionRunnerTasklet &)=default
void run() override
Definition: tasklets.hpp:79
The base class for tasklets.
Definition: tasklets.hpp:45
virtual bool isEndMarker() const
Definition: tasklets.hpp:53
TaskletInterface(int refCount=1)
Definition: tasklets.hpp:47
int referenceCount() const
Definition: tasklets.hpp:58
void dereference()
Definition: tasklets.hpp:55
virtual ~TaskletInterface()
Definition: tasklets.hpp:51
virtual void run()=0
Handles where a given tasklet is run.
Definition: tasklets.hpp:93
int workerThreadIndex() const
Returns the index of the current worker thread.
bool failure() const
static thread_local int workerThreadIndex_
Definition: tasklets.hpp:208
int numWorkerThreads() const
Returns the number of worker threads for the tasklet runner.
Definition: tasklets.hpp:157
void run_()
do the work until the queue received an end tasklet
std::mutex taskletQueueMutex_
Definition: tasklets.hpp:204
~TaskletRunner()
Destructor.
static thread_local TaskletRunner * taskletRunner_
Definition: tasklets.hpp:207
std::condition_variable workAvailableCondition_
Definition: tasklets.hpp:205
void barrier()
Make sure that all tasklets have been completed after this method has been called.
std::shared_ptr< FunctionRunnerTasklet< Fn > > dispatchFunction(Fn &fn, int numInvocations=1)
Convenience method to construct a new function runner tasklet and dispatch it immediately.
Definition: tasklets.hpp:171
void dispatch(std::shared_ptr< TaskletInterface > tasklet)
Add a new tasklet.
TaskletRunner(unsigned numWorkers)
Creates a tasklet runner with numWorkers underling threads for doing work.
static void startWorkerThread_(TaskletRunner *taskletRunner, int workerThreadIndex)
std::vector< std::unique_ptr< std::thread > > threads_
Definition: tasklets.hpp:202
TaskletRunner(const TaskletRunner &)=delete
std::queue< std::shared_ptr< TaskletInterface > > taskletQueue_
Definition: tasklets.hpp:203
Definition: blackoilboundaryratevector.hh:39