{"id":43743,"date":"2015-09-16T19:34:06","date_gmt":"2015-09-16T16:34:06","guid":{"rendered":"http:\/\/www.javacodegeeks.com\/?p=43743"},"modified":"2023-12-13T11:51:57","modified_gmt":"2023-12-13T09:51:57","slug":"forkjoin-framework","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2015\/09\/forkjoin-framework.html","title":{"rendered":"Fork\/Join Framework"},"content":{"rendered":"

This article is part of our Academy Course titled Java Concurrency Essentials<\/a>.<\/p>\n

In this course, you will dive into the magic of concurrency. You will be introduced to the fundamentals of concurrency and concurrent code and you will learn about concepts like atomicity, synchronization and thread safety. Check it out here<\/a>!<\/em><\/p>\n

\n

Table Of Contents<\/h4>\n
\n
1. Introduction<\/a><\/dt>\n
2. Fork\/Join<\/a><\/dt>\n
\n
\n
2.1. RecursiveTask<\/a><\/dt>\n
2.2. RecursiveAction<\/a><\/dt>\n
2.3. ForkJoinPool and ExecutorService<\/a><\/dt>\n<\/dl>\n<\/dd>\n<\/dl>\n<\/div>\n

<\/a>1. Introduction<\/h2>\n

This article gives an introduction into the Fork\/Join Framework that is part of the JDK since version 1.7. It describes the basic features of the frameworks and provides some examples in order to provide some practical experience.<\/p>\n

<\/a>2. Fork\/Join<\/h2>\n

The base class of the Fork\/Join Framework is java.util.concurrent.ForkJoinPool<\/code>. This class implements the two interfaces Executor<\/code> and ExecutorService<\/code> and subclasses the AbstractExecutorService<\/code>. Hence the ForkJoinPool<\/code> is basically a thread pool that takes special kinds of tasks, namely the ForkJoinTask<\/code>. This class implements the already known interface Future<\/code> and therewith methods like get()<\/code>, cancel()<\/code> and isDone()<\/code>. Beyond that this class also provides two methods that gave the whole framework its name: fork()<\/code> and join()<\/code>.<\/p>\n

While a call of fork()<\/code> will start an asynchronous execution of the task, a call of join()<\/code> will wait until the task has finished and retrieve its result. Hence we can split a given task into multiple smaller tasks, fork each task and finally wait for all tasks to finish. This makes the implementation of complex problems easier.<\/p>\n

In computer science this approach is also known as divide-and-conquer approach. Every time a problem is too complex to solve at once, it is divided into multiple smaller and easier to solve problems. This can be written in pseudo code like that:<\/p>\n

\nif(problem.getSize() > THRESHOLD) {\n\tSmallerProblem smallerProblem1 = new SmallerProblem();\n\tsmallerProblem1.fork();\n\tSmallerProblem smallerProblem2 = new SmallerProblem();\n\tsmallerProblem2.fork();\n\treturn problem.solve(smallerProblem1.join(), smallerProblem2.join());\n} else {\n\treturn problem.solve();\n}\n<\/pre>\n
First we check if the current size of the problem is bigger than a given threshold. If this is the case, we divide the problem into smaller problems, fork()<\/code> each new task and then wait for the results by calling join()<\/code>. As join()<\/code> returns the results for each subtask, we have to find the best solution of the smaller problems and return this as our best solution. These steps are repeated until the given threshold is too low and the problem so small that we can compute its solution directly without further division.<\/p>\n
<\/a>2.1. RecursiveTask<\/h3>\n
To grasp this procedure a little bit better, we implement an algorithm that finds the smallest number within an array of integer values. This problem is not one you would solve in your day to day work using a ForkJoinPool<\/code>, but the following implementation shows the basic principles very clearly. In the main()<\/code> method we setup an integer array with random values and create a new ForkJoinPool<\/code>.<\/p>\n
The first parameter passed to its constructor is an indicator for the level of desired parallelism. Here we query the Runtime<\/code> for the number of available CPU cores. Then we call the invoke()<\/code> method and pass an instance of FindMin<\/code>. FindMin<\/code> extends the class RecursiveTask<\/code>, which itself is a subclass of the ForkJoinTask<\/code> mentioned before. The class ForkJoinTask<\/code> has actually two subclasses: one is designed for tasks that return a value (RecursiveTask<\/code>) and one that is designed for tasks without return value (RecursiveAction<\/code>). The superclass forces us to implement compute()<\/code>. Here we take a look at the given slice of the integer array and decide whether the current problem is too big to be solved immediately or not.<\/p>\n
When finding the smallest number within an array, the smallest problem size to be solved directly is to compare two elements with each other and return the smallest value of these. If we have currently more than two elements, we divide the array into two parts and find again the smallest number within these two parts. This is done by creating two new instances of FindMin<\/code>.
 
<\/div> <\/div><\/p>\n
The constructor is fed with the array and the start and end index. Then we start the execution of these two task asynchronously by calling fork()<\/code>. This call submits the two tasks in the queue of the thread pool. The thread pool implements a strategy called work-stealing, i.e. if all other threads have enough to do, the current threads steals its work from one of the other tasks. This makes sure that tasks get executed as fast as possible.<\/p>\n
\npublic class FindMin extends RecursiveTask<Integer> {\n\tprivate static final long serialVersionUID = 1L;\n\tprivate int[] numbers;\n\tprivate int startIndex;\n\tprivate int endIndex;\n\n\tpublic FindMin(int[] numbers, int startIndex, int endIndex) {\n\t\tthis.numbers = numbers;\n\t\tthis.startIndex = startIndex;\n\t\tthis.endIndex = endIndex;\n\t}\n\n\t@Override\n\tprotected Integer compute() {\n\t\tint sliceLength = (endIndex - startIndex) + 1;\n\t\tif (sliceLength > 2) {\n\t\t\tFindMin lowerFindMin = new FindMin(numbers, startIndex, startIndex + (sliceLength \/ 2) - 1);\n\t\t\tlowerFindMin.fork();\n\t\t\tFindMin upperFindMin = new FindMin(numbers, startIndex + (sliceLength \/ 2), endIndex);\n\t\t\tupperFindMin.fork();\n\t\t\treturn Math.min(lowerFindMin.join(), upperFindMin.join());\n\t\t} else {\n\t\t\treturn Math.min(numbers[startIndex], numbers[endIndex]);\n\t\t}\n\t}\n\n\tpublic static void main(String[] args) {\n\t\tint[] numbers = new int[100];\n\t\tRandom random = new Random(System.currentTimeMillis());\n\t\tfor (int i = 0; i < numbers.length; i++) {\n\t\t\tnumbers[i] = random.nextInt(100);\n\t\t}\n\t\tForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());\n\t\tInteger min = pool.invoke(new FindMin(numbers, 0, numbers.length - 1));\n\t\tSystem.out.println(min);\n\t}\n}\n<\/pre>\n
<\/a>2.2. RecursiveAction<\/h3>\n
As mentioned above next to RecursiveTask<\/code> we also have the class RecursiveAction<\/code>. In contrast to RecursiveTask<\/code> it does not have to return a value, hence it can be used for asynchronous computations that can be directly performed on a given data structure. Such an example is the computation of a grayscale image out of a colored image. All we have to do is to iterate over each pixel of the image and compute the grayscale value out of the RGB value using the following formula:<\/p>\n
gray = 0.2126 * red + 0.7152 * green + 0.0722 * blue<\/pre>\n
The floating point numbers represent how much the specific color contributes to our human perception of gray. As the highest value is used for green, we can conclude that a grayscale image is computed to nearly 3\/4 just of the green part. So the basic implementation would look like this, assuming that image is our object representing the actual pixel data and the methods setRGB()<\/code> and getRGB()<\/code> are used to retrieve the actual RGB value:<\/p>\n
\nfor (int row = 0; row < height; row++) {\n\tfor (int column = 0; column < bufferedImage.getWidth(); column++) {\n\t\tint grayscale = computeGrayscale(image.getRGB(column, row));\n\t\timage.setRGB(column, row, grayscale);\n\t}\n}\n<\/pre>\n
The above implementation works fine on a single CPU machine. But if we have more than one CPU available, we might want to distribute this work to the available cores. So instead of iterating in two nested for loops over all pixels, we can use a ForkJoinPool<\/code> and submit a new task for each row (or column) of the image. Once one row has been converted to grayscale, the current thread can work on another row.<\/p>\n
This principle is implemented in the following example:<\/p>\n
\npublic class GrayscaleImageAction extends RecursiveAction {\n\tprivate static final long serialVersionUID = 1L;\n\tprivate int row;\n\tprivate BufferedImage bufferedImage;\n\n\tpublic GrayscaleImageAction(int row, BufferedImage bufferedImage) {\n\t\tthis.row = row;\n\t\tthis.bufferedImage = bufferedImage;\n\t}\n\n\t@Override\n\tprotected void compute() {\n\t\tfor (int column = 0; column < bufferedImage.getWidth(); column++) {\n\t\t\tint rgb = bufferedImage.getRGB(column, row);\n\t\t\tint r = (rgb >> 16) & 0xFF;\n\t\t\tint g = (rgb >> 8) & 0xFF;\n\t\t\tint b = (rgb & 0xFF);\n\t\t\tint gray = (int) (0.2126 * (float) r + 0.7152 * (float) g + 0.0722 * (float) b);\n\t\t\tgray = (gray << 16) + (gray << 8) + gray;\n\t\t\tbufferedImage.setRGB(column, row, gray);\n\t\t}\n\t}\n\n\tpublic static void main(String[] args) throws IOException {\n\t\tForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());\n\t\tBufferedImage bufferedImage = ImageIO.read(new File(args[0]));\n\t\tfor (int row = 0; row < bufferedImage.getHeight(); row++) {\n\t\t\tGrayscaleImageAction action = new GrayscaleImageAction(row, bufferedImage);\n\t\t\tpool.execute(action);\n\t\t}\n\t\tpool.shutdown();\n\t\tImageIO.write(bufferedImage, \"jpg\", new File(args[1]));\n\t}\n}\n<\/pre>\n
Within our main() method we read the image using Java’s ImageIO<\/code> class. The returned instance of BufferedImage<\/code> has all the methods we need. We can query the number of rows and columns and retrieve and set the RGB value for each pixel. So all we do is to iterate over all rows and submit a new GrayscaleImageAction<\/code> to our ForkJoinPool<\/code>. The latter has received a hint about the available processors as a parameter to his constructor.<\/p>\n
The ForkJoinPool<\/code> now starts the tasks asynchronously by invoking their compute()<\/code> method. In this method we iterate over each row and update the corresponding RGB values by its grayscale value. After having submitted all tasks to the pool we wait in the main thread for the shutdown of the whole pool and then write the updated BufferedImage<\/code> back to disc by using the ImageIO.write()<\/code> method.<\/p>\n
Astonishingly we need only a few lines of code more than we would have needed without using the available processors. This again shows how much work we can save by using the available resources of the java.util.concurrent<\/code> package.<\/p>\n
The ForkJoinPool<\/code> offers three different methods for submitting a task:<\/p>\n