3 // Copyright (C) 2007 Free Software Foundation, Inc.
5 // This file is part of the GNU ISO C++ Library. This library is free
6 // software; you can redistribute it and/or modify it under the terms
7 // of the GNU General Public License as published by the Free Software
8 // Foundation; either version 2, or (at your option) any later
11 // This library is distributed in the hope that it will be useful, but
12 // WITHOUT ANY WARRANTY; without even the implied warranty of
13 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 // General Public License for more details.
16 // You should have received a copy of the GNU General Public License
17 // along with this library; see the file COPYING. If not, write to
18 // the Free Software Foundation, 59 Temple Place - Suite 330, Boston,
19 // MA 02111-1307, USA.
21 // As a special exception, you may use this file as part of a free
22 // software library without restriction. Specifically, if other files
23 // instantiate templates or use macros or inline functions from this
24 // file, or you compile this file and link it with other files to
25 // produce an executable, this file does not by itself cause the
26 // resulting executable to be covered by the GNU General Public
27 // License. This exception does not however invalidate any other
28 // reasons why the executable file might be covered by the GNU General
31 /** @file parallel/workstealing.h
32 * @brief Parallelization of embarrassingly parallel execution by
33 * means of work-stealing.
35 * Work stealing is described in
37 * R. D. Blumofe and C. E. Leiserson.
38 * Scheduling multithreaded computations by work stealing.
39 * Journal of the ACM, 46(5):720–748, 1999.
41 * This file is a GNU parallel extension to the Standard C++ Library.
44 // Written by Felix Putze.
46 #ifndef _GLIBCXX_PARALLEL_WORKSTEALING_H
47 #define _GLIBCXX_PARALLEL_WORKSTEALING_H 1
49 #include <parallel/parallel.h>
50 #include <parallel/random_number.h>
51 #include <parallel/compatibility.h>
53 namespace __gnu_parallel
56 #define _GLIBCXX_JOB_VOLATILE volatile
58 /** @brief One job for a certain thread. */
59 template<typename _DifferenceTp>
62 typedef _DifferenceTp difference_type;
64 /** @brief First element.
66 * Changed by owning and stealing thread. By stealing thread,
67 * always incremented. */
68 _GLIBCXX_JOB_VOLATILE difference_type first;
70 /** @brief Last element.
72 * Changed by owning thread only. */
73 _GLIBCXX_JOB_VOLATILE difference_type last;
75 /** @brief Number of elements, i. e. @c last-first+1.
77 * Changed by owning thread only. */
78 _GLIBCXX_JOB_VOLATILE difference_type load;
81 /** @brief Work stealing algorithm for random access iterators.
83 * Uses O(1) additional memory. Synchronization at job lists is
84 * done with atomic operations.
85 * @param begin Begin iterator of element sequence.
86 * @param end End iterator of element sequence.
87 * @param op User-supplied functor (comparator, predicate, adding
89 * @param f Functor to "process" an element with op (depends on
90 * desired functionality, e. g. for std::for_each(), ...).
91 * @param r Functor to "add" a single result to the already
92 * processed elements (depends on functionality).
93 * @param base Base value for reduction.
94 * @param output Pointer to position where final result is written to
95 * @param bound Maximum number of elements processed (e. g. for
97 * @return User-supplied functor (that may contain a part of the result).
99 template<typename RandomAccessIterator, typename Op, typename Fu, typename Red, typename Result>
101 for_each_template_random_access_workstealing(RandomAccessIterator begin,
102 RandomAccessIterator end,
104 Result base, Result& output,
105 typename std::iterator_traits<RandomAccessIterator>::difference_type bound)
107 _GLIBCXX_CALL(end - begin)
109 typedef std::iterator_traits<RandomAccessIterator> traits_type;
110 typedef typename traits_type::difference_type difference_type;
113 difference_type chunk_size = static_cast<difference_type>(Settings::workstealing_chunk_size);
116 difference_type length = (bound < 0) ? (end - begin) : bound;
118 // To avoid false sharing in a cache line.
119 const int stride = Settings::cache_line_size * 10 / sizeof(Job<difference_type>) + 1;
121 // Total number of threads currently working.
122 thread_index_t busy = 0;
123 thread_index_t num_threads = get_max_threads();
124 difference_type num_threads_min = num_threads < end - begin ? num_threads : end - begin;
126 omp_lock_t output_lock;
127 omp_init_lock(&output_lock);
129 // No more threads than jobs, at least one thread.
130 difference_type num_threads_max = num_threads_min > 1 ? num_threads_min : 1;
131 num_threads = static_cast<thread_index_t>(num_threads_max);
133 // Create job description array.
134 Job<difference_type> *job = new Job<difference_type>[num_threads * stride];
136 // Write base value to output.
139 #pragma omp parallel shared(busy) num_threads(num_threads)
141 // Initialization phase.
143 // Flags for every thread if it is doing productive work.
144 bool iam_working = false;
147 thread_index_t iam = omp_get_thread_num();
150 Job<difference_type>& my_job = job[iam * stride];
152 // Random number (for work stealing).
153 thread_index_t victim;
155 // Local value for reduction.
156 Result result = Result();
158 // Number of elements to steal in one attempt.
159 difference_type steal;
161 // Every thread has its own random number generator (modulo num_threads).
162 random_number rand_gen(iam, num_threads);
165 // This thread is currently working.
170 // How many jobs per thread? last thread gets the rest.
171 my_job.first = static_cast<difference_type>(iam * (length / num_threads));
173 my_job.last = (iam == (num_threads - 1)) ? (length - 1) : ((iam + 1) * (length / num_threads) - 1);
174 my_job.load = my_job.last - my_job.first + 1;
176 // Init result with first value (to have a base value for reduction).
177 if (my_job.first <= my_job.last)
179 // Cannot use volatile variable directly.
180 difference_type my_first = my_job.first;
181 result = f(op, begin + my_first);
186 RandomAccessIterator current;
191 // Work on own or stolen start
194 // Work until no productive thread left.
195 #pragma omp flush(busy)
197 // Thread has own work to do
198 while (my_job.first <= my_job.last)
200 // fetch-and-add call
201 // Reserve current job block (size chunk_size) in my queue.
202 difference_type current_job = fetch_and_add<difference_type>(&(my_job.first), chunk_size);
204 // Update load, to make the three values consistent,
205 // first might have been changed in the meantime
206 my_job.load = my_job.last - my_job.first + 1;
207 for (difference_type job_counter = 0; job_counter < chunk_size && current_job <= my_job.last; job_counter++)
210 current = begin + current_job;
214 result = r(result, f(op, current));
217 #pragma omp flush(busy)
221 // After reaching this point, a thread's job list is empty.
225 // This thread no longer has work.
231 difference_type supposed_first, supposed_last, supposed_load;
234 // Find random nonempty deque (not own) and do consistency check.
236 #pragma omp flush(busy)
238 supposed_first = job[victim * stride].first;
239 supposed_last = job[victim * stride].last;
240 supposed_load = job[victim * stride].load;
243 && ((supposed_load <= 0) || ((supposed_first + supposed_load - 1) != supposed_last)));
248 if (supposed_load > 0)
250 // Has work and work to do.
251 // Number of elements to steal (at least one).
252 steal = (supposed_load < 2) ? 1 : supposed_load / 2;
254 // Protects against stealing threads
255 // omp_set_lock(&(job[victim * stride].lock));
257 // Push victim's start forward.
258 difference_type stolen_first = fetch_and_add<difference_type>(&(job[victim * stride].first), steal);
259 difference_type stolen_try = stolen_first + steal - difference_type(1);
261 // Protects against working thread
262 // omp_unset_lock(&(job[victim * stride].lock));
264 my_job.first = stolen_first;
266 // Avoid std::min dependencies.
267 my_job.last = stolen_try < supposed_last ? stolen_try : supposed_last;
269 my_job.load = my_job.last - my_job.first + 1;
271 //omp_unset_lock(&(my_job.lock));
274 // Has potential work again.
278 #pragma omp flush(busy)
280 #pragma omp flush(busy)
281 } // end while busy > 0
282 // Add accumulated result to output.
283 omp_set_lock(&output_lock);
284 output = r(output, result);
285 omp_unset_lock(&output_lock);
287 //omp_destroy_lock(&(my_job.lock));
292 // Points to last element processed (needed as return value for
293 // some algorithms like transform)
294 f.finish_iterator = begin + length;
296 omp_destroy_lock(&output_lock);