Класс CountedCompleter<T>

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

Задача 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 с этой дополнительной функциональностью. Однако любой завершитель, у которого, в свою очередь, есть другой завершитель, служит только внутренним помощником для других вычислений, поэтому его собственный статус задачи (как сообщается в методах, таких как ForkJoinTask.isDone()) произволен; этот статус изменяется только при явных вызовах complete(T), ForkJoinTask.cancel(boolean), ForkJoinTask.completeExceptionally(Throwable) или при возникновении исключения в методе compute. При любом исключительном завершении исключение может быть передано завершителю задачи (и его завершителю, и так далее), если он существует и не завершился иначе. Аналогично, отмена внутренней задачи CountedCompleter имеет только локальный эффект на эту задачу, поэтому часто бесполезна.

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

Параллельное рекурсивное разбиение. CountedCompleters могут быть организованы в деревья, аналогичные тем, что часто используются с RecursiveAction, хотя конструкции, используемые для их настройки, обычно различаются. Здесь завершитель каждой задачи является её родителем в дереве вычислений. Несмотря на то, что они требуют немного больше учёта, 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), один из способов сделать это в дизайнах divide and conquer — это заставить каждую подзадачу записывать своего брата, чтобы к нему можно было получить доступ в методе 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();