parallel.h
Go to the documentation of this file.
/* Copyright (C) 2013 Chris Vine
The library comprised in this file or of which this file is part is
distributed by Chris Vine under the GNU Lesser General Public
License as follows:
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public License
as published by the Free Software Foundation; either version 2.1 of
the License, or (at your option) any later version.
This library 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
Lesser General Public License, version 2.1, for more details.
You should have received a copy of the GNU Lesser General Public
License, version 2.1, along with this library (see the file LGPL.TXT
which came with this source code package in the c++-gtk-utils
sub-directory); if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
However, it is not intended that the object code of a program whose
source code instantiates a template from this file or uses macros or
inline functions (of any length) should by reason only of that
instantiation or use be subject to the restrictions of use in the GNU
Lesser General Public License. With that in mind, the words "and
macros, inline functions and instantiations of templates (of any
length)" shall be treated as substituted for the words "and small
macros and small inline functions (ten lines or less in length)" in
the fourth paragraph of section 5 of that licence. This does not
affect any other reason why object code may be subject to the
restrictions in that licence (nor for the avoidance of doubt does it
affect the application of section 2 of that licence to modifications
of the source code in this file).
*/
#ifndef CGU_PARALLEL_H
#define CGU_PARALLEL_H
#include <utility> // for std::move and std::forward
#include <memory> // for std::unique_ptr
#include <iterator> // for std::iterator_traits and std::distance
#include <exception> // for std::exception
#include <functional> // for std::bind
#include <type_traits> // for std::remove_reference and std::remove_const
#include <c++-gtk-utils/callback.h>
#include <c++-gtk-utils/task_manager.h>
#include <c++-gtk-utils/mutex.h>
#include <c++-gtk-utils/shared_handle.h>
namespace Cgu {
namespace Thread {
struct ParallelError: public std::exception {
virtual const char* what() const throw() {return "ParallelError\n";}
};
#ifndef DOXYGEN_PARSING
namespace ParallelHelper {
template <class ArgType, class DiffType, class Iterator>
void for_each_cb_func(const Cgu::Callback::SafeFunctorArg<ArgType&>& s_task,
Iterator iter,
Mutex* m, Cond* cond,
DiffType* remaining) {
s_task(*iter);
Mutex::Lock l{*m};
--*remaining;
cond->signal();
}
template <class FType, class ArgType, class DestType>
void transform1_func(const FType& func,
ArgType& arg,
DestType& res) {
res = func(arg);
}
template <class ArgType, class DestType, class DiffType, class SourceIterator>
void transform1_cb_func(const Cgu::Callback::SafeFunctorArg<ArgType&, DestType&>& s_task,
SourceIterator iter,
Mutex* m, Cond* cond,
DestType* results,
DiffType index,
DiffType* remaining) {
DestType res;
s_task(*iter, res);
Mutex::Lock l{*m};
// move to 'results' within the mutex because g++ <= 4.7 does not
// correctly implement the C++11 memory model on some 64 bit
// platforms (this is a slight pessimization for gcc >= 4.8)
results[index] = std::move(res);
--*remaining;
cond->signal();
}
template <class FType, class Arg1Type,
class Arg2Type, class DestType>
void transform2_func(const FType& func,
Arg1Type& arg1,
Arg2Type& arg2,
DestType& res) {
res = func(arg1, arg2);
}
template <class Arg1Type, class Arg2Type, class DestType,
class DiffType, class SourceIterator1, class SourceIterator2>
void transform2_cb_func(const Cgu::Callback::SafeFunctorArg<Arg1Type&, Arg2Type&, DestType&>& s_task,
SourceIterator1 iter1,
SourceIterator2 iter2,
Mutex* m, Cond* cond,
DestType* results,
DiffType index,
DiffType* remaining) {
DestType res;
s_task(*iter1, *iter2, res);
Mutex::Lock l{*m};
// move to 'results' within the mutex because g++ <= 4.7 does not
// correctly implement the C++11 memory model on some 64 bit
// platforms (this is a slight pessimization for gcc >= 4.8)
results[index] = std::move(res);
--*remaining;
cond->signal();
}
template <class DiffType>
void fail_func(Mutex* m, Cond* cond,
bool* error, DiffType* remaining) {
Mutex::Lock l{*m};
--*remaining;
*error = true;
cond->signal();
}
} // namespace ParallelHelper
#endif // DOXYGEN_PARSING
/**
* \#include <c++-gtk-utils/parallel.h>
*
* This function applies a callable object to each element of a
* container in the range ['first', 'last'), by executing each such
* application as a task of a Thread::TaskManager object. Tasks are
* added to the Thread::TaskManager object in the order in which the
* respective elements appear in the container, but no other ordering
* arises, and they will execute in parallel to the extent that the
* Thread::TaskManager object has sufficient threads available to do
* so.
*
* Apart from that, and that this function returns void, it does the
* same as std::for_each(). It can mutate container elements if the
* callable object takes its argument by non-const reference. It will
* not return until the callable object has been applied to all of the
* elements in the range ['first', 'last').
*
* This function can be called by a task running on the same
* TaskManager object.
*
* @param tm The Thread::TaskManager object on which the tasks will
* run.
* @param first The beginning of the range to which 'func' is to be
* applied.
* @param last One past the last element to which 'func' is to be
* applied.
* @param func A callable object to be applied to each element in the
* range ['first', 'last'), such as formed by a lambda expression or
* the result of std::bind. It should take a single unbound argument
* of the value type of the container to which 'first' and 'last'
* relate or a const or non-const reference to that type. Any return
* value is discarded. If an exception propagates from 'func', the
* exception will be consumed while the for each loop is running, and
* an attempt will still be made to apply 'func' to all remaining
* elements in the range ['first', 'last'), and only after that
* attempt has completed will the exception Cgu::Thread::ParallelError
* be thrown.
* @exception std::bad_alloc This exception will be thrown if memory
* is exhausted and the system throws in that case. (On systems with
* over-commit/lazy-commit combined with virtual memory (swap), it is
* rarely useful to check for memory exhaustion). If this exception
* is thrown, some tasks may nonetheless have already started by
* virtue of the call to this function, but subsequent ones will not.
* @exception Cgu::Thread::TaskError This exception will be thrown if
* stop_all() has previously been called on the Thread::TaskManager
* object, or if another thread calls stop_all() after this method is
* called but before it has returned. It will also be thrown if the
* Thread::TaskManager object's is_error() method would return true
* because its internal thread pool loop implementation has thrown
* std::bad_alloc, or a thread has failed to start correctly. (On
* systems with over-commit/lazy-commit combined with virtual memory
* (swap), it is rarely useful to check for memory exhaustion, but
* there may be some specialized cases where the return value of
* is_error() is useful.) If this exception is thrown, some tasks may
* nonetheless have already started by virtue of the call to this
* function.
* @exception Cgu::Thread::ParallelError This exception will be thrown
* if an exception propagates from the 'func' callable object when it
* executes on being applied to one or more elements of the container.
* Such an exception will not stop an attempt being made to apply
* 'func' (successfully or unsuccessfully) to all elements in the
* range ['first', 'last'). Cgu::Thread::ParallelError will be thrown
* after such attempted application has finished.
* @exception Cgu::Thread::MutexError This exception will be thrown if
* initialization of a mutex used by this function fails. (It is
* often not worth checking for this, as it means either memory is
* exhausted or pthread has run out of other resources to create new
* mutexes.) If this exception is thrown, no tasks will start.
* @exception Cgu::Thread::CondError This exception will be thrown if
* initialization of a condition variable used by this function fails.
* (It is often not worth checking for this, as it means either memory
* is exhausted or pthread has run out of other resources to create
* new condition variables.) If this exception is thrown, no tasks
* will start.
* @note An exception might also be thrown if the copy or move
* constructor of the 'func' callable objects throws. If such an
* exception is thrown, no tasks will start.
*
* Since 2.0.19/2.2.2
*/
template <class Iterator, class Func>
void parallel_for_each(TaskManager& tm,
Iterator first,
Iterator last,
Func&& func) {
typedef typename std::iterator_traits<Iterator>::value_type ArgType;
typedef typename std::iterator_traits<Iterator>::difference_type DiffType;
DiffType remaining = std::distance(first, last);
Mutex mutex;
Cond cond;
bool error = false;
// construct SafeFunctorArg objects so that they can be shared
// between different tasks
Cgu::Callback::SafeFunctorArg<ArgType&> s_task{
Cgu::Callback::lambda<ArgType&>(std::forward<Func>(func))
};
Cgu::Callback::SafeFunctor s_fail{
Cgu::Callback::make(&ParallelHelper::fail_func<DiffType>,
&mutex,
&cond,
&error,
&remaining)
};
for (; first != last; ++first) {
std::unique_ptr<const Cgu::Callback::Callback> task_cb(
Cgu::Callback::make_ref(&ParallelHelper::for_each_cb_func<ArgType, DiffType, Iterator>,
s_task,
first,
&mutex,
&cond,
&remaining)
);
std::unique_ptr<const Cgu::Callback::Callback> fail_cb(
Cgu::Callback::lambda<>([s_fail] () {s_fail();})
);
tm.add_task(std::move(task_cb), std::move(fail_cb));
}
Mutex::Lock l{mutex};
while (remaining > 0) cond.wait(mutex);
if (error) throw ParallelError();
}
/**
* \#include <c++-gtk-utils/parallel.h>
*
* This function maps over a container in the range ['first', 'last'),
* applying a unary callable object to each element of the container
* in that range and storing the result in the destination range, by
* executing each such application as a task of a Thread::TaskManager
* object. Tasks are added to the Thread::TaskManager object in the
* order in which the respective elements appear in the source
* container, but no other ordering arises, and they will execute in
* parallel to the extent that the Thread::TaskManager object has
* sufficient threads available to do so.
*
* Apart from that, this function does the same as the version of
* std::transform() taking a unary function. It will not return until
* the callable object has been applied to all of the elements in the
* range ['first', 'last').
*
* This function can be called by a task running on the same
* TaskManager object (a result can therefore be delivered
* asynchronously by calling this function in a task created by the
* make_task_when(), make_task_compose() or make_task_when_full()
* Thread::TaskManager methods). A task can carry out a map-reduce
* operation by passing the result of calling this function to
* std::accumulate() to perform a fold-left or fold-right on that
* result.
*
* A separate overload of this function takes a binary callable
* object.
*
* Here is a trivial example of a map-reduce operation which maps over
* a vector by multiplying each element by 2 in separate tasks, and
* then folds-left using std::accumulate() (std::accumulate() can fold
* using any callable object, but in this example the default of
* addition is used):
*
* @code
* using namespace Cgu;
* std::vector<int> v{1, 2, 3, 4, 5};
* Thread::TaskManager tm;
*
* Thread::parallel_transform(tm,
* v.begin(),
* v.end(),
* v.begin(),
* [] (int elt) {return elt * 2;});
* // res will be equal to 30
* int res = std::accumulate(v.begin(), v.end(), 0);
* @endcode
*
* @param tm The Thread::TaskManager object on which the tasks will
* run.
* @param first The beginning of the range to which 'func' is to be
* applied.
* @param last One past the last element to which 'func' is to be
* applied.
* @param dest The beginning of the range to which the result of
* applying 'func' to the elements in the range ['first', 'last') is
* to be stored. As in the case of std::transform, this can overlap
* with or be the same as the source range. It may also be an
* inserter iterator.
* @param func A callable object to be applied to each element in the
* range ['first', 'last'), such as formed by a lambda expression or
* the result of std::bind. It should take a single unbound argument
* of the value type of the container to which 'first' and 'last'
* relate or a const or non-const reference to that type. If an
* exception propagates from 'func', the exception will be consumed
* while the transform loop is running, and an attempt will still be
* made to apply 'func' to all remaining elements in the range
* ['first', 'last'), and only after that attempt has completed will
* the exception Cgu::Thread::ParallelError be thrown.
* @exception std::bad_alloc This exception will be thrown if memory
* is exhausted and the system throws in that case. (On systems with
* over-commit/lazy-commit combined with virtual memory (swap), it is
* rarely useful to check for memory exhaustion). If this exception
* is thrown, some tasks may nonetheless have already started by
* virtue of the call to this function, but subsequent ones will not.
* @exception Cgu::Thread::TaskError This exception will be thrown if
* stop_all() has previously been called on the Thread::TaskManager
* object, or if another thread calls stop_all() after this method is
* called but before it has returned. It will also be thrown if the
* Thread::TaskManager object's is_error() method would return true
* because its internal thread pool loop implementation has thrown
* std::bad_alloc, or a thread has failed to start correctly. (On
* systems with over-commit/lazy-commit combined with virtual memory
* (swap), it is rarely useful to check for memory exhaustion, but
* there may be some specialized cases where the return value of
* is_error() is useful.) If this exception is thrown, some tasks may
* nonetheless have already started by virtue of the call to this
* function.
* @exception Cgu::Thread::ParallelError This exception will be thrown
* if an exception propagates from the 'func' callable object when it
* executes on being applied to one or more elements of the source
* container. Such an exception will not stop an attempt being made
* to apply 'func' (successfully or unsuccessfully) to all elements in
* the range ['first', 'last'). Cgu::Thread::ParallelError will be
* thrown after such attempted application has finished.
* @exception Cgu::Thread::MutexError This exception will be thrown if
* initialization of a mutex used by this function fails. (It is
* often not worth checking for this, as it means either memory is
* exhausted or pthread has run out of other resources to create new
* mutexes.) If this exception is thrown, no tasks will start.
* @exception Cgu::Thread::CondError This exception will be thrown if
* initialization of a condition variable used by this function fails.
* (It is often not worth checking for this, as it means either memory
* is exhausted or pthread has run out of other resources to create
* new condition variables.) If this exception is thrown, no tasks
* will start.
* @note An exception might also be thrown if the copy or move
* constructor of the 'func' callable objects throws. If such an
* exception is thrown, no tasks will start.
*
* Since 2.0.19/2.2.2
*/
template <class SourceIterator, class DestIterator, class Func>
void parallel_transform(TaskManager& tm,
SourceIterator first,
SourceIterator last,
DestIterator dest,
Func&& func) {
if (first == last) return;
typedef typename std::iterator_traits<SourceIterator>::value_type ArgType;
typedef typename std::iterator_traits<SourceIterator>::difference_type DiffType;
typedef typename std::remove_const<typename std::remove_reference<Func>::type>::type FType;
typedef decltype(func(*first)) DestType;
DiffType elts = std::distance(first, last);
DiffType remaining = elts;
ScopedHandle<DestType*> results{new DestType[elts]};
Mutex mutex;
Cond cond;
bool error = false;
// construct SafeFunctorArg objects so that they can be shared
// between different tasks
Cgu::Callback::SafeFunctorArg<ArgType&, DestType&> s_task{
Cgu::Callback::make_ref(&ParallelHelper::transform1_func<FType, ArgType, DestType>,
std::forward<Func>(func))
};
Cgu::Callback::SafeFunctor s_fail{
Cgu::Callback::make(&ParallelHelper::fail_func<DiffType>,
&mutex,
&cond,
&error,
&remaining)
};
for (DiffType index = 0; first != last; ++first, ++index) {
std::unique_ptr<const Cgu::Callback::Callback> task_cb(
Cgu::Callback::lambda<>(std::bind(&ParallelHelper::transform1_cb_func<ArgType, DestType, DiffType, SourceIterator>,
s_task,
first,
&mutex,
&cond,
results.get(),
index,
&remaining))
);
std::unique_ptr<const Cgu::Callback::Callback> fail_cb(
Cgu::Callback::lambda<>([s_fail] () {s_fail();})
);
tm.add_task(std::move(task_cb), std::move(fail_cb));
}
Mutex::Lock l{mutex};
while (remaining > 0) cond.wait(mutex);
if (error) throw ParallelError();
for (DiffType index = 0; index < elts; ++dest, ++index) {
*dest = std::move(results[index]);
}
}
/**
* \#include <c++-gtk-utils/parallel.h>
*
* This function maps over two containers, one in the range ['first1',
* 'last1') and the other beginning at 'first2', applying a binary
* callable object to each element of the containers in those ranges
* and storing the result in the destination range, by executing each
* such application as a task of a Thread::TaskManager object. Tasks
* are added to the Thread::TaskManager object in the order in which
* the respective elements appear in the source containers, but no
* other ordering arises, and they will execute in parallel to the
* extent that the Thread::TaskManager object has sufficient threads
* available to do so.
*
* Apart from that, this function does the same as the version of
* std::transform() taking a binary function. It will not return
* until the callable object has been applied to all of the elements
* in the source ranges.
*
* This function can be called by a task running on the same
* TaskManager object (a result can therefore be delivered
* asynchronously by calling this function in a task created by the
* make_task_when(), make_task_compose() or make_task_when_full()
* Thread::TaskManager methods). A task can carry out a map-reduce
* operation by passing the result of calling this function to
* std::accumulate() to perform a fold-left or fold-right on that
* result.
*
* A separate overload of this function takes a unary callable object.
*
* Here is a trivial example of a map-reduce operation which maps over
* two vectors by adding respective elements of the vectors in
* separate tasks, and then folds-left using std::accumulate()
* (std::accumulate() can fold using any callable object, but in this
* example the default of addition is used):
*
* @code
* using namespace Cgu;
* std::vector<int> v1{2, 4, 6, 8, 10};
* std::vector<int> v2{10, 20, 30, 40, 50};
* std::vector<int> v3;
* Thread::TaskManager tm;
*
* Thread::parallel_transform(tm,
* v1.begin(),
* v1.end(),
* v2.begin(),
* std::back_inserter(v3),
* std::plus<int>());
* // res will be equal to 180
* int res = std::accumulate(v3.begin(), v3.end(), 0);
* @endcode
*
* @param tm The Thread::TaskManager object on which the tasks will
* run.
* @param first1 The beginning of the range which is to be passed as
* the first argument of 'func'.
* @param last1 One past the last element of the range which is to be
* passed as the first argument of 'func'.
* @param first2 The beginning of the range which is to be passed as
* the second argument of 'func'.
* @param dest The beginning of the range to which the result of
* applying 'func' to the elements in the source ranges is to be
* stored. As in the case of std::transform, this can overlap with or
* be the same as one of the source ranges. It may also be an
* inserter iterator.
* @param func A binary callable object to be applied to each element
* in the source ranges, such as formed by a lambda expression or the
* result of std::bind. It should take two unbound arguments of the
* value types of the containers to which 'first1' and 'first2' relate
* or const or non-const references to those types. If an exception
* propagates from 'func', the exception will be consumed while the
* transform loop is running, and an attempt will still be made to
* apply 'func' to all remaining elements of the source ranges, and
* only after that attempt has completed will the exception
* Cgu::Thread::ParallelError be thrown.
* @exception std::bad_alloc This exception will be thrown if memory
* is exhausted and the system throws in that case. (On systems with
* over-commit/lazy-commit combined with virtual memory (swap), it is
* rarely useful to check for memory exhaustion). If this exception
* is thrown, some tasks may nonetheless have already started by
* virtue of the call to this function, but subsequent ones will not.
* @exception Cgu::Thread::TaskError This exception will be thrown if
* stop_all() has previously been called on the Thread::TaskManager
* object, or if another thread calls stop_all() after this method is
* called but before it has returned. It will also be thrown if the
* Thread::TaskManager object's is_error() method would return true
* because its internal thread pool loop implementation has thrown
* std::bad_alloc, or a thread has failed to start correctly. (On
* systems with over-commit/lazy-commit combined with virtual memory
* (swap), it is rarely useful to check for memory exhaustion, but
* there may be some specialized cases where the return value of
* is_error() is useful.) If this exception is thrown, some tasks may
* nonetheless have already started by virtue of the call to this
* function.
* @exception Cgu::Thread::ParallelError This exception will be thrown
* if an exception propagates from the 'func' callable object when it
* executes on being applied to one or more elements of the source
* ranges. Such an exception will not stop an attempt being made to
* apply 'func' (successfully or unsuccessfully) to all elements in
* the source ranges. Cgu::Thread::ParallelError will be thrown after
* such attempted application has finished.
* @exception Cgu::Thread::MutexError This exception will be thrown if
* initialization of a mutex used by this function fails. (It is
* often not worth checking for this, as it means either memory is
* exhausted or pthread has run out of other resources to create new
* mutexes.) If this exception is thrown, no tasks will start.
* @exception Cgu::Thread::CondError This exception will be thrown if
* initialization of a condition variable used by this function fails.
* (It is often not worth checking for this, as it means either memory
* is exhausted or pthread has run out of other resources to create
* new condition variables.) If this exception is thrown, no tasks
* will start.
* @note An exception might also be thrown if the copy or move
* constructor of the 'func' callable objects throws. If such an
* exception is thrown, no tasks will start.
*
* Since 2.0.19/2.2.2
*/
template <class SourceIterator1, class SourceIterator2, class DestIterator, class Func>
void parallel_transform(TaskManager& tm,
SourceIterator1 first1,
SourceIterator1 last1,
SourceIterator2 first2,
DestIterator dest,
Func&& func) {
if (first1 == last1) return;
typedef typename std::iterator_traits<SourceIterator1>::value_type Arg1Type;
typedef typename std::iterator_traits<SourceIterator1>::difference_type DiffType;
typedef typename std::iterator_traits<SourceIterator2>::value_type Arg2Type;
typedef typename std::remove_const<typename std::remove_reference<Func>::type>::type FType;
typedef decltype(func(*first1, *first2)) DestType;
DiffType elts = std::distance(first1, last1);
DiffType remaining = elts;
ScopedHandle<DestType*> results{new DestType[elts]};
Mutex mutex;
Cond cond;
bool error = false;
// construct SafeFunctorArg objects so that they can be shared
// between different tasks
Cgu::Callback::SafeFunctorArg<Arg1Type&, Arg2Type&, DestType&> s_task{
Cgu::Callback::make_ref(&ParallelHelper::transform2_func<FType, Arg1Type, Arg2Type, DestType>,
std::forward<Func>(func))
};
Cgu::Callback::SafeFunctor s_fail{
Cgu::Callback::make(&ParallelHelper::fail_func<DiffType>,
&mutex,
&cond,
&error,
&remaining)
};
for (DiffType index = 0; first1 != last1; ++first1, ++first2, ++index) {
std::unique_ptr<const Cgu::Callback::Callback> task_cb(
Cgu::Callback::lambda<>(std::bind(&ParallelHelper::transform2_cb_func<Arg1Type, Arg2Type, DestType, DiffType, SourceIterator1, SourceIterator2>,
s_task,
first1,
first2,
&mutex,
&cond,
results.get(),
index,
&remaining))
);
std::unique_ptr<const Cgu::Callback::Callback> fail_cb(
Cgu::Callback::lambda<>([s_fail] () {s_fail();})
);
tm.add_task(std::move(task_cb), std::move(fail_cb));
}
Mutex::Lock l{mutex};
while (remaining > 0) cond.wait(mutex);
if (error) throw ParallelError();
for (DiffType index = 0; index < elts; ++dest, ++index) {
*dest = std::move(results[index]);
}
}
} // namespace Thread
} // namespace Cgu
#endif // CGU_PARALLEL_H