ae.utils.parallelism source code

1 /**
2  * ae.utils.parallelism
3  *
4  * License:
5  *   This Source Code Form is subject to the terms of
6  *   the Mozilla Public License, v. 2.0. If a copy of
7  *   the MPL was not distributed with this file, You
8  *   can obtain one at http://mozilla.org/MPL/2.0/.
9  *
10  * Authors:
11  *   Vladimir Panteleev <ae@cy.md>
12  */
13 
14 module ae.utils.parallelism;
15 
16 import std.algorithm.mutation;
17 import std.algorithm.sorting;
18 import std.parallelism;
19 import std.range : chunks, iota;
20 
21 // https://gist.github.com/63e139a16b9b278fb5d449ace611e7b8
22 
23 /// Sort `r` using all CPU cores.
24 auto parallelSort(alias less = "a < b", R)(R r)
25 {
26 	auto impl(size_t depth = 0)(R order)
27 	{
28 		static if (depth < 8)
29 			if ((1L << depth) < totalCPUs)
30 				foreach (chunk; order.chunks(order.length / 2 + 1).parallel(1))
31 					impl!(depth + 1)(chunk);
32 
33 		return order.sort!(less, SwapStrategy.stable, R);
34 	}
35 	return impl(r);
36 }
37 
38 unittest
39 {
40 	assert([3, 1, 2].parallelSort.release == [1, 2, 3]);
41 }
42 
43 
44 /// Parallel map.  Like TaskPool.amap, but uses functors for
45 /// predicates instead of alias arguments, and as such does not have
46 /// the multiple-context problem.
47 /// https://forum.dlang.org/post/qnigarkuxxnqwdernhzv@forum.dlang.org
48 auto parallelEagerMap(R, Pred)(R input, Pred pred, size_t workUnitSize = 0)
49 {
50 	if (workUnitSize == 0)
51 		workUnitSize = taskPool.defaultWorkUnitSize(input.length);
52 	alias RT = typeof(pred(input[0]));
53 	auto result = new RT[input.length];
54 	foreach (i; input.length.iota.parallel(workUnitSize))
55 		result[i] = pred(input[i]);
56 	return result;
57 }
58 
59 unittest
60 {
61 	assert([1, 2, 3].parallelEagerMap((int n) => n + 1) == [2, 3, 4]);
62 }