What happens when you try this...
class abst extends T {
public int add(int a, int b) {
return a-b;
}
// What happens when you add below method?
// a.add() will cause compilation error because add() is being called
// using T type variable from class abst & add() is private to class T
// so it is not visible from class abst.
public void action() {
T a = new abst();
System.out.println(a.add(10, 20));
}
}
abstract class T {
private int add(int a, int b) {
return a+b;
}
// What will be the output of the following
// Here observe the access specifier of add(), it is private to class T
// & below it is being called from class T itself. add() is private & so it will not
// be overridden by class abst & while below call scope of add() of class T is closer than
// that of add() from class abst. So output will be 30
// But if we change private to public then it will be overridden by class abst & add()
// of abst will be called in below call & out put will be -10
public static void main(String[] args) {
T a = new abst();
System.out.println(a.add(10, 20));
}
}
===============================================================================================
Fork & Join
I was just revisiting the Fork-Join examples & checked below links but was not satisfied with the kind implementation was given on these for the summation
http://tutorials.jenkov.com/java-util-concurrent/java-fork-and-join-forkjoinpool.html
https://www.mkyong.com/java/java-fork-join-framework-examples/
And I have to say that the examples of addition of the numbers in the array doesn't seem to be the right one in case of threading examples, as naïve people then start applying threading in every task, which is really a bad habit.
If you write a program to add the numbers in an array using the Streams or Fork join then you will find both are slower than the normal addition using for loop. Wonder why?
'Addition' task doesn't take much time but creating the new threads, context switching, creating streams & then iterating through that stream take much time.
In such examples, performance of Fork-join & stream are almost the same but normal addition is quite faster.
So before going for any threading concept, one must weigh the cost of threading & maintenance of such code.
But still at below link I found better implementation for Fork-Join example -
https://www.tutorialspoint.com/java_concurrency/concurrency_fork_join.htm#
And if still you are planning to use fork-join then will suggest to read the below article -
http://coopsoft.com/ar/CalamityArticle.html
So multithreading is good to use when you can divide your work into different independent tasks which can be done in parallel & each task takes time to complete. So rather doing such tasks sequentially, it is good to complete such tasks in parallel to save the considerable time. I updated the code shown to include the wait during addition, plus added few lines to show the performance of the same task done sequentially or using the streams or using Fork-Join. And added one block where I am using Executor framework to have the pool of threads, & it outperforms others. Just my thought, Stream shouldn't be used in general to use Java8 or look cool to use Java8. Streams should be used for very specific purposes only otherwise these impact the performance, Output of the below code is shown at the end which shows Fork-Join & Executor results, as results for Stream & Sequential addition will take enough time but if you want then can execute the below code with the different values for length, threshold & taskWait to observe the real big differences.
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;
public class ForkJoinAdd extends RecursiveTask<Long> {
private long[] numbers;
private final int start;
private final int end;
public static final int threshold = 1000;
private static int length = 1000;
private static long sum = 0;
private static long result = 0;
private static int taskWait = 1;
public ForkJoinAdd() {
start = 0;
end = 0;
numbers = new long[1];
}
public ForkJoinAdd(long[] numbers) {
this.numbers = numbers;
this.start = 0;
this.end = numbers.length;
}
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
System.out.println("Sum : " + ForkJoinAdd.startForkJoinSum(length));
System.out.println("Time Taken with Fork-Join : "+ (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
int[] arr = new int[length];
for(int i = 1; i <= length; i++) {
arr[i-1] = i;
}
ForkJoinAdd fja = new ForkJoinAdd();
Arrays.stream(arr).forEach(a -> {
try {
fja.addition(a);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("Sum : " + sum);
System.out.println("Time Taken using Stream : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis(); sum = 0; for(int i = 0; i < length; i++)
{
try {
fja.addition(arr[i]);
} catch (InterruptedException e) {
e.printStackTrace();
}
} System.out.println("Sum : " + sum);
System.out.println("Time Taken using sequential addition : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
ExecutorService es = Executors.newCachedThreadPool(); // cachedPool is giving better performance here than using
// newFixedThreadPool(80); It is difficult to decide on nthe right number of threads to have in the pool as
// extra number of threads will impact the performance & less number will also impact the performance, so let
// Java decide here if it needs to create new threads & how many.
int starts = 0;
Collection<Summation> callabels = new ArrayList<>(length/threshold+1);
int i = 0;
int[] arrT;
for(; i < length; i++) {
if(i - starts == threshold-1) {
arrT = Arrays.copyOfRange(arr, starts, i);
callabels.add(new Summation(arrT));
starts = i;
}
}
if(i - starts < threshold-1) {
callabels.add(new Summation(Arrays.copyOfRange(arr, starts, i+1)));
}
try {
List<Future<Long>> results = es.invokeAll(callabels);
es.awaitTermination(2, TimeUnit.MILLISECONDS);
es.shutdown();
while(!es.isTerminated());
results.forEach(a -> {try {
result+=a.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}});
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(result);
System.out.println("Time Taken using Executor: " + (System.currentTimeMillis() - start)+"ms");
}
@Override
protected Long compute() {
int length = end - start;
if (numbers.length <= threshold) {
try {
return add();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long[] numbersT = Arrays.copyOfRange(numbers, 0, threshold);
ForkJoinAdd firstTask = new ForkJoinAdd(numbersT);
firstTask.fork(); //start asynchronously
numbers = Arrays.copyOfRange(numbers, threshold, length);
ForkJoinAdd secondTask = new ForkJoinAdd(numbers);
Long secondTaskResult = secondTask.compute();
Long firstTaskResult = firstTask.join();
return firstTaskResult + secondTaskResult;
}
private long add() throws InterruptedException {
long result = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for (int i = start; i < end; i++) {
result += numbers[i];
}
return result;
}
private void addition(int a) throws InterruptedException {
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
sum+=a;
}
public static long startForkJoinSum(long n) {
long[] numbers = LongStream.rangeClosed(1, n).toArray();
ForkJoinTask<Long> task = new ForkJoinAdd(numbers);
return new ForkJoinPool().invoke(task);
}
static class Summation implements Callable<Long> {
private int[] arr;
public Summation(int[] arr) {
this.arr = arr;
}
@Override
public Long call() throws InterruptedException {
long sum = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for(int i : arr) {
sum += i;
}
return sum;
}
}
}
Output
----------
Sum : 500000000500000000
Time Taken with Fork-Join : 268458ms
500000000500000000
Time Taken using Executor: 9981ms
Conclusion :- If possible then use the Executor framework yourself to have the thread pool rather using the Fork-Join. I am still not sure about the benefits of Fork-Join concept, if work stealing concept doesn't improve the performance.
===============================================================================================
Let's continue the above analysis. As I am not getting any money for this, so not analysing the code of Fork-Join classes in JDK regarding what is wrong there. But I am just evaluating it as per my understanding & as shown above I also don't recommend it. And below is the sample with the same code as above, instead I am just using the RecursiveAction now & I observed that if I don't put the sleep() between 2 fork() calls then results are inconsistent & incorrect. I am not sure if I have done something wrong in the code but I was not expecting such behaviour. Please let me know if you find any issue in the below code to resolve this.
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;
public class ForkJoinAdd extends RecursiveAction {
private long[] numbers;
private final int start;
private final int end;
public static final int threshold = 100;
private static int length = 1000;
private static long sum = 0;
private static long streamSum = 0;
private static long result = 0;
private static int taskWait = 1;
public ForkJoinAdd() {
start = 0;
end = 0;
numbers = new long[1];
}
public ForkJoinAdd(long[] numbers) {
this.numbers = numbers;
this.start = 0;
this.end = numbers.length;
}
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
ForkJoinAdd.startForkJoinSum(length);
System.out.println("Sum : " + sum);
System.out.println("Time Taken with Fork-Join : "+ (System.currentTimeMillis() - start)+"ms");
Thread.sleep(10);
start = System.currentTimeMillis();
int[] arr = new int[length];
for(int i = 1; i <= length; i++) {
arr[i-1] = i;
}
ForkJoinAdd fja = new ForkJoinAdd();
// sum = 0;
Arrays.stream(arr).forEach(a -> {
try {
fja.addition(a);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("Sum : " + streamSum);
System.out.println("Time Taken using Stream : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
streamSum = 0;
for(int i = 0; i < length; i++) {
try {
fja.addition(arr[i]);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("Sum : " + streamSum);
System.out.println("Time Taken using sequential addition : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
ExecutorService es = Executors.newCachedThreadPool(); // cachedPool is giving better performance here than using
// newFixedThreadPool(80); It is difficult to decide on the right number of threads to have in the pool as
// extra number of threads will impact the performance & less number will also impact the performance, so let
// Java decide here if it needs to create new threads & how many.
int starts = 0;
Collection<Summation> callabels = new ArrayList<>(length/threshold+1);
int i = 0;
int[] arrT;
for(; i < length; i++) {
if(i - starts == threshold-1) {
arrT = Arrays.copyOfRange(arr, starts, i);
callabels.add(new Summation(arrT));
starts = i;
}
}
if(i - starts < threshold-1) {
callabels.add(new Summation(Arrays.copyOfRange(arr, starts, i+1)));
}
try {
List<Future<Long>> results = es.invokeAll(callabels);
es.awaitTermination(2, TimeUnit.MILLISECONDS);
es.shutdown();
while(!es.isTerminated());
results.forEach(a -> {try {
result+=a.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}});
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Sum : " + result);
System.out.println("Time Taken using Executor: " + (System.currentTimeMillis() - start)+"ms");
}
@Override
protected void compute() {
if (numbers.length <= threshold) {
try {
sum+=add();
return;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long[] numbersT = Arrays.copyOfRange(numbers, 0, threshold);
ForkJoinAdd firstTask = new ForkJoinAdd(numbersT);
numbers = Arrays.copyOfRange(numbers, threshold, numbers.length);
ForkJoinAdd secondTask = new ForkJoinAdd(numbers);
firstTask.fork(); //start asynchronously
try {
// If we remove or reduce the sleep time then results will be inconsistent & incorrect.
Thread.sleep(2l);
} catch (InterruptedException e) {
e.printStackTrace();
}
secondTask.fork();
firstTask.join();
secondTask.join();
// System.out.println(firstTask.isDone());
// System.out.println(secondTask.isDone());
// return firstTaskResult + secondTaskResult;
}
private long add() throws InterruptedException {
long result = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for (int i = start; i < end; i++) {
result += numbers[i];
}
return result;
}
private void addition(int a) throws InterruptedException {
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
streamSum+=a;
}
public static void startForkJoinSum(long n) throws InterruptedException {
long[] numbers = LongStream.rangeClosed(1, n).toArray();
ForkJoinAdd task = new ForkJoinAdd(numbers);
ForkJoinPool fjp = new ForkJoinPool();
fjp.invoke(task);
// Below statement seems to close the pool even before
// all the tasks are submitted, causing wrong results.
// fjp.shutdown();
// One can tweak the wait time below before the pool is shut down
// & it doesn't accept the pending tasks causing wrong results.
fjp.awaitTermination(0, TimeUnit.MILLISECONDS);
}
static class Summation implements Callable<Long> {
private int[] arr;
public Summation(int[] arr) {
this.arr = arr;
}
@Override
public Long call() throws InterruptedException {
long sum = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for(int i : arr) {
sum += i;
}
return sum;
}
}
}
Output
======
Sum : 500500
Time Taken with Fork-Join : 86ms
Sum : 500500
Time Taken using Stream : 1258ms
Sum : 500500
Time Taken using sequential addition : 1260ms
Sum : 500500
Time Taken using Executor: 13ms
Conclusion : Avoid use of Fork-Join blindly or if you have alternative.
===============================================================================================
public int add(int a, int b) {
return a-b;
}
// What happens when you add below method?
// a.add() will cause compilation error because add() is being called
// using T type variable from class abst & add() is private to class T
// so it is not visible from class abst.
public void action() {
T a = new abst();
System.out.println(a.add(10, 20));
}
}
abstract class T {
private int add(int a, int b) {
return a+b;
}
// What will be the output of the following
// Here observe the access specifier of add(), it is private to class T
// & below it is being called from class T itself. add() is private & so it will not
// be overridden by class abst & while below call scope of add() of class T is closer than
// that of add() from class abst. So output will be 30
// But if we change private to public then it will be overridden by class abst & add()
// of abst will be called in below call & out put will be -10
public static void main(String[] args) {
T a = new abst();
System.out.println(a.add(10, 20));
}
}
===============================================================================================
Fork & Join
I was just revisiting the Fork-Join examples & checked below links but was not satisfied with the kind implementation was given on these for the summation
http://tutorials.jenkov.com/java-util-concurrent/java-fork-and-join-forkjoinpool.html
https://www.mkyong.com/java/java-fork-join-framework-examples/
And I have to say that the examples of addition of the numbers in the array doesn't seem to be the right one in case of threading examples, as naïve people then start applying threading in every task, which is really a bad habit.
If you write a program to add the numbers in an array using the Streams or Fork join then you will find both are slower than the normal addition using for loop. Wonder why?
'Addition' task doesn't take much time but creating the new threads, context switching, creating streams & then iterating through that stream take much time.
In such examples, performance of Fork-join & stream are almost the same but normal addition is quite faster.
So before going for any threading concept, one must weigh the cost of threading & maintenance of such code.
But still at below link I found better implementation for Fork-Join example -
https://www.tutorialspoint.com/java_concurrency/concurrency_fork_join.htm#
And if still you are planning to use fork-join then will suggest to read the below article -
http://coopsoft.com/ar/CalamityArticle.html
So multithreading is good to use when you can divide your work into different independent tasks which can be done in parallel & each task takes time to complete. So rather doing such tasks sequentially, it is good to complete such tasks in parallel to save the considerable time. I updated the code shown to include the wait during addition, plus added few lines to show the performance of the same task done sequentially or using the streams or using Fork-Join. And added one block where I am using Executor framework to have the pool of threads, & it outperforms others. Just my thought, Stream shouldn't be used in general to use Java8 or look cool to use Java8. Streams should be used for very specific purposes only otherwise these impact the performance, Output of the below code is shown at the end which shows Fork-Join & Executor results, as results for Stream & Sequential addition will take enough time but if you want then can execute the below code with the different values for length, threshold & taskWait to observe the real big differences.
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;
public class ForkJoinAdd extends RecursiveTask<Long> {
private long[] numbers;
private final int start;
private final int end;
public static final int threshold = 1000;
private static int length = 1000;
private static long sum = 0;
private static long result = 0;
private static int taskWait = 1;
public ForkJoinAdd() {
start = 0;
end = 0;
numbers = new long[1];
}
public ForkJoinAdd(long[] numbers) {
this.numbers = numbers;
this.start = 0;
this.end = numbers.length;
}
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
System.out.println("Sum : " + ForkJoinAdd.startForkJoinSum(length));
System.out.println("Time Taken with Fork-Join : "+ (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
int[] arr = new int[length];
for(int i = 1; i <= length; i++) {
arr[i-1] = i;
}
ForkJoinAdd fja = new ForkJoinAdd();
Arrays.stream(arr).forEach(a -> {
try {
fja.addition(a);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("Sum : " + sum);
System.out.println("Time Taken using Stream : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis(); sum = 0; for(int i = 0; i < length; i++)
{
try {
fja.addition(arr[i]);
} catch (InterruptedException e) {
e.printStackTrace();
}
} System.out.println("Sum : " + sum);
System.out.println("Time Taken using sequential addition : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
ExecutorService es = Executors.newCachedThreadPool(); // cachedPool is giving better performance here than using
// newFixedThreadPool(80); It is difficult to decide on nthe right number of threads to have in the pool as
// extra number of threads will impact the performance & less number will also impact the performance, so let
// Java decide here if it needs to create new threads & how many.
int starts = 0;
Collection<Summation> callabels = new ArrayList<>(length/threshold+1);
int i = 0;
int[] arrT;
for(; i < length; i++) {
if(i - starts == threshold-1) {
arrT = Arrays.copyOfRange(arr, starts, i);
callabels.add(new Summation(arrT));
starts = i;
}
}
if(i - starts < threshold-1) {
callabels.add(new Summation(Arrays.copyOfRange(arr, starts, i+1)));
}
try {
List<Future<Long>> results = es.invokeAll(callabels);
es.awaitTermination(2, TimeUnit.MILLISECONDS);
es.shutdown();
while(!es.isTerminated());
results.forEach(a -> {try {
result+=a.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}});
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(result);
System.out.println("Time Taken using Executor: " + (System.currentTimeMillis() - start)+"ms");
}
@Override
protected Long compute() {
int length = end - start;
if (numbers.length <= threshold) {
try {
return add();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long[] numbersT = Arrays.copyOfRange(numbers, 0, threshold);
ForkJoinAdd firstTask = new ForkJoinAdd(numbersT);
firstTask.fork(); //start asynchronously
numbers = Arrays.copyOfRange(numbers, threshold, length);
ForkJoinAdd secondTask = new ForkJoinAdd(numbers);
Long secondTaskResult = secondTask.compute();
Long firstTaskResult = firstTask.join();
return firstTaskResult + secondTaskResult;
}
private long add() throws InterruptedException {
long result = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for (int i = start; i < end; i++) {
result += numbers[i];
}
return result;
}
private void addition(int a) throws InterruptedException {
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
sum+=a;
}
public static long startForkJoinSum(long n) {
long[] numbers = LongStream.rangeClosed(1, n).toArray();
ForkJoinTask<Long> task = new ForkJoinAdd(numbers);
return new ForkJoinPool().invoke(task);
}
static class Summation implements Callable<Long> {
private int[] arr;
public Summation(int[] arr) {
this.arr = arr;
}
@Override
public Long call() throws InterruptedException {
long sum = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for(int i : arr) {
sum += i;
}
return sum;
}
}
}
Output
----------
Sum : 500000000500000000
Time Taken with Fork-Join : 268458ms
500000000500000000
Time Taken using Executor: 9981ms
Conclusion :- If possible then use the Executor framework yourself to have the thread pool rather using the Fork-Join. I am still not sure about the benefits of Fork-Join concept, if work stealing concept doesn't improve the performance.
===============================================================================================
Let's continue the above analysis. As I am not getting any money for this, so not analysing the code of Fork-Join classes in JDK regarding what is wrong there. But I am just evaluating it as per my understanding & as shown above I also don't recommend it. And below is the sample with the same code as above, instead I am just using the RecursiveAction now & I observed that if I don't put the sleep() between 2 fork() calls then results are inconsistent & incorrect. I am not sure if I have done something wrong in the code but I was not expecting such behaviour. Please let me know if you find any issue in the below code to resolve this.
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.TimeUnit;
import java.util.stream.LongStream;
public class ForkJoinAdd extends RecursiveAction {
private long[] numbers;
private final int start;
private final int end;
public static final int threshold = 100;
private static int length = 1000;
private static long sum = 0;
private static long streamSum = 0;
private static long result = 0;
private static int taskWait = 1;
public ForkJoinAdd() {
start = 0;
end = 0;
numbers = new long[1];
}
public ForkJoinAdd(long[] numbers) {
this.numbers = numbers;
this.start = 0;
this.end = numbers.length;
}
public static void main(String[] args) throws InterruptedException {
long start = System.currentTimeMillis();
ForkJoinAdd.startForkJoinSum(length);
System.out.println("Sum : " + sum);
System.out.println("Time Taken with Fork-Join : "+ (System.currentTimeMillis() - start)+"ms");
Thread.sleep(10);
start = System.currentTimeMillis();
int[] arr = new int[length];
for(int i = 1; i <= length; i++) {
arr[i-1] = i;
}
ForkJoinAdd fja = new ForkJoinAdd();
// sum = 0;
Arrays.stream(arr).forEach(a -> {
try {
fja.addition(a);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
System.out.println("Sum : " + streamSum);
System.out.println("Time Taken using Stream : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
streamSum = 0;
for(int i = 0; i < length; i++) {
try {
fja.addition(arr[i]);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("Sum : " + streamSum);
System.out.println("Time Taken using sequential addition : " + (System.currentTimeMillis() - start)+"ms");
start = System.currentTimeMillis();
ExecutorService es = Executors.newCachedThreadPool(); // cachedPool is giving better performance here than using
// newFixedThreadPool(80); It is difficult to decide on the right number of threads to have in the pool as
// extra number of threads will impact the performance & less number will also impact the performance, so let
// Java decide here if it needs to create new threads & how many.
int starts = 0;
Collection<Summation> callabels = new ArrayList<>(length/threshold+1);
int i = 0;
int[] arrT;
for(; i < length; i++) {
if(i - starts == threshold-1) {
arrT = Arrays.copyOfRange(arr, starts, i);
callabels.add(new Summation(arrT));
starts = i;
}
}
if(i - starts < threshold-1) {
callabels.add(new Summation(Arrays.copyOfRange(arr, starts, i+1)));
}
try {
List<Future<Long>> results = es.invokeAll(callabels);
es.awaitTermination(2, TimeUnit.MILLISECONDS);
es.shutdown();
while(!es.isTerminated());
results.forEach(a -> {try {
result+=a.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}});
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Sum : " + result);
System.out.println("Time Taken using Executor: " + (System.currentTimeMillis() - start)+"ms");
}
@Override
protected void compute() {
if (numbers.length <= threshold) {
try {
sum+=add();
return;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
long[] numbersT = Arrays.copyOfRange(numbers, 0, threshold);
ForkJoinAdd firstTask = new ForkJoinAdd(numbersT);
numbers = Arrays.copyOfRange(numbers, threshold, numbers.length);
ForkJoinAdd secondTask = new ForkJoinAdd(numbers);
firstTask.fork(); //start asynchronously
try {
// If we remove or reduce the sleep time then results will be inconsistent & incorrect.
Thread.sleep(2l);
} catch (InterruptedException e) {
e.printStackTrace();
}
secondTask.fork();
firstTask.join();
secondTask.join();
// System.out.println(firstTask.isDone());
// System.out.println(secondTask.isDone());
// return firstTaskResult + secondTaskResult;
}
private long add() throws InterruptedException {
long result = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for (int i = start; i < end; i++) {
result += numbers[i];
}
return result;
}
private void addition(int a) throws InterruptedException {
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
streamSum+=a;
}
public static void startForkJoinSum(long n) throws InterruptedException {
long[] numbers = LongStream.rangeClosed(1, n).toArray();
ForkJoinAdd task = new ForkJoinAdd(numbers);
ForkJoinPool fjp = new ForkJoinPool();
fjp.invoke(task);
// Below statement seems to close the pool even before
// all the tasks are submitted, causing wrong results.
// fjp.shutdown();
// One can tweak the wait time below before the pool is shut down
// & it doesn't accept the pending tasks causing wrong results.
fjp.awaitTermination(0, TimeUnit.MILLISECONDS);
}
static class Summation implements Callable<Long> {
private int[] arr;
public Summation(int[] arr) {
this.arr = arr;
}
@Override
public Long call() throws InterruptedException {
long sum = 0;
Thread.sleep(TimeUnit.MILLISECONDS.toMillis(taskWait));
for(int i : arr) {
sum += i;
}
return sum;
}
}
}
Output
======
Sum : 500500
Time Taken with Fork-Join : 86ms
Sum : 500500
Time Taken using Stream : 1258ms
Sum : 500500
Time Taken using sequential addition : 1260ms
Sum : 500500
Time Taken using Executor: 13ms
Conclusion : Avoid use of Fork-Join blindly or if you have alternative.
===============================================================================================
