https://github.com/nivance/Java7Demos/blob/master/forkjoin/MergeSort.java
http://www.programmingopiethehokie.com/2014/07/java-7-forkjoin-and-java-8-streams.html
The fork/join framework provides an easy way to parallelize work that can be broken up into smaller parts. Work is distributed to threads in a pool, and idle threads can steal work from busy threads (work stealing).
Task != Thread, Work stealing
Even though tasks behave a lot like threads, we do not want to make then equivalent to threads. Tasks are much more "lightweight" than threads, and in general, a fork/join program should be reasonably efficient even if a large number of tasks are created. For this reason, allocating one thread per task is not a good strategy, because the overhead of creating, running, and terminating threads could create overhead which reduces the possible parallel speedup.
In practice, fork/join frameworks use a thread pool in which a fixed number of threads are created. Each thread has a queue of tasks that are awaiting a chance to execute. When a task is started (forked), it is added to the queue of the thread that is executing its parent task.
Because each thread can be executing only one task at a time, each thread's task queues can accumulate tasks which are not currently executing. Threads that have no tasks allocated to them will attempt to steal a task from a thread whose queue has at least one task - this is called work stealing. By this mechanism, tasks are distributed to all of the threads in the thread pool.
By using a thread pool with work stealing, a fork/join framework can allow a relatively fine-grained division of the problem, but only create the minimum number of threads needed to fully exploit the available CPU cores. Typically, the thread pool will have one thread per available CPU core.
http://cecs.wright.edu/~pmateti/Courses/7900/Lectures/jdk1.8.0_20-samples/sample/forkjoin/mergesort/MergeSort.java
http://www.javacodegeeks.com/2011/02/java-forkjoin-parallel-programming.html
http://ricardozuasti.com/2012/java-concurrency-examples-forkjoin-framework/
http://highperformanceinjava.blogspot.com/2013/01/forkjoin-for-parallel-sorting-and-more.html
protected void compute() {
if (begin < end) {
int pivot = OurQuickSort.partition(array, begin, end);
if (end - begin < THRESHOLD) {
OurQuickSort.quicksort(array, begin, pivot - 1);
OurQuickSort.quicksort(array, pivot + 1, end);
} else {
invokeAll(
new SortAction(array, begin, pivot - 1),
new SortAction(array, pivot + 1, end));
}
}
}
http://java.dzone.com/articles/javas-fork-join-framework
http://www.java2s.com/Open-Source/Java_Free_Code/Framework/Download_parallel_sort_Free_Java_Code.htm
Implementation of merge sort and quicksort using the Fork/Join framework (RecursiveAction) of Java 7.
http://www.programmingopiethehokie.com/2014/07/java-7-forkjoin-and-java-8-streams.html
The fork/join framework provides an easy way to parallelize work that can be broken up into smaller parts. Work is distributed to threads in a pool, and idle threads can steal work from busy threads (work stealing).
Fork/join parallelism is a style of parallel programming useful for exploiting the parallelism inherent in divide and conquer algorithms on shared memory multiprocessors.
The idea is quite simple: a larger task can be divided into smaller tasks whose solutions can then be combined. As long as the smaller tasks are independent, they can be executed in parallel.
Example: finding the maximum value in a large array of integers.
Example: finding the maximum value in a large array of integers. Each task is given three data:
- array to search
- start and end indices to search
The computation starts with the creation of a "root task", which represents the overall problem.
rootTask = new Task(arr, 0, arr.length)
Assume that task class has a compute method which searches the specified range. The compute method will check to see how many elements are in the range it needs to search. If the number of elements in the range is below a threshold value, then a sequential computation is performed. Otherwise, the range is split into two sub-ranges containing half of the elements. Subtasks are created to search the sub-ranges. The maximum of the maximum values in the subranges is the maximum value in the overall range.
Task != Thread, Work stealing
Even though tasks behave a lot like threads, we do not want to make then equivalent to threads. Tasks are much more "lightweight" than threads, and in general, a fork/join program should be reasonably efficient even if a large number of tasks are created. For this reason, allocating one thread per task is not a good strategy, because the overhead of creating, running, and terminating threads could create overhead which reduces the possible parallel speedup.
In practice, fork/join frameworks use a thread pool in which a fixed number of threads are created. Each thread has a queue of tasks that are awaiting a chance to execute. When a task is started (forked), it is added to the queue of the thread that is executing its parent task.
Because each thread can be executing only one task at a time, each thread's task queues can accumulate tasks which are not currently executing. Threads that have no tasks allocated to them will attempt to steal a task from a thread whose queue has at least one task - this is called work stealing. By this mechanism, tasks are distributed to all of the threads in the thread pool.
By using a thread pool with work stealing, a fork/join framework can allow a relatively fine-grained division of the problem, but only create the minimum number of threads needed to fully exploit the available CPU cores. Typically, the thread pool will have one thread per available CPU core.
class FindMaxTask extends RecursiveTask<Integer> {
private int[] arr;
private int start, end;
public FindMaxTask(int[] arr, int start, int end) {
this.arr = arr;
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
if (end - start <= THRESHOLD) {
// number of elements is at or below the threshold - compute directly
return computeDirectly();
} else {
// number of elements is above the threshold - split into subtasks
int mid = start + (end-start) / 2;
FindMaxTask left = new FindMaxTask(arr, start, mid);
FindMaxTask right = new FindMaxTask(arr, mid, end);
// invoke the tasks in parallel and wait for them to complete
invokeAll(left, right);
// maximum of overall range is maximum of sub-ranges
return Math.max(left.join(), right.join());
}
}
private Integer computeDirectly() {
int max = Integer.MIN_VALUE;
for (int i = start; i < end; i++) {
if (arr[i] > max) {
max = arr[i];
}
}
return max;
}
}
Note that the invokeAll method may be called to execute two or more tasks in parallel (waiting for them to complete). The join method waits for a task to complete (if it has not completed yet) and returns the result of its compute method.
To do the overall computation on an entire array, we just need to create a root task and execute it using a ForkJoinPool object:
int[] data = some array of integer values ForkJoinPool pool = new ForkJoinPool(); FindMaxTask rootTask = new FindMaxTask(data, 0, data.length); Integer result = pool.invoke(rootTask);
int threshold = data.length / Runtime.getRuntime().availableProcessors(); MergeSortTask rootTask = new MergeSortTask(data, 0, data.length, threshold);
http://cecs.wright.edu/~pmateti/Courses/7900/Lectures/jdk1.8.0_20-samples/sample/forkjoin/mergesort/MergeSort.java
public class MergeSort {
private final ForkJoinPool pool;
private static class MergeSortTask extends RecursiveAction {
private final int[] array;
private final int low;
private final int high;
private static final int THRESHOLD = 8;
/**
* Creates a {@code MergeSortTask} containing the array and the bounds of the array
*
* @param array the array to sort
* @param low the lower element to start sorting at
* @param high the non-inclusive high element to sort to
*/
protected MergeSortTask(int[] array, int low, int high) {
this.array = array;
this.low = low;
this.high = high;
}
@Override
protected void compute() {
if (high - low <= THRESHOLD) {
Arrays.sort(array, low, high);
} else {
int middle = low + ((high - low) >> 1);
// Execute the sub tasks and wait for them to finish
invokeAll(new MergeSortTask(array, low, middle), new MergeSortTask(array, middle, high));
// Then merge the results
merge(middle);
}
}
/**
* Merges the two sorted arrays this.low, middle - 1 and middle, this.high - 1
* @param middle the index in the array where the second sorted list begins
*/
private void merge(int middle) {
if (array[middle - 1] < array[middle]) {
return; // the arrays are already correctly sorted, so we can skip the merge
}
int[] copy = new int[high - low];
System.arraycopy(array, low, copy, 0, copy.length);
int copyLow = 0;
int copyHigh = high - low;
int copyMiddle = middle - low;
for (int i = low, p = copyLow, q = copyMiddle; i < high; i++) {
if (q >= copyHigh || (p < copyMiddle && copy[p] < copy[q]) ) {
array[i] = copy[p++];
} else {
array[i] = copy[q++];
}
}
}
}
/**
* Creates a {@code MergeSort} containing a ForkJoinPool with the indicated parallelism level
* @param parallelism the parallelism level used
*/
public MergeSort(int parallelism) {
pool = new ForkJoinPool(parallelism);
}
/**
* Sorts all the elements of the given array using the ForkJoin framework
* @param array the array to sort
*/
public void sort(int[] array) {
ForkJoinTask<Void> job = pool.submit(new MergeSortTask(array, 0, array.length));
job.join();
}
}
http://www.javacodegeeks.com/2011/02/java-forkjoin-parallel-programming.html
03 | public class FibonacciProblem { |
04 | |
05 | public int n; |
06 |
07 | public FibonacciProblem(int n) { |
08 | this.n = n; |
09 | } |
10 | |
11 | public long solve() { |
12 | return fibonacci(n); |
13 | } |
14 | |
15 | private long fibonacci(int n) { |
16 | System.out.println("Thread: " + |
17 | Thread.currentThread().getName() + " calculates " + n); |
18 | if (n <= 1) |
19 | return n; |
20 | else |
21 | return fibonacci(n-1) + fibonacci(n-2); |
22 | } |
23 |
24 | } |
public class FibonacciTask extends RecursiveTask<Long> { |
06 |
07 | private static final long serialVersionUID = 6136927121059165206L; |
08 | |
09 | private static final int THRESHOLD = 5; |
10 |
11 | private FibonacciProblem problem; |
12 | public long result; |
13 | |
14 | public FibonacciTask(FibonacciProblem problem) { |
15 | this.problem = problem; |
16 | } |
17 |
18 | @Override |
19 | public Long compute() { |
20 | if (problem.n < THRESHOLD) { // easy problem, don't bother with parallelism |
21 | result = problem.solve(); |
22 | } |
23 | else { |
24 | FibonacciTask worker1 = new FibonacciTask(new FibonacciProblem(problem.n-1)); |
25 | FibonacciTask worker2 = new FibonacciTask(new FibonacciProblem(problem.n-2)); |
26 | worker1.fork(); |
27 | result = worker2.compute() + worker1.join(); |
28 |
29 | } |
30 | return result; |
31 | } |
32 |
33 | } |
03 | public class FibonacciProblem { |
04 | |
05 | public int n; |
06 |
07 | public FibonacciProblem(int n) { |
08 | this.n = n; |
09 | } |
10 | |
11 | public long solve() { |
12 | return fibonacci(n); |
13 | } |
14 | |
15 | private long fibonacci(int n) { |
16 | System.out.println("Thread: " + |
17 | Thread.currentThread().getName() + " calculates " + n); |
18 | if (n <= 1) |
19 | return n; |
20 | else |
21 | return fibonacci(n-1) + fibonacci(n-2); |
22 | } |
23 |
24 | } |
http://ricardozuasti.com/2012/java-concurrency-examples-forkjoin-framework/
public class QuickSort<T extends Comparable> extends RecursiveAction { private List<T> data; private int left; private int right; public QuickSort(List<T> data){ this.data=data; this.left = 0; this.right = data.size() - 1; } public QuickSort(List<T> data, int left, int right){ this.data = data; this.left = left; this.right = right; } @Override protected void compute() { if (left < right){ int pivotIndex = left + ((right - left)/2); pivotIndex = partition(pivotIndex); invokeAll(new QuickSort(data, left, pivotIndex-1), new QuickSort(data, pivotIndex+1, right)); } } private int partition(int pivotIndex){ T pivotValue = data.get(pivotIndex); swap(pivotIndex, right); int storeIndex = left; for (int i=left; i<right; i++){ if (data.get(i).compareTo(pivotValue) < 0){ swap(i, storeIndex); storeIndex++; } } swap(storeIndex, right); return storeIndex; } private void swap(int i, int j){ if (i != j){ T iValue = data.get(i); data.set(i, data.get(j)); data.set(j, iValue); } }}protected void compute() {
if (begin < end) {
int pivot = OurQuickSort.partition(array, begin, end);
if (end - begin < THRESHOLD) {
OurQuickSort.quicksort(array, begin, pivot - 1);
OurQuickSort.quicksort(array, pivot + 1, end);
} else {
invokeAll(
new SortAction(array, begin, pivot - 1),
new SortAction(array, pivot + 1, end));
}
}
}
http://java.dzone.com/articles/javas-fork-join-framework
http://www.java2s.com/Open-Source/Java_Free_Code/Framework/Download_parallel_sort_Free_Java_Code.htm
Implementation of merge sort and quicksort using the Fork/Join framework (RecursiveAction) of Java 7.
