Java Thread Pool fixed size and optimal utilization of all processor cores

Question

How to use exactly 8 threads for expensive parts all the time?

I have a problem with a number crunch, for which I created a simple structure. My problem is to find an elegant and easy way to optimally use all the processor cores.

To get good performance, I use a fixed-size stream pool of 8. The idea is to use as many threads as there are hardware threads for optimal performance.

The simplified use of pseudocode in the structure is as follows:

interface Task {
  data[] compute(data[]);
}

Task task = new Loop(new Chain(new DoX(), new DoY(), new Split(2, new DoZ())));
result = task.compute(data);

• The Loop Task will run until some completion criteria are met.
• The chain defines the chain of tasks (for example, in the above r = t1.compute (r); r = t2.compute (r); r = t3.compute (r); return r;)
• Split Task will split the data and complete the task in parts (for example, create 2 parts and return new data [] {t1.compute (part1), t1.compute (part2)})

Currently, stream processing is implemented in the separation task. Thus, the Split task will transfer the calculations of t1.compute (part1) and t1.compute (part2) to the thread pool.

Approach 1, possibly a complete dead end

, Split Task get() . , Split Task Split Task, wait future.get() , Split Task . , 8 . , , , .

1) , future.get() , ? , , .get(), , ? [ , ]

2, , , -

( ) , (partN) . , , partN-1, , . ( partN-1, partN-2... .

1), , , future.get() , .

3, ,

, 2 , . , (?) .

3.a.: , Split , , , . , . , Split Task, , , .

3.b.: , Split , , . Split Task, , .

2a) ?

2b) , , ? ?

.

3) 8 ? , ? , JVM ?

4) ?

---

[EDIT]

- ( 2)

ForkJoinPool .

. , ForkJoinPool (. ). .

http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinPool.html

" , IO ".

" 32767"

http://homes.cs.washington.edu/~djg/teachingMaterials/grossmanSPAC_forkJoinFramework.html

" ForkJoin , - 100 10000."

" " , 10 000 . / , , .

, 3a - , , ForkJoin, , , , , , , ( , , OS, , , ).

, 2 : , , , ForkJoinPool, , , .

, Split ( Split node, ), future.get(), , future.isDone(). , , . , , , .

: , , , ( , , , ?)

, , ThreadPool Split Task, - .

: ThreadPool - ThreadPool (, Job). ThreadPool .

, , . , . .

:

, "" , , . , . , , , ( , 2 ). ForkJoinPool, managedBlock() ForkJoinPool.ManagedBlocker, . , CompletionService , 2- , , , , , ForkJoinPool.

Answer 1

ForkJoinPool, . Loop Split, , , . RecursiveTask, . join() - (jdk1.7.0_45). . , RecursiveTask , Loop Split ( , ). , ForkJoinPool , google, .

: 2 , 1 Loop Split. FakeRecursiveTask ( ), ( Loop Split). HardWork , , - , doHardWork(work).

. , . , ( 8 ).

public class ThreadPool3 {
        private static int maxNumWorkerThreads;
        private static ExecutorService workerPool = null;
        private static ExecutorService helperPool = null;

        public static void initThreadPool(int maxNumWorkerThreads_) {
                int availProcessors = Runtime.getRuntime().availableProcessors();
                if (maxNumWorkerThreads_ <= 0) {
                        maxNumWorkerThreads_ = availProcessors;
                }
                maxNumWorkerThreads = maxNumWorkerThreads_;

                if (availProcessors != maxNumWorkerThreads) {
                        System.out.println("WARN: maxNumWorkerThreads (" + maxNumWorkerThreads + ") != availProcessors (" + availProcessors + ")");
                }
                workerPool = Executors.newFixedThreadPool(maxNumWorkerThreads);
                BlockingQueue<Runnable> workQueue = new SynchronousQueue<Runnable>();
                helperPool = new ThreadPoolExecutor(0, 4 * maxNumWorkerThreads, 60, TimeUnit.MINUTES, workQueue, Executors.defaultThreadFactory(),
        new ThreadPoolExecutor.CallerRunsPolicy());
        }


        public static abstract class HardWork implements Callable<Void> {
                @Override
                public abstract Void call() throws Exception;
        }

        public static void doHardWork(List<HardWork> tasks) throws Exception {
                workerPool.invokeAll(tasks);
        }


        /**
        * fake ForkJoinPoolInterface:
        *
        */
        public static abstract class FakeRecursiveTask<T> implements Callable<T> {
                private Future<T> resultFuture = null;

                /**
                * fake interface:
                */
                public abstract T compute();

                /**
                * fake interface:
                */
                public T invoke() {
                        return compute();
                }

                /**
                * fake interface:
                */
                public void fork() {
                        resultFuture = helperPool.submit(this);
                }

                /**
                * fake interface:
                */
                public T join() {
                        try {
                                return resultFuture.get();
                        }
                        catch (Exception e) {
                                throw new RuntimeException(e);
                        }
                }

                @Override
                public T call() throws Exception {
                        return compute();
                }
        }


        public static void shutdownThreadPool() {
                if (workerPool != null) {
                        workerPool.shutdown();
                }
                if (helperPool != null) {
                        helperPool.shutdown();
                }
        }
}

Answer 2

, , " ", "" .

, niave, , , , , , .

- "fork/join", Java 7. Oracle Java () , :

• Java ForkJoin Framework -

Answer 3

, Future.get(). CompletableFuture.then CompletableFuture.both , Java8. , , . Java8, Guava, ( ) . , . https://github.com/rfqu/df4j. Its advantage is that task objects can be reused, so you need to create fewer objects with numbers. If you provide a more detailed description of your problem (for example, in the usual sequential form or using an infinite number of threads), I can help you implement your program with df4j.