Класс CountedCompleter<T>

public abstract class CountedCompleter<T>
extends ForkJoinTask<T>

A ForkJoinTask с действием завершения, выполняемым при срабатывании и отсутствии оставшихся ожидающих действий. CountedCompleters, как правило, более устойчивы к задержкам и блокировкам подзадач, чем другие формы ForkJoinTasks, но программировать их менее интуитивно. Использование CountedCompleter аналогично использованию других компонентов, основанных на завершении (таких как CompletionHandler), за исключением того, что для запуска действия завершения onCompletion(CountedCompleter) могут потребоваться несколько ожидаемых завершений, а не только одно. Если не задано иное, счётчик ожиданий начинается с нуля, но может (атомарно) изменяться с помощью методов setPendingCount(int), addToPendingCount(int) и compareAndSetPendingCount(int, int). При вызове tryComplete(), если счётчик ожидающих действий не равен нулю, он уменьшается; в противном случае выполняется действие завершения, а если у этого комплетера есть свой комплетер, процесс продолжается с его комплетером. Как и в случае с связанными компонентами синхронизации, такими как Phaser и Semaphore, эти методы влияют только на внутренние счётчики; они не устанавливают дополнительную внутреннюю учётную запись. В частности, идентификаторы ожидающих задач не сохраняются. Как показано ниже, вы можете создавать подклассы, которые при необходимости записывают некоторые или все ожидающие задачи или их результаты. Как показано ниже, также предоставляются вспомогательные методы, поддерживающие настройку обхода завершений. Однако, поскольку CountedCompleters предоставляют только базовые механизмы синхронизации, может быть полезно создавать дополнительные абстрактные подклассы, которые поддерживают связи, поля и дополнительные вспомогательные методы, подходящие для набора связанных применений.

Конкретный класс CountedCompleter должен определить метод compute(), который в большинстве случаев (как показано ниже) должен вызывать tryComplete() один раз перед возвратом. Класс также может по желанию переопределять метод onCompletion(CountedCompleter) для выполнения действия при нормальном завершении и метод onExceptionalCompletion(Throwable, CountedCompleter) для выполнения действия при возникновении исключения.

CountedCompleters чаще всего не возвращают результаты, в этом случае они обычно объявляются как CountedCompleter<Void>, и всегда будут возвращать null в качестве значения результата. В других случаях вы должны переопределить метод getRawResult(), чтобы предоставить результат из join(), invoke(), и связанные методы. В общем случае этот метод должен возвращать значение поля (или функции от одного или нескольких полей) объекта CountedCompleter, содержащего результат при завершении. Метод setRawResult(T) по умолчанию не играет никакой роли в CountedCompleters. Возможно, но редко применимо, переопределить этот метод для поддержания других объектов или полей, хранящих данные результата.

CountedCompleter, который сам не имеет комплетера (т.е., для которого getCompleter() возвращает null) может использоваться как обычная ForkJoinTask с этой дополнительной функциональностью. Однако любой комплетер, у которого, в свою очередь, есть другой комплетер, служит только внутренним вспомогательным компонентом для других вычислений, поэтому его собственный статус задачи (как сообщается в методах, таких как Future.isDone()) является произвольным; этот статус изменяется только при явных вызовах complete(T), ForkJoinTask.cancel(boolean), ForkJoinTask.completeExceptionally(Throwable) или при исключительном завершении метода compute. При любом исключительном завершении исключение может быть передано комплетеру задачи (и его комплетеру, и так далее), если он существует и ещё не завершился. Аналогично, отмена внутреннего CountedCompleter оказывает только локальное влияние на этот комплетер, поэтому часто бесполезна.

Примеры использования.

