Loading...
Searching...
No Matches
tasklets.hpp
Go to the documentation of this file.
// -*- mode: C++; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*-
// vi: set et ts=4 sw=4 sts=4:
/*
This file is part of the Open Porous Media project (OPM).
OPM is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 2 of the License, or
(at your option) any later version.
OPM is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with OPM. If not, see <http://www.gnu.org/licenses/>.
Consult the COPYING file in the top-level source directory of this
module for the precise wording of the license and the list of
copyright holders.
*/
#ifndef OPM_TASKLETS_HPP
#define OPM_TASKLETS_HPP
#include <atomic>
#include <condition_variable>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
namespace Opm {
class TaskletInterface
{
public:
explicit TaskletInterface(int refCount = 1)
: referenceCount_(refCount)
{}
virtual ~TaskletInterface() {}
virtual void run() = 0;
virtual bool isEndMarker () const { return false; }
void dereference()
{ -- referenceCount_; }
int referenceCount() const
{ return referenceCount_; }
private:
int referenceCount_;
};
template <class Fn>
class FunctionRunnerTasklet : public TaskletInterface
{
public:
FunctionRunnerTasklet(const FunctionRunnerTasklet&) = default;
FunctionRunnerTasklet(int numInvocations, const Fn& fn)
: TaskletInterface(numInvocations)
, fn_(fn)
{}
void run() override
{ fn_(); }
private:
const Fn& fn_;
};
class TaskletRunner
{
class BarrierTasklet : public TaskletInterface
{
public:
explicit BarrierTasklet(unsigned numWorkers);
void run() override;
void wait();
private:
unsigned numWorkers_;
unsigned numWaiting_;
std::condition_variable barrierCondition_;
std::mutex barrierMutex_;
};
class TerminateThreadTasklet : public TaskletInterface
{
public:
void run() override
{}
bool isEndMarker() const override
{ return true; }
};
public:
// prohibit copying of tasklet runners
TaskletRunner(const TaskletRunner&) = delete;
explicit TaskletRunner(unsigned numWorkers);
~TaskletRunner();
bool failure() const;
int workerThreadIndex() const;
int numWorkerThreads() const
{ return threads_.size(); }
void dispatch(std::shared_ptr<TaskletInterface> tasklet);
template <class Fn>
std::shared_ptr<FunctionRunnerTasklet<Fn> > dispatchFunction(Fn &fn, int numInvocations = 1)
{
using Tasklet = FunctionRunnerTasklet<Fn>;
auto tasklet = std::make_shared<Tasklet>(numInvocations, fn);
this->dispatch(tasklet);
return tasklet;
}
void barrier();
private:
// Atomic flag that is set to failure if any of the tasklets run by the TaskletRunner fails.
// This flag is checked before new tasklets run or get dispatched and in case it is true, the
// thread execution will be stopped / no new tasklets will be started and the program will abort.
// To set the flag and load the flag, we use std::memory_order_relaxed.
// Atomic operations tagged memory_order_relaxed are not synchronization operations; they do not
// impose an order among concurrent memory accesses. They guarantee atomicity and modification order
// consistency. This is the right choice for the setting here, since it is enough to broadcast failure
// before new run or get dispatched.
std::atomic<bool> failureFlag_ = false;
protected:
// main function of the worker thread
static void startWorkerThread_(TaskletRunner* taskletRunner, int workerThreadIndex);
void run_();
std::vector<std::unique_ptr<std::thread> > threads_;
std::queue<std::shared_ptr<TaskletInterface> > taskletQueue_;
std::mutex taskletQueueMutex_;
std::condition_variable workAvailableCondition_;
static thread_local TaskletRunner* taskletRunner_;
static thread_local int workerThreadIndex_;
};
} // end namespace Opm
#endif // OPM_TASKLETS_HPP
Opm::FunctionRunnerTasklet
A simple tasklet that runs a function that returns void and does not take any arguments a given numbe...
Definition tasklets.hpp:71
Opm::TaskletInterface
The base class for tasklets.
Definition tasklets.hpp:45
Opm::TaskletRunner
Handles where a given tasklet is run.
Definition tasklets.hpp:93
Opm::TaskletRunner::workerThreadIndex
int workerThreadIndex() const
Returns the index of the current worker thread.
Definition tasklets.cpp:94
Opm::TaskletRunner::numWorkerThreads
int numWorkerThreads() const
Returns the number of worker threads for the tasklet runner.
Definition tasklets.hpp:157
Opm::TaskletRunner::run_
void run_()
do the work until the queue received an end tasklet
Definition tasklets.cpp:157
Opm::TaskletRunner::~TaskletRunner
~TaskletRunner()
Destructor.
Definition tasklets.cpp:77
Opm::TaskletRunner::barrier
void barrier()
Make sure that all tasklets have been completed after this method has been called.
Definition tasklets.cpp:135
Opm::TaskletRunner::dispatchFunction
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
Opm::TaskletRunner::dispatch
void dispatch(std::shared_ptr< TaskletInterface > tasklet)
Add a new tasklet.
Definition tasklets.cpp:101
Opm
This file contains a set of helper functions used by VFPProd / VFPInj.
Definition blackoilbioeffectsmodules.hh:43