ForkJoinPool を使ってみる

1
import java.util.Arrays;
2
import java.util.concurrent.RecursiveAction;
3

4
public class SortAction extends RecursiveAction {
5

6
    final long[] array;
7
    final int lo, hi;
8

9
    SortAction(long[] array, int lo, int hi) {
10
        this.array = array;
11
        this.lo = lo;
12
        this.hi = hi;
13
    }
14

15
    SortAction(long[] array) {
16
        this(array, 0, array.length);
17
    }
18

19
    @Override
20
    protected void compute() {
21
        if (hi - lo < THRESHOLD) {
22
            sortSequentially(lo, hi);
23
        } else {
24
            int mid = (lo + hi) >>> 1; // 2 で割るのと同等
25

26
            invokeAll(new SortAction(array, lo, mid), new SortAction(array, mid, hi));
27

28
            merge(lo, mid, hi);
29
        }
30
    }
31

32
    // implementation details follow:
33
    static final int THRESHOLD = 1000;
34

35
    void sortSequentially(int lo, int hi) {
36
        Arrays.sort(array, lo, hi);
37
    }
38

39
    void merge(int lo, int mid, int hi) {
40
        // array を分割した後ろは再利用している
41
        long[] buf = Arrays.copyOfRange(array, lo, mid);
42
        for (int i = 0, j = lo, k = mid; i < buf.length; j++)  {
43
            array[j] = (k == hi || buf[i] < array[k])
44
                ? buf[i++]
45
                : array[k++];
46
        }
47
    }
48
}

1
import java.util.ArrayList;
2
import java.util.Collections;
3
import java.util.Comparator;
4
import java.util.List;
5
import java.util.concurrent.RecursiveTask;
6

7
public class MergeSortTask<T extends Comparable<? super T>, U extends Comparator<T>> extends RecursiveTask<List<T>> {
8

9
    /** ソート対象リスト。 */
10
    private final List<T> xs;
11

12
    private final Comparator<T> comparator;
13

14
    public MergeSortTask(List<T> xs, Comparator<T> comparator) {
15
        this.xs = xs;
16
        this.comparator = comparator;
17
    }
18

19
    @Override
20
    protected List<T> compute() {
21
        // リストが空、もしくは要素数が 1 の場合はソート済みとして返す
22
        if (xs.size() <= 1) {
23
            return xs;
24
        }
25

26
        // リストを左右に分割
27
        final var mid = Math.floorDiv(xs.size(), 2);
28
        final var left = new MergeSortTask<>(xs.subList(0, mid), comparator);
29
        final var right = new MergeSortTask<>(xs.subList(mid, xs.size()), comparator);
30

31
        // 右側のソートの非同期実行を調整
32
        right.fork();
33

34
        final var sortedLeft = left.compute();
35
        final var sortedRight = right.join();
36

37
        final var sorted = new ArrayList<T>();
38
        int l = 0, r = 0;
39
        while (l < sortedLeft.size() && r < sortedRight.size()) {
40
            final var x = sortedLeft.get(l);
41
            final var y = sortedRight.get(r);
42

43
            if (comparator.compare(x, y) <= 0) {
44
                sorted.add(x);
45
                l++;
46
            } else {
47
                sorted.add(y);
48
                r++;
49
            }
50
        }
51

52
        while (l < sortedLeft.size()) {
53
            sorted.add(sortedLeft.get(l++));
54
        }
55

56
        while (r < sortedRight.size()) {
57
            sorted.add(sortedRight.get(r++));
58
        }
59

60
        return Collections.unmodifiableList(sorted);
61
    }
62
}

1
    public ForkJoinPool(int parallelism,
2
                        ForkJoinWorkerThreadFactory factory,
3
                        UncaughtExceptionHandler handler,
4
                        boolean asyncMode,
5
                        int corePoolSize,
6
                        int maximumPoolSize,
7
                        int minimumRunnable,
8
                        Predicate<? super ForkJoinPool> saturate,
9
                        long keepAliveTime,
10
                        TimeUnit unit) {
11
        checkPermission();
12
        int p = parallelism;
13
        if (p <= 0 || p > MAX_CAP || p > maximumPoolSize || keepAliveTime <= 0L)
14
            throw new IllegalArgumentException();
15
        if (factory == null || unit == null)
16
            throw new NullPointerException();
17
        this.parallelism = p;
18
        this.factory = factory;
19
        this.ueh = handler;
20
        this.saturate = saturate;
21
        this.config = asyncMode ? FIFO : 0;
22
        this.keepAlive = Math.max(unit.toMillis(keepAliveTime), TIMEOUT_SLOP);
23
        int corep = Math.clamp(corePoolSize, p, MAX_CAP);
24
        int maxSpares = Math.clamp(maximumPoolSize - p, 0, MAX_CAP);
25
        int minAvail = Math.clamp(minimumRunnable, 0, MAX_CAP);
26
        this.bounds = (long)(minAvail & SMASK) | (long)(maxSpares << SWIDTH) |
27
            ((long)corep << 32);
28
        int size = 1 << (33 - Integer.numberOfLeadingZeros(p - 1));
29
        this.registrationLock = new ReentrantLock();
30
        this.queues = new WorkQueue[size];
31
        String pid = Integer.toString(getAndAddPoolIds(1) + 1);
32
        String name = "ForkJoinPool-" + pid;
33
        this.workerNamePrefix = name + "-worker-";
34
        this.container = SharedThreadContainer.create(name);
35
    }