Параллельная рекурсивная декомпозиция. CountedCompleters могут быть организованы в деревья, аналогичные тем, которые часто используются с RecursiveActions, хотя конструкции, участвующие в их настройке, обычно различаются. Здесь комплетер каждой задачи является её родителем в дереве вычислений. Несмотря на то, что они подразумевают немного больше учёта, CountedCompleters могут быть лучшим выбором при применении, возможно, ресурсоёмкой операции (которую нельзя далее разделить) к каждому элементу массива или коллекции; особенно когда операция занимает значительно разное время для завершения для некоторых элементов, либо из-за внутренней вариации (например, ввода-вывода), либо из-за побочных эффектов, таких как сборка мусора. Поскольку CountedCompleters предоставляют свои собственные продолжения, другие задачи не должны блокироваться в ожидании их выполнения.

Например, вот начальная версия вспомогательного метода, который использует рекурсивную декомпозицию деления на два, чтобы разделить работу на отдельные части (листовые задачи). Даже когда работа разделяется на отдельные вызовы, древовидные техники обычно предпочтительнее, чем прямое создание вилки листовых задач, поскольку они уменьшают межпоточную коммуникацию и улучшают балансировку нагрузки. В рекурсивном случае второе из каждой пары подзадач, завершающих работу, запускает завершение их родителя (поскольку не выполняется комбинация результатов, реализация по умолчанию без действий метода onCompletion не переопределяется). Вспомогательный метод настраивает корневую задачу и вызывает её (здесь неявно используется ForkJoinPool.commonPool()). Просто и надёжно (но не оптимально) всегда устанавливать счётчик ожиданий на количество дочерних задач и вызывать tryComplete() непосредственно перед возвратом.

public static <E> void forEach(E[] array, Consumer<E> action) {
   class Task extends CountedCompleter<Void> {
     final int lo, hi;
     Task(Task parent, int lo, int hi) {
       super(parent); this.lo = lo; this.hi = hi;
     }

     public void compute() {
       if (hi - lo >= 2) {
         int mid = (lo + hi) >>> 1;
         // must set pending count before fork
         setPendingCount(2);
         new Task(this, mid, hi).fork(); // right child
         new Task(this, lo, mid).fork(); // left child
       }
       else if (hi > lo)
         action.accept(array[lo]);
       tryComplete();
     }
   }
   new Task(null, 0, array.length).invoke();
 }

Этот дизайн можно улучшить, заметив, что в рекурсивном случае задача не имеет ничего общего после создания вилки своей правой задачи, поэтому может напрямую вызвать свою левую задачу перед возвратом. (Это аналог удаления рекурсии хвоста.) Также, когда последним действием в задаче является создание вилки или вызов подзадачи («вызов хвоста»), вызов tryComplete() может быть оптимизирован, за счёт того, что счётчик ожиданий будет выглядеть «с ошибкой в один».

public void compute() {
       if (hi - lo >= 2) {
         int mid = (lo + hi) >>> 1;
         setPendingCount(1); // looks off by one, but correct!
         new Task(this, mid, hi).fork(); // right child
         new Task(this, lo, mid).compute(); // direct invoke
       } else {
         if (hi > lo)
           action.accept(array[lo]);
         tryComplete();
       }
     }

В качестве дополнительной оптимизации обратите внимание, что левой задаче даже не нужно существовать. Вместо создания новой, мы можем продолжить использование исходной задачи и добавить счётчик ожиданий для каждой вилки. Кроме того, поскольку ни одна задача в этом дереве не реализует метод onCompletion(CountedCompleter), tryComplete можно заменить на propagateCompletion().

public void compute() {
       int n = hi - lo;
       for (; n >= 2; n /= 2) {
         addToPendingCount(1);
         new Task(this, lo + n/2, lo + n).fork();
       }
       if (n > 0)
         action.accept(array[lo]);
       propagateCompletion();
     }

При возможности предварительного расчёта счётчиков ожиданий, они могут быть установлены в конструкторе:

public static <E> void forEach(E[] array, Consumer<E> action) {
   class Task extends CountedCompleter<Void> {
     final int lo, hi;
     Task(Task parent, int lo, int hi) {
       super(parent, 31 - Integer.numberOfLeadingZeros(hi - lo));
       this.lo = lo; this.hi = hi;
     }

     public void compute() {
       for (int n = hi - lo; n >= 2; n /= 2)
         new Task(this, lo + n/2, lo + n).fork();
       action.accept(array[lo]);
       propagateCompletion();
     }
   }
   if (array.length > 0)
     new Task(null, 0, array.length).invoke();
 }

