gcc/libstdc++-v3/include/parallel/workstealing.h

313 lines
9.4 KiB
C++
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

// -*- C++ -*-
// Copyright (C) 2007, 2008, 2009, 2010 Free Software Foundation, Inc.
//
// This file is part of the GNU ISO C++ Library. This library 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 3, 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
// General Public License for more details.
// Under Section 7 of GPL version 3, you are granted additional
// permissions described in the GCC Runtime Library Exception, version
// 3.1, as published by the Free Software Foundation.
// You should have received a copy of the GNU General Public License and
// a copy of the GCC Runtime Library Exception along with this program;
// see the files COPYING3 and COPYING.RUNTIME respectively. If not, see
// <http://www.gnu.org/licenses/>.
/** @file parallel/workstealing.h
* @brief Parallelization of embarrassingly parallel execution by
* means of work-stealing.
*
* Work stealing is described in
*
* R. D. Blumofe and C. E. Leiserson.
* Scheduling multithreaded computations by work stealing.
* Journal of the ACM, 46(5):720748, 1999.
*
* This file is a GNU parallel extension to the Standard C++ Library.
*/
// Written by Felix Putze.
#ifndef _GLIBCXX_PARALLEL_WORKSTEALING_H
#define _GLIBCXX_PARALLEL_WORKSTEALING_H 1
#include <parallel/parallel.h>
#include <parallel/random_number.h>
#include <parallel/compatibility.h>
namespace __gnu_parallel
{
#define _GLIBCXX_JOB_VOLATILE volatile
/** @brief One __job for a certain thread. */
template<typename _DifferenceTp>
struct _Job
{
typedef _DifferenceTp _DifferenceType;
/** @brief First element.
*
* Changed by owning and stealing thread. By stealing thread,
* always incremented. */
_GLIBCXX_JOB_VOLATILE _DifferenceType _M_first;
/** @brief Last element.
*
* Changed by owning thread only. */
_GLIBCXX_JOB_VOLATILE _DifferenceType _M_last;
/** @brief Number of elements, i.e. @c _M_last-_M_first+1.
*
* Changed by owning thread only. */
_GLIBCXX_JOB_VOLATILE _DifferenceType _M_load;
};
/** @brief Work stealing algorithm for random access iterators.
*
* Uses O(1) additional memory. Synchronization at job lists is
* done with atomic operations.
* @param __begin Begin iterator of element sequence.
* @param __end End iterator of element sequence.
* @param __op User-supplied functor (comparator, predicate, adding
* functor, ...).
* @param __f Functor to @a process an element with __op (depends on
* desired functionality, e. g. for std::for_each(), ...).
* @param __r Functor to @a add a single __result to the already
* processed elements (depends on functionality).
* @param __base Base value for reduction.
* @param __output Pointer to position where final result is written to
* @param __bound Maximum number of elements processed (e. g. for
* std::count_n()).
* @return User-supplied functor (that may contain a part of the result).
*/
template<typename _RAIter,
typename _Op,
typename _Fu,
typename _Red,
typename _Result>
_Op
__for_each_template_random_access_workstealing(_RAIter __begin,
_RAIter __end, _Op __op,
_Fu& __f, _Red __r,
_Result __base,
_Result& __output,
typename std::iterator_traits<_RAIter>::difference_type __bound)
{
_GLIBCXX_CALL(__end - __begin)
typedef std::iterator_traits<_RAIter> _TraitsType;
typedef typename _TraitsType::difference_type _DifferenceType;
const _Settings& __s = _Settings::get();
_DifferenceType __chunk_size =
static_cast<_DifferenceType>(__s.workstealing_chunk_size);
// How many jobs?
_DifferenceType __length = (__bound < 0) ? (__end - __begin) : __bound;
// To avoid false sharing in a cache line.
const int __stride = (__s.cache_line_size * 10
/ sizeof(_Job<_DifferenceType>) + 1);
// Total number of threads currently working.
_ThreadIndex __busy = 0;
_Job<_DifferenceType> *__job;
omp_lock_t __output_lock;
omp_init_lock(&__output_lock);
// Write base value to output.
__output = __base;
// No more threads than jobs, at least one thread.
_ThreadIndex __num_threads = __gnu_parallel::max<_ThreadIndex>
(1, __gnu_parallel::min<_DifferenceType>(__length,
__get_max_threads()));
# pragma omp parallel shared(__busy) num_threads(__num_threads)
{
# pragma omp single
{
__num_threads = omp_get_num_threads();
// Create job description array.
__job = new _Job<_DifferenceType>[__num_threads * __stride];
}
// Initialization phase.
// Flags for every thread if it is doing productive work.
bool __iam_working = false;
// Thread id.
_ThreadIndex __iam = omp_get_thread_num();
// This job.
_Job<_DifferenceType>& __my_job = __job[__iam * __stride];
// Random number (for work stealing).
_ThreadIndex __victim;
// Local value for reduction.
_Result __result = _Result();
// Number of elements to steal in one attempt.
_DifferenceType __steal;
// Every thread has its own random number generator
// (modulo __num_threads).
_RandomNumber __rand_gen(__iam, __num_threads);
// This thread is currently working.
# pragma omp atomic
++__busy;
__iam_working = true;
// How many jobs per thread? last thread gets the rest.
__my_job._M_first = static_cast<_DifferenceType>
(__iam * (__length / __num_threads));
__my_job._M_last = (__iam == (__num_threads - 1)
? (__length - 1)
: ((__iam + 1) * (__length / __num_threads) - 1));
__my_job._M_load = __my_job._M_last - __my_job._M_first + 1;
// Init result with _M_first value (to have a base value for reduction)
if (__my_job._M_first <= __my_job._M_last)
{
// Cannot use volatile variable directly.
_DifferenceType __my_first = __my_job._M_first;
__result = __f(__op, __begin + __my_first);
++__my_job._M_first;
--__my_job._M_load;
}
_RAIter __current;
# pragma omp barrier
// Actual work phase
// Work on own or stolen current start
while (__busy > 0)
{
// Work until no productive thread left.
# pragma omp flush(__busy)
// Thread has own work to do
while (__my_job._M_first <= __my_job._M_last)
{
// fetch-and-add call
// Reserve current job block (size __chunk_size) in my queue.
_DifferenceType __current_job =
__fetch_and_add<_DifferenceType>(&(__my_job._M_first),
__chunk_size);
// Update _M_load, to make the three values consistent,
// _M_first might have been changed in the meantime
__my_job._M_load = __my_job._M_last - __my_job._M_first + 1;
for (_DifferenceType __job_counter = 0;
__job_counter < __chunk_size
&& __current_job <= __my_job._M_last;
++__job_counter)
{
// Yes: process it!
__current = __begin + __current_job;
++__current_job;
// Do actual work.
__result = __r(__result, __f(__op, __current));
}
# pragma omp flush(__busy)
}
// After reaching this point, a thread's __job list is empty.
if (__iam_working)
{
// This thread no longer has work.
# pragma omp atomic
--__busy;
__iam_working = false;
}
_DifferenceType __supposed_first, __supposed_last,
__supposed_load;
do
{
// Find random nonempty deque (not own), do consistency check.
__yield();
# pragma omp flush(__busy)
__victim = __rand_gen();
__supposed_first = __job[__victim * __stride]._M_first;
__supposed_last = __job[__victim * __stride]._M_last;
__supposed_load = __job[__victim * __stride]._M_load;
}
while (__busy > 0
&& ((__supposed_load <= 0)
|| ((__supposed_first + __supposed_load - 1)
!= __supposed_last)));
if (__busy == 0)
break;
if (__supposed_load > 0)
{
// Has work and work to do.
// Number of elements to steal (at least one).
__steal = (__supposed_load < 2) ? 1 : __supposed_load / 2;
// Push __victim's current start forward.
_DifferenceType __stolen_first =
__fetch_and_add<_DifferenceType>
(&(__job[__victim * __stride]._M_first), __steal);
_DifferenceType __stolen_try = (__stolen_first + __steal
- _DifferenceType(1));
__my_job._M_first = __stolen_first;
__my_job._M_last = __gnu_parallel::min(__stolen_try,
__supposed_last);
__my_job._M_load = __my_job._M_last - __my_job._M_first + 1;
// Has potential work again.
# pragma omp atomic
++__busy;
__iam_working = true;
# pragma omp flush(__busy)
}
# pragma omp flush(__busy)
} // end while __busy > 0
// Add accumulated result to output.
omp_set_lock(&__output_lock);
__output = __r(__output, __result);
omp_unset_lock(&__output_lock);
}
delete[] __job;
// Points to last element processed (needed as return value for
// some algorithms like transform)
__f._M_finish_iterator = __begin + __length;
omp_destroy_lock(&__output_lock);
return __op;
}
} // end namespace
#endif /* _GLIBCXX_PARALLEL_WORKSTEALING_H */