How to Block a Queue in ForkJoinPool?

Question

I need to block threads on ForkJoinPool when its queue is full. This can be done in the standard ThreadPoolExecutor, e.g.:

private static ExecutorService newFixedThreadPoolWithQueueSize(int nThreads, int queueSize) {
    return new ThreadPoolExecutor(nThreads, nThreads,
            5000L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(queueSize, true), new ThreadPoolExecutor.CallerRunsPolicy());
}

I know, there is some Dequeue inside ForkJoinPool, but I don't have access to it via its API.

Update: Please see the answer below.

Gambrinus

What are the reasons you need this? My first thought is that if you need that kind of fine grained control over your jobs you are better off using a standard ExecutorService as you describe in the question. – Welltimed

@Kristoffer E The main reason is OutOfMemory. – Gambrinus

Answer 1

After some research I am happy to answer the question:

Reason: There is no such option in the ForkJoinPool's implementation due to the following reason. The majority of j.u.c. Executors assume single concurrent queue and many threads. This leads to the queue contention and degrades performance when reading/writing to the queue by multiple threads. Thus, such an approach is not quite scalable --> High contention on the queue can generate a large number of context switches and CPU-business.

Implementation: In the ForkJoinPool each thread has a separate double-ended queue (Deque) backed by an array. To minimize contention, Work-stealing happens at the tail of the deque, whereas task-submission happens at the head by current thread (worker). The tail contains the largest portion of work. In other words, stealing from the tail by another worker-thread minimizes the number of times to interact with other workers --> less contention, better overall performance.

• The idea is described in the official white paper "Java Fork/Join Framework" by Doug Lea.

• Scalability benchmarks are shown in "Let it crash - Scalability of Fork Join Pool"

Work-around thoughts: There's are global submission queues. Submissions from non-FJ threads enter into submission queues (Workers take these tasks). There are also Worker-queues mentioned above.

Maximum size for the queues is limited by the number:

   /**
     * Maximum size for queue arrays. Must be a power of two less
     * than or equal to 1 << (31 - width of array entry) to ensure
     * lack of wraparound of index calculations, but defined to a
     * value a bit less than this to help users trap runaway
     * programs before saturating systems.
     */
    static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M

When the queue is full an unchecked exception is thrown:

RejectedExecutionException("Queue capacity exceeded")

This is described in javadocs.

(Also, see ThreadPool's constructor for UncaughtExceptionHandler)

I tend to claim that current implementation doesn't have such a mechanism and this should be implemented in the consuming API by us.

For example, this could be done as follows:

1. Implement exponential back-Off logic that tries periodically resubmitting the tasks by incrementing time interval of the next retry. Or..
2. Write a throttler that checks periodically the size of submissionQueue (see ForkJoinPool.getQueuedSubmissionCount()).

Here's the official JSR-166E java code of ForkJoinPool for more information.

Gambrinus

Is there a solution hidden in this answer that I don't see or is it more like a discussion? (I run into the same problem) – Kelton