В Java 8 Executors.newWorkStealingPool() также обеспечивает очередь задач?

Question

Есть ли очередь ожидающих выполнения задач, используемых в сочетании с Java 8? Executors.newWorkStealingPool()?

Например, предположим, что число # доступных ядер равно 2, и Executors.newWorkStealingPool() пусто, потому что 2 задачи уже запущены. Что произойдет, если третье задание будет передано исполнителю по краже работы? Это в очереди? И если это так, каковы границы, если таковые имеются в указанной очереди?

Заранее спасибо.

Answer 1

Существует ли очередь ожидающих выполнения задач, используемая в сочетании с Executors.newWorkStealingPool() в Java 8?

Да, каждый поток поддерживает свою собственную деку. Когда один поток завершает выполнение своих задач, он берет задачу из очереди другого потока и выполняет ее.

И если это так, каковы границы, если таковые имеются в указанной очереди?

Максимальный размер очереди ограничен числом: static final int MAXIMUM_QUEUE_CAPACITY = 1 << 26; // 64M

Когда очередь заполнена, возникает непроверенное исключение: RejectedExecutionException("Queue capacity exceeded")

Answer 2

Из grepcode Исполнителей и ForkJoinPool

Executors,newWorkStealingPool возвращается ForkJoinPool

Исполнители:

 public static ExecutorService newWorkStealingPool() {
        return new ForkJoinPool
            (Runtime.getRuntime().availableProcessors(),
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }

ForkJoinPool:

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode) {
        this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

На execute():

public void execute(ForkJoinTask<?> task) {
        if (task == null)
            throw new NullPointerException();
        externalPush(task);
    }

externalPush звонки externalSubmit и вы можете увидеть WorkQueue подробности в этой реализации.

externalSubmit:

// Внешние операции

/**
 * Full version of externalPush, handling uncommon cases, as well
 * as performing secondary initialization upon the first
 * submission of the first task to the pool.  It also detects
 * first submission by an external thread and creates a new shared
 * queue if the one at index if empty or contended.
 *
 * @param task the task. Caller must ensure non-null.

 */

Вы можете найти более подробную информацию о размерах очереди в WorkQueue учебный класс

 static final class WorkQueue {

Документация по WokrQueue:

/**
     * Queues supporting work-stealing as well as external task
     * submission. See above for descriptions and algorithms.
     * Performance on most platforms is very sensitive to placement of
     * instances of both WorkQueues and their arrays -- we absolutely
     * do not want multiple WorkQueue instances or multiple queue
     * arrays sharing cache lines. The @Contended annotation alerts
     * JVMs to try to keep instances apart.
     */
    @sun.misc.Contended

 /**
     * Capacity of work-stealing queue array upon initialization.
     * Must be a power of two; at least 4, but should be larger to
     * reduce or eliminate cacheline sharing among queues.
     * Currently, it is much larger, as a partial workaround for
     * the fact that JVMs often place arrays in locations that
     * share GC bookkeeping (especially cardmarks) such that
     * per-write accesses encounter serious memory contention.
     */
    static final int INITIAL_QUEUE_CAPACITY = 1 << 13;

    /**
     * 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