Дополнительные оптимизации таких классов могут включать специализацию классов для листовых шагов, деление на, скажем, четыре, а не два на итерацию, и использование адаптивного порога вместо всегда разделения на отдельные элементы.

Поиск. Дерево CountedCompleters может искать значение или свойство в разных частях структуры данных и сообщать о результате в AtomicReference как только оно будет найдено. Другие могут опросить результат, чтобы избежать ненужной работы. (Вы также могли бы отменить другие задачи, но обычно проще и эффективнее просто позволить им заметить, что результат установлен, и в этом случае пропустить дальнейшую обработку.) Ещё раз проиллюстрируем на примере массива, используя полное разделение (снова, на практике, листовые задачи почти всегда будут обрабатывать больше одного элемента):

class Searcher<E> extends CountedCompleter<E> {
   final E[] array; final AtomicReference<E> result; final int lo, hi;
   Searcher(CountedCompleter<?> p, E[] array, AtomicReference<E> result, int lo, int hi) {
     super(p);
     this.array = array; this.result = result; this.lo = lo; this.hi = hi;
   }
   public E getRawResult() { return result.get(); }
   public void compute() { // similar to ForEach version 3
     int l = lo, h = hi;
     while (result.get() == null && h >= l) {
       if (h - l >= 2) {
         int mid = (l + h) >>> 1;
         addToPendingCount(1);
         new Searcher(this, array, result, mid, h).fork();
         h = mid;
       }
       else {
         E x = array[l];
         if (matches(x) && result.compareAndSet(null, x))
           quietlyCompleteRoot(); // root task is now joinable
         break;
       }
     }
     tryComplete(); // normally complete whether or not found
   }
   boolean matches(E e) { ... } // return true if found

   public static <E> E search(E[] array) {
       return new Searcher<E>(null, array, new AtomicReference<E>(), 0, array.length).invoke();
   }
 }

В этом примере, а также в других, где задачи не имеют других эффектов, кроме compareAndSet общего результата, заключительный безусловный вызов tryComplete может быть сделан условным (if (result.get() == null) tryComplete();), так как дальнейшая учётная запись для управления завершениями не требуется после завершения корневой задачи.

Запись подзадач. Задачи CountedCompleter, которые комбинируют результаты нескольких подзадач, обычно нуждаются в доступе к этим результатам в методе onCompletion(CountedCompleter). Как показано в следующем классе (который выполняет упрощённую форму map-reduce, где отображения и сокращения являются всех типа E), один из способов сделать это в схемах «разделяй и властвуй» — это заставить каждую подзадачу записывать своего брата, чтобы к нему можно было обратиться в методе onCompletion. Эта техника применяется к сокращениям, в которых порядок комбинирования левых и правых результатов не имеет значения; упорядоченные сокращения требуют явных обозначений слева/справа. Также могут применяться варианты других оптимизаций, увиденных в примерах выше.

class MyMapper<E> { E apply(E v) {  ...  } }
 class MyReducer<E> { E apply(E x, E y) {  ...  } }
 class MapReducer<E> extends CountedCompleter<E> {
   final E[] array; final MyMapper<E> mapper;
   final MyReducer<E> reducer; final int lo, hi;
   MapReducer<E> sibling;
   E result;
   MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
              MyReducer<E> reducer, int lo, int hi) {
     super(p);
     this.array = array; this.mapper = mapper;
     this.reducer = reducer; this.lo = lo; this.hi = hi;
   }
   public void compute() {
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       MapReducer<E> left = new MapReducer(this, array, mapper, reducer, lo, mid);
       MapReducer<E> right = new MapReducer(this, array, mapper, reducer, mid, hi);
       left.sibling = right;
       right.sibling = left;
       setPendingCount(1); // only right is pending
       right.fork();
       left.compute();     // directly execute left
     }
     else {
       if (hi > lo)
           result = mapper.apply(array[lo]);
       tryComplete();
     }
   }
   public void onCompletion(CountedCompleter<?> caller) {
     if (caller != this) {
       MapReducer<E> child = (MapReducer<E>)caller;
       MapReducer<E> sib = child.sibling;
       if (sib == null || sib.result == null)
         result = child.result;
       else
         result = reducer.apply(child.result, sib.result);
     }
   }
   public E getRawResult() { return result; }

   public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
     return new MapReducer<E>(null, array, mapper, reducer,
                              0, array.length).invoke();
   }
 }

Здесь метод onCompletion принимает форму, общую для многих проектных решений завершения, которые объединяют результаты. Этот метод в стиле обратного вызова срабатывает один раз на задачу в любом из двух различных контекстов, в которых счётчик ожиданий равен или становится равным нулю: (1) самой задачей, если её счётчик ожиданий равен нулю при вызове tryComplete, или (2) любыми из её подзадач при их завершении и уменьшении счётчика ожиданий до нуля. Аргумент caller различает случаи. Чаще всего, когда вызывающий элемент является this, никаких действий не требуется. В противном случае вызывающий элемент может быть использован (обычно через приведение типов) для предоставления значения (и/или ссылок на другие значения) для объединения. Предполагая правильное использование счётчиков ожиданий, действия внутри onCompletion происходят (один раз) при завершении задачи и её подзадач. Для обеспечения безопасности потоков доступа к полям этой задачи или других завершенных задач в этом методе не требуется дополнительная синхронизация.

Обходы завершений. Если использование onCompletion для обработки завершений неприменимо или неудобно, вы можете использовать методы firstComplete() и nextComplete() для создания пользовательских обходов. Например, для определения MapReducer, который разделяет только правые задачи в форме третьего примера ForEach, завершения должны кооперативно уменьшаться вдоль неисчерпанных связей подзадач, что можно сделать следующим образом:

class MapReducer<E> extends CountedCompleter<E> { // version 2
   final E[] array; final MyMapper<E> mapper;
   final MyReducer<E> reducer; final int lo, hi;
   MapReducer<E> forks, next; // record subtask forks in list
   E result;
   MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
              MyReducer<E> reducer, int lo, int hi, MapReducer<E> next) {
     super(p);
     this.array = array; this.mapper = mapper;
     this.reducer = reducer; this.lo = lo; this.hi = hi;
     this.next = next;
   }
   public void compute() {
     int l = lo, h = hi;
     while (h - l >= 2) {
       int mid = (l + h) >>> 1;
       addToPendingCount(1);
       (forks = new MapReducer(this, array, mapper, reducer, mid, h, forks)).fork();
       h = mid;
     }
     if (h > l)
       result = mapper.apply(array[l]);
     // process completions by reducing along and advancing subtask links
     for (CountedCompleter<?> c = firstComplete(); c != null; c = c.nextComplete()) {
       for (MapReducer t = (MapReducer)c, s = t.forks; s != null; s = t.forks = s.next)
         t.result = reducer.apply(t.result, s.result);
     }
   }
   public E getRawResult() { return result; }

   public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
     return new MapReducer<E>(null, array, mapper, reducer,
                              0, array.length, null).invoke();
   }
 }

Триггеры. Некоторые CountedCompleters сами никогда не создаются вилкой, а вместо этого служат элементами соединений в других конструкциях; в том числе в тех, в которых завершение одной или нескольких асинхронных задач запускает другую асинхронную задачу. Например:

class HeaderBuilder extends CountedCompleter<...> { ... }
 class BodyBuilder extends CountedCompleter<...> { ... }
 class PacketSender extends CountedCompleter<...> {
   PacketSender(...) { super(null, 1); ... } // trigger on second completion
   public void compute() { } // never called
   public void onCompletion(CountedCompleter<?> caller) { sendPacket(); }
 }
 // sample use:
 PacketSender p = new PacketSender();
 new HeaderBuilder(p, ...).fork();
 new BodyBuilder(p, ...).fork();