1 //---------------------------------------------------------------------------//
2 // Copyright (c) 2015 Jakub Szuppe <j.szuppe@gmail.com>
4 // Distributed under the Boost Software License, Version 1.0
5 // See accompanying file LICENSE_1_0.txt or copy at
6 // http://www.boost.org/LICENSE_1_0.txt
8 // See http://boostorg.github.com/compute for more information.
9 //---------------------------------------------------------------------------//
11 #ifndef BOOST_COMPUTE_ALGORITHM_DETAIL_REDUCE_BY_KEY_WITH_SCAN_HPP
12 #define BOOST_COMPUTE_ALGORITHM_DETAIL_REDUCE_BY_KEY_WITH_SCAN_HPP
17 #include <boost/compute/command_queue.hpp>
18 #include <boost/compute/functional.hpp>
19 #include <boost/compute/algorithm/inclusive_scan.hpp>
20 #include <boost/compute/container/vector.hpp>
21 #include <boost/compute/container/detail/scalar.hpp>
22 #include <boost/compute/detail/meta_kernel.hpp>
23 #include <boost/compute/detail/iterator_range_size.hpp>
24 #include <boost/compute/detail/read_write_single_value.hpp>
25 #include <boost/compute/type_traits.hpp>
26 #include <boost/compute/utility/program_cache.hpp>
34 /// Fills \p new_keys_first with unsigned integer keys generated from vector
35 /// of original keys \p keys_first. New keys can be distinguish by simple equality
38 /// \param keys_first iterator pointing to the first key
39 /// \param number_of_keys number of keys
40 /// \param predicate binary predicate for key comparison
41 /// \param new_keys_first iterator pointing to the new keys vector
42 /// \param preferred_work_group_size preferred work group size
43 /// \param queue command queue to perform the operation
45 /// Binary function \p predicate must take two keys as arguments and
46 /// return true only if they are considered the same.
48 /// The first new key equals zero and the last equals number of unique keys
51 /// No local memory usage.
52 template<class InputKeyIterator, class BinaryPredicate>
53 inline void generate_uint_keys(InputKeyIterator keys_first,
54 size_t number_of_keys,
55 BinaryPredicate predicate,
56 vector<uint_>::iterator new_keys_first,
57 size_t preferred_work_group_size,
61 std::iterator_traits<InputKeyIterator>::value_type key_type;
63 detail::meta_kernel k("reduce_by_key_new_key_flags");
64 k.add_set_arg<const uint_>("count", uint_(number_of_keys));
67 k.decl<const uint_>("gid") << " = get_global_id(0);\n" <<
68 k.decl<uint_>("value") << " = 0;\n" <<
69 "if(gid >= count){\n return;\n}\n" <<
71 k.decl<key_type>("key") << " = " <<
72 keys_first[k.var<const uint_>("gid")] << ";\n" <<
73 k.decl<key_type>("previous_key") << " = " <<
74 keys_first[k.var<const uint_>("gid - 1")] << ";\n" <<
75 " value = " << predicate(k.var<key_type>("previous_key"),
76 k.var<key_type>("key")) <<
81 new_keys_first[k.var<const uint_>("gid")] << " = value;\n";
83 const context &context = queue.get_context();
84 kernel kernel = k.compile(context);
86 size_t work_group_size = preferred_work_group_size;
87 size_t work_groups_no = static_cast<size_t>(
88 std::ceil(float(number_of_keys) / work_group_size)
91 queue.enqueue_1d_range_kernel(kernel,
93 work_groups_no * work_group_size,
96 inclusive_scan(new_keys_first, new_keys_first + number_of_keys,
97 new_keys_first, queue);
101 /// Calculate carry-out for each work group.
102 /// Carry-out is a pair of the last key processed by a work group and sum of all
103 /// values under this key in this work group.
104 template<class InputValueIterator, class OutputValueIterator, class BinaryFunction>
105 inline void carry_outs(vector<uint_>::iterator keys_first,
106 InputValueIterator values_first,
108 vector<uint_>::iterator carry_out_keys_first,
109 OutputValueIterator carry_out_values_first,
110 BinaryFunction function,
111 size_t work_group_size,
112 command_queue &queue)
115 std::iterator_traits<OutputValueIterator>::value_type value_out_type;
117 detail::meta_kernel k("reduce_by_key_with_scan_carry_outs");
118 k.add_set_arg<const uint_>("count", uint_(count));
119 size_t local_keys_arg = k.add_arg<uint_ *>(memory_object::local_memory, "lkeys");
120 size_t local_vals_arg = k.add_arg<value_out_type *>(memory_object::local_memory, "lvals");
123 k.decl<const uint_>("gid") << " = get_global_id(0);\n" <<
124 k.decl<const uint_>("wg_size") << " = get_local_size(0);\n" <<
125 k.decl<const uint_>("lid") << " = get_local_id(0);\n" <<
126 k.decl<const uint_>("group_id") << " = get_group_id(0);\n" <<
128 k.decl<uint_>("key") << ";\n" <<
129 k.decl<value_out_type>("value") << ";\n" <<
130 "if(gid < count){\n" <<
131 k.var<uint_>("key") << " = " <<
132 keys_first[k.var<const uint_>("gid")] << ";\n" <<
133 k.var<value_out_type>("value") << " = " <<
134 values_first[k.var<const uint_>("gid")] << ";\n" <<
135 "lkeys[lid] = key;\n" <<
136 "lvals[lid] = value;\n" <<
139 // Calculate carry out for each work group by performing Hillis/Steele scan
140 // where only last element (key-value pair) is saved
141 k.decl<value_out_type>("result") << " = value;\n" <<
142 k.decl<uint_>("other_key") << ";\n" <<
143 k.decl<value_out_type>("other_value") << ";\n" <<
145 "for(" << k.decl<uint_>("offset") << " = 1; " <<
146 "offset < wg_size; offset *= 2){\n"
147 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
148 " if(lid >= offset){\n"
149 " other_key = lkeys[lid - offset];\n" <<
150 " if(other_key == key){\n" <<
151 " other_value = lvals[lid - offset];\n" <<
152 " result = " << function(k.var<value_out_type>("result"),
153 k.var<value_out_type>("other_value")) << ";\n" <<
156 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
157 " lvals[lid] = result;\n" <<
161 "if(lid == (wg_size - 1)){\n" <<
162 carry_out_keys_first[k.var<const uint_>("group_id")] << " = key;\n" <<
163 carry_out_values_first[k.var<const uint_>("group_id")] << " = result;\n" <<
166 size_t work_groups_no = static_cast<size_t>(
167 std::ceil(float(count) / work_group_size)
170 const context &context = queue.get_context();
171 kernel kernel = k.compile(context);
172 kernel.set_arg(local_keys_arg, local_buffer<uint_>(work_group_size));
173 kernel.set_arg(local_vals_arg, local_buffer<value_out_type>(work_group_size));
175 queue.enqueue_1d_range_kernel(kernel,
177 work_groups_no * work_group_size,
182 /// Calculate carry-in by performing inclusive scan by key on carry-outs vector.
183 template<class OutputValueIterator, class BinaryFunction>
184 inline void carry_ins(vector<uint_>::iterator carry_out_keys_first,
185 OutputValueIterator carry_out_values_first,
186 OutputValueIterator carry_in_values_first,
187 size_t carry_out_size,
188 BinaryFunction function,
189 size_t work_group_size,
190 command_queue &queue)
193 std::iterator_traits<OutputValueIterator>::value_type value_out_type;
195 uint_ values_pre_work_item = static_cast<uint_>(
196 std::ceil(float(carry_out_size) / work_group_size)
199 detail::meta_kernel k("reduce_by_key_with_scan_carry_ins");
200 k.add_set_arg<const uint_>("carry_out_size", uint_(carry_out_size));
201 k.add_set_arg<const uint_>("values_per_work_item", values_pre_work_item);
202 size_t local_keys_arg = k.add_arg<uint_ *>(memory_object::local_memory, "lkeys");
203 size_t local_vals_arg = k.add_arg<value_out_type *>(memory_object::local_memory, "lvals");
206 k.decl<uint_>("id") << " = get_global_id(0) * values_per_work_item;\n" <<
207 k.decl<uint_>("idx") << " = id;\n" <<
208 k.decl<const uint_>("wg_size") << " = get_local_size(0);\n" <<
209 k.decl<const uint_>("lid") << " = get_local_id(0);\n" <<
210 k.decl<const uint_>("group_id") << " = get_group_id(0);\n" <<
212 k.decl<uint_>("key") << ";\n" <<
213 k.decl<value_out_type>("value") << ";\n" <<
214 k.decl<uint_>("previous_key") << ";\n" <<
215 k.decl<value_out_type>("result") << ";\n" <<
217 "if(id < carry_out_size){\n" <<
218 k.var<uint_>("previous_key") << " = " <<
219 carry_out_keys_first[k.var<const uint_>("id")] << ";\n" <<
220 k.var<value_out_type>("result") << " = " <<
221 carry_out_values_first[k.var<const uint_>("id")] << ";\n" <<
222 carry_in_values_first[k.var<const uint_>("id")] << " = result;\n" <<
225 k.decl<const uint_>("end") << " = (id + values_per_work_item) <= carry_out_size" <<
226 " ? (values_per_work_item + id) : carry_out_size;\n" <<
228 "for(idx = idx + 1; idx < end; idx += 1){\n" <<
229 " key = " << carry_out_keys_first[k.var<const uint_>("idx")] << ";\n" <<
230 " value = " << carry_out_values_first[k.var<const uint_>("idx")] << ";\n" <<
231 " if(previous_key == key){\n" <<
232 " result = " << function(k.var<value_out_type>("result"),
233 k.var<value_out_type>("value")) << ";\n" <<
237 " " << carry_in_values_first[k.var<const uint_>("idx")] << " = result;\n" <<
238 " previous_key = key;\n"
241 // save the last key and result to local memory
242 "lkeys[lid] = previous_key;\n" <<
243 "lvals[lid] = result;\n" <<
245 // Hillis/Steele scan
246 "for(" << k.decl<uint_>("offset") << " = 1; " <<
247 "offset < wg_size; offset *= 2){\n"
248 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
249 " if(lid >= offset){\n"
250 " key = lkeys[lid - offset];\n" <<
251 " if(previous_key == key){\n" <<
252 " value = lvals[lid - offset];\n" <<
253 " result = " << function(k.var<value_out_type>("result"),
254 k.var<value_out_type>("value")) << ";\n" <<
257 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
258 " lvals[lid] = result;\n" <<
260 "barrier(CLK_LOCAL_MEM_FENCE);\n" <<
263 // load key-value reduced by previous work item
264 " previous_key = lkeys[lid - 1];\n" <<
265 " result = lvals[lid - 1];\n" <<
268 // add key-value reduced by previous work item
269 "for(idx = id; idx < id + values_per_work_item; idx += 1){\n" <<
270 // make sure all carry-ins are saved in global memory
271 " barrier( CLK_GLOBAL_MEM_FENCE );\n" <<
272 " if(lid > 0 && idx < carry_out_size) {\n"
273 " key = " << carry_out_keys_first[k.var<const uint_>("idx")] << ";\n" <<
274 " value = " << carry_in_values_first[k.var<const uint_>("idx")] << ";\n" <<
275 " if(previous_key == key){\n" <<
276 " value = " << function(k.var<value_out_type>("result"),
277 k.var<value_out_type>("value")) << ";\n" <<
279 " " << carry_in_values_first[k.var<const uint_>("idx")] << " = value;\n" <<
284 const context &context = queue.get_context();
285 kernel kernel = k.compile(context);
286 kernel.set_arg(local_keys_arg, local_buffer<uint_>(work_group_size));
287 kernel.set_arg(local_vals_arg, local_buffer<value_out_type>(work_group_size));
289 queue.enqueue_1d_range_kernel(kernel,
297 /// Perform final reduction by key. Each work item:
298 /// 1. Perform local work-group reduction (Hillis/Steele scan)
299 /// 2. Add carry-in (if keys are right)
300 /// 3. Save reduced value if next key is different than processed one
301 template<class InputKeyIterator, class InputValueIterator,
302 class OutputKeyIterator, class OutputValueIterator,
303 class BinaryFunction>
304 inline void final_reduction(InputKeyIterator keys_first,
305 InputValueIterator values_first,
306 OutputKeyIterator keys_result,
307 OutputValueIterator values_result,
309 BinaryFunction function,
310 vector<uint_>::iterator new_keys_first,
311 vector<uint_>::iterator carry_in_keys_first,
312 OutputValueIterator carry_in_values_first,
313 size_t carry_in_size,
314 size_t work_group_size,
315 command_queue &queue)
318 std::iterator_traits<OutputValueIterator>::value_type value_out_type;
320 detail::meta_kernel k("reduce_by_key_with_scan_final_reduction");
321 k.add_set_arg<const uint_>("count", uint_(count));
322 size_t local_keys_arg = k.add_arg<uint_ *>(memory_object::local_memory, "lkeys");
323 size_t local_vals_arg = k.add_arg<value_out_type *>(memory_object::local_memory, "lvals");
326 k.decl<const uint_>("gid") << " = get_global_id(0);\n" <<
327 k.decl<const uint_>("wg_size") << " = get_local_size(0);\n" <<
328 k.decl<const uint_>("lid") << " = get_local_id(0);\n" <<
329 k.decl<const uint_>("group_id") << " = get_group_id(0);\n" <<
331 k.decl<uint_>("key") << ";\n" <<
332 k.decl<value_out_type>("value") << ";\n"
334 "if(gid < count){\n" <<
335 k.var<uint_>("key") << " = " <<
336 new_keys_first[k.var<const uint_>("gid")] << ";\n" <<
337 k.var<value_out_type>("value") << " = " <<
338 values_first[k.var<const uint_>("gid")] << ";\n" <<
339 "lkeys[lid] = key;\n" <<
340 "lvals[lid] = value;\n" <<
343 // Hillis/Steele scan
344 k.decl<value_out_type>("result") << " = value;\n" <<
345 k.decl<uint_>("other_key") << ";\n" <<
346 k.decl<value_out_type>("other_value") << ";\n" <<
348 "for(" << k.decl<uint_>("offset") << " = 1; " <<
349 "offset < wg_size ; offset *= 2){\n"
350 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
351 " if(lid >= offset) {\n" <<
352 " other_key = lkeys[lid - offset];\n" <<
353 " if(other_key == key){\n" <<
354 " other_value = lvals[lid - offset];\n" <<
355 " result = " << function(k.var<value_out_type>("result"),
356 k.var<value_out_type>("other_value")) << ";\n" <<
359 " barrier(CLK_LOCAL_MEM_FENCE);\n" <<
360 " lvals[lid] = result;\n" <<
363 "if(gid >= count) {\n return;\n};\n" <<
365 k.decl<const bool>("save") << " = (gid < (count - 1)) ?"
366 << new_keys_first[k.var<const uint_>("gid + 1")] << " != key" <<
370 k.decl<uint_>("carry_in_key") << ";\n" <<
371 "if(group_id > 0 && save) {\n" <<
372 " carry_in_key = " << carry_in_keys_first[k.var<const uint_>("group_id - 1")] << ";\n" <<
373 " if(key == carry_in_key){\n" <<
374 " other_value = " << carry_in_values_first[k.var<const uint_>("group_id - 1")] << ";\n" <<
375 " result = " << function(k.var<value_out_type>("result"),
376 k.var<value_out_type>("other_value")) << ";\n" <<
380 // Save result only if the next key is different or it's the last element.
382 keys_result[k.var<uint_>("key")] << " = " << keys_first[k.var<const uint_>("gid")] << ";\n" <<
383 values_result[k.var<uint_>("key")] << " = result;\n" <<
387 size_t work_groups_no = static_cast<size_t>(
388 std::ceil(float(count) / work_group_size)
391 const context &context = queue.get_context();
392 kernel kernel = k.compile(context);
393 kernel.set_arg(local_keys_arg, local_buffer<uint_>(work_group_size));
394 kernel.set_arg(local_vals_arg, local_buffer<value_out_type>(work_group_size));
396 queue.enqueue_1d_range_kernel(kernel,
398 work_groups_no * work_group_size,
403 /// Returns preferred work group size for reduce by key with scan algorithm.
404 template<class KeyType, class ValueType>
405 inline size_t get_work_group_size(const device& device)
407 std::string cache_key = std::string("__boost_reduce_by_key_with_scan")
408 + "k_" + type_name<KeyType>() + "_v_" + type_name<ValueType>();
411 boost::shared_ptr<parameter_cache> parameters =
412 detail::parameter_cache::get_global_cache(device);
415 static_cast<size_t>(parameters->get(cache_key, "wgsize", 256)),
416 static_cast<size_t>(device.get_info<CL_DEVICE_MAX_WORK_GROUP_SIZE>())
422 /// 1. For each work group carry-out value is calculated (it's done by key-oriented
423 /// Hillis/Steele scan). Carry-out is a pair of the last key processed by work
424 /// group and sum of all values under this key in work group.
425 /// 2. From every carry-out carry-in is calculated by performing inclusive scan
427 /// 3. Final reduction by key is performed (key-oriented Hillis/Steele scan),
428 /// carry-in values are added where needed.
429 template<class InputKeyIterator, class InputValueIterator,
430 class OutputKeyIterator, class OutputValueIterator,
431 class BinaryFunction, class BinaryPredicate>
432 inline size_t reduce_by_key_with_scan(InputKeyIterator keys_first,
433 InputKeyIterator keys_last,
434 InputValueIterator values_first,
435 OutputKeyIterator keys_result,
436 OutputValueIterator values_result,
437 BinaryFunction function,
438 BinaryPredicate predicate,
439 command_queue &queue)
442 std::iterator_traits<InputValueIterator>::value_type value_type;
444 std::iterator_traits<InputKeyIterator>::value_type key_type;
446 std::iterator_traits<OutputValueIterator>::value_type value_out_type;
448 const context &context = queue.get_context();
449 size_t count = detail::iterator_range_size(keys_first, keys_last);
455 const device &device = queue.get_device();
456 size_t work_group_size = get_work_group_size<value_type, key_type>(device);
458 // Replace original key with unsigned integer keys generated based on given
459 // predicate. New key is also an index for keys_result and values_result vectors,
460 // which points to place where reduced value should be saved.
461 vector<uint_> new_keys(count, context);
462 vector<uint_>::iterator new_keys_first = new_keys.begin();
463 generate_uint_keys(keys_first, count, predicate, new_keys_first,
464 work_group_size, queue);
466 // Calculate carry-out and carry-in vectors size
467 const size_t carry_out_size = static_cast<size_t>(
468 std::ceil(float(count) / work_group_size)
470 vector<uint_> carry_out_keys(carry_out_size, context);
471 vector<value_out_type> carry_out_values(carry_out_size, context);
472 carry_outs(new_keys_first, values_first, count, carry_out_keys.begin(),
473 carry_out_values.begin(), function, work_group_size, queue);
475 vector<value_out_type> carry_in_values(carry_out_size, context);
476 carry_ins(carry_out_keys.begin(), carry_out_values.begin(),
477 carry_in_values.begin(), carry_out_size, function, work_group_size,
480 final_reduction(keys_first, values_first, keys_result, values_result,
481 count, function, new_keys_first, carry_out_keys.begin(),
482 carry_in_values.begin(), carry_out_size, work_group_size,
485 const size_t result = read_single_value<uint_>(new_keys.get_buffer(),
491 /// Return true if requirements for running reduce by key with scan on given
492 /// device are met (at least one work group of preferred size can be run).
493 template<class InputKeyIterator, class InputValueIterator,
494 class OutputKeyIterator, class OutputValueIterator>
495 bool reduce_by_key_with_scan_requirements_met(InputKeyIterator keys_first,
496 InputValueIterator values_first,
497 OutputKeyIterator keys_result,
498 OutputValueIterator values_result,
500 command_queue &queue)
503 std::iterator_traits<InputValueIterator>::value_type value_type;
505 std::iterator_traits<InputKeyIterator>::value_type key_type;
507 std::iterator_traits<OutputValueIterator>::value_type value_out_type;
512 (void) values_result;
514 const device &device = queue.get_device();
515 // device must have dedicated local memory storage
516 if(device.get_info<CL_DEVICE_LOCAL_MEM_TYPE>() != CL_LOCAL)
521 // local memory size in bytes (per compute unit)
522 const size_t local_mem_size = device.get_info<CL_DEVICE_LOCAL_MEM_SIZE>();
524 // preferred work group size
525 size_t work_group_size = get_work_group_size<key_type, value_type>(device);
527 // local memory size needed to perform parallel reduction
528 size_t required_local_mem_size = 0;
530 required_local_mem_size += sizeof(uint_) * work_group_size;
531 // reduced values size
532 required_local_mem_size += sizeof(value_out_type) * work_group_size;
534 return (required_local_mem_size <= local_mem_size);
537 } // end detail namespace
538 } // end compute namespace
539 } // end boost namespace
541 #endif // BOOST_COMPUTE_ALGORITHM_DETAIL_REDUCE_BY_KEY_WITH_SCAN_HPP