Класс 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(), если счетчик ожидающих действий не равен нулю, он уменьшается; в противном случае выполняется действие завершения, и если у этого completer есть собственный completer, процесс продолжается с его completer. Как и в случае с связанными компонентами синхронизации, такими как Phaser и Semaphore, эти методы влияют только на внутренние счетчики; они не выполняют дополнительную внутреннюю обработку. В частности, идентификаторы ожидающих задач не сохраняются. Как показано ниже, можно создавать подклассы, которые при необходимости записывают некоторые или все ожидающие задачи или их результаты. Как показано ниже, также предоставляются служебные методы, поддерживающие настройку обходов завершения. Однако, поскольку CountedCompleters предоставляют только базовые механизмы синхронизации, может быть полезно создавать дополнительные абстрактные подклассы, которые поддерживают связи, поля и дополнительные вспомогательные методы, подходящие для набора связанных случаев использования.

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

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

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

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

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

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

class MyOperation<E> { void apply(E e) { ... }  }

 class ForEach<E> extends CountedCompleter<Void> {

   public static <E> void forEach(E[] array, MyOperation<E> op) {
     new ForEach<E>(null, array, op, 0, array.length).invoke();
   }

   final E[] array; final MyOperation<E> op; final int lo, hi;
   ForEach(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi) {
     super(p);
     this.array = array; this.op = op; this.lo = lo; this.hi = hi;
   }

   public void compute() { // version 1
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       setPendingCount(2); // must set pending count before fork
       new ForEach(this, array, op, mid, hi).fork(); // right child
       new ForEach(this, array, op, lo, mid).fork(); // left child
     }
     else if (hi > lo)
       op.apply(array[lo]);
     tryComplete();
   }
 }

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

class ForEach<E> ...
   public void compute() { // version 2
     if (hi - lo >= 2) {
       int mid = (lo + hi) >>> 1;
       setPendingCount(1); // only one pending
       new ForEach(this, array, op, mid, hi).fork(); // right child
       new ForEach(this, array, op, lo, mid).compute(); // direct invoke
     }
     else {
       if (hi > lo)
         op.apply(array[lo]);
       tryComplete();
     }
   }

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

class ForEach<E> ...
   public void compute() { // version 3
     int l = lo,  h = hi;
     while (h - l >= 2) {
       int mid = (l + h) >>> 1;
       addToPendingCount(1);
       new ForEach(this, array, op, mid, h).fork(); // right child
       h = mid;
     }
     if (h > l)
       op.apply(array[l]);
     propagateCompletion();
   }

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

Поиск. Дерево 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();
   }
 }

В этом примере, а также в других случаях, когда задачи не имеют других эффектов, кроме сравнения и установки общего результата, заключительный безусловный вызов 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();

Конструкторы

CountedCompleter

protected CountedCompleter(CountedCompleter<?> completer,
                           int initialPendingCount)

Создаёт новый CountedCompleter с заданным completer и начальным значением pending count.

Параметры:

completer - completer данной задачи, или null при отсутствии

initialPendingCount - начальное значение pending count

CountedCompleter

protected CountedCompleter(CountedCompleter<?> completer)

Создаёт новый CountedCompleter с заданным completer и начальным значением pending count, равным нулю.

Параметры:

completer - completer данной задачи, или null при отсутствии

CountedCompleter

protected CountedCompleter()

Создаёт новый CountedCompleter без completer и начальным значением pending count, равным нулю.

Методы

compute

public abstract void compute()

Основное вычисление, выполняемое данной задачей.

onCompletion

public void onCompletion(CountedCompleter<?> caller)

Выполняет действие при вызове метода tryComplete(), когда pending count равен нулю, или при вызове безусловного метода complete(T). По умолчанию этот метод ничего не делает. Вы можете различать случаи, проверяя идентификатор переданного аргумента вызывающей стороны. Если он не равен this, то это обычно подзадача, которая может содержать результаты (и/или ссылки на другие результаты) для объединения.

Параметры:

caller - задача, вызывающая этот метод (которая может быть и самой задачей)

onExceptionalCompletion

public boolean onExceptionalCompletion(Throwable ex,
                                       CountedCompleter<?> caller)

Выполняет действие при вызове метода ForkJoinTask.completeExceptionally(Throwable) или при возникновении исключения в методе compute(), если эта задача ещё не завершилась обычным образом. При входе в этот метод данная задача ForkJoinTask.isCompletedAbnormally(). Возвращаемое значение этого метода управляет дальнейшей передачей: Если true и у этой задачи есть completer, который ещё не завершился, то этот completer также завершается с тем же исключением, что и этот completer. По умолчанию этот метод ничего не делает, кроме как возвращает true.

Параметры:

ex - исключение

caller - задача, вызывающая этот метод (которая может быть и самой задачей)

Возвращает:

true если это исключение должно быть передано completer этой задачи, если он существует

getCompleter

public final CountedCompleter<?> getCompleter()

Возвращает completer, установленный в конструкторе этой задачи, или null при отсутствии.

Возвращает:

completer

getPendingCount

public final int getPendingCount()

Возвращает текущее значение pending count.

Возвращает:

текущее значение pending count

setPendingCount

public final void setPendingCount(int count)

Устанавливает значение pending count в заданное значение.

Параметры:

count - значение

addToPendingCount

public final void addToPendingCount(int delta)

Добавляет (атомарно) заданное значение к pending count.

Параметры:

delta - значение для добавления

compareAndSetPendingCount

public final boolean compareAndSetPendingCount(int expected,
                                               int count)

Устанавливает (атомарно) pending count в заданное значение только в том случае, если он в настоящее время имеет заданное ожидаемое значение.

Параметры:

expected - ожидаемое значение

count - новое значение

Возвращает:

true при успехе

decrementPendingCountUnlessZero

public final int decrementPendingCountUnlessZero()

Если pending count не равен нулю, (атомарно) уменьшает его.

Возвращает:

начальное (неуменьшенное) значение pending count, хранящееся при входе в этот метод

getRoot

public final CountedCompleter<?> getRoot()

Возвращает корень текущего вычисления; т. е. эту задачу, если у неё нет completer, в противном случае корень completer.

Возвращает:

корень текущего вычисления

tryComplete

public final void tryComplete()

Если pending count не равен нулю, уменьшает его; в противном случае вызывает onCompletion(CountedCompleter), а затем аналогичным образом пытается завершить completer этой задачи, если он существует, в противном случае помечает эту задачу как завершённую.

propagateCompletion

public final void propagateCompletion()

Эквивалентно tryComplete(), но не вызывает onCompletion(CountedCompleter) по пути завершения: если pending count не равен нулю, уменьшает его; в противном случае аналогично пытается завершить completer этой задачи, если он существует, в противном случае помечает эту задачу как завершённую. Этот метод может быть полезен в случаях, когда onCompletion не должен или не должен вызываться для каждого completer в вычислении.

complete

public void complete(T rawResult)

Переопределяет:

complete в классе ForkJoinTask<T>

Параметры:

rawResult - значение rawResult

firstComplete

public final CountedCompleter<?> firstComplete()

Если pending count этой задачи равен нулю, возвращает эту задачу; в противном случае уменьшает pending count и возвращает null. Этот метод предназначен для использования с nextComplete() в циклах обхода завершения.

Возвращает:

эту задачу, если pending count был равен нулю, иначе null

nextComplete

public final CountedCompleter<?> nextComplete()

Если у этой задачи нет completer, вызывает ForkJoinTask.quietlyComplete() и возвращает null. Или, если pending count completer не равен нулю, уменьшает его и возвращает null. В противном случае возвращает completer. Этот метод может использоваться как часть цикла обхода завершения для однородных иерархий задач:

for (CountedCompleter<?> c = firstComplete();
      c != null;
      c = c.nextComplete()) {
   // ... process c ...
 }

Возвращает:

completer, или null при отсутствии

quietlyCompleteRoot

public final void quietlyCompleteRoot()

Эквивалентно getRoot().quietlyComplete().

helpComplete

public final void helpComplete(int maxTasks)

Если эта задача ещё не завершена, пытается обработать не более заданного числа других необработанных задач, для которых эта задача находится на пути завершения, если таковые существуют.

Параметры:

maxTasks - максимальное число задач для обработки. Если меньше или равно нулю, то задачи не обрабатываются.

exec

protected final boolean exec()

Реализует соглашения о выполнении для CountedCompleters.

Определено в:

exec в классе ForkJoinTask<T>

Возвращает:

true если эта задача известна как завершённая обычным образом

getRawResult

public T getRawResult()

Возвращает результат вычисления. По умолчанию возвращает null, что подходит для Void действий, но в других случаях следует переопределить, почти всегда, чтобы вернуть поле или функцию поля, которое содержит результат после завершения.

Определено в:

getRawResult в классе ForkJoinTask<T>

Возвращает:

результат вычисления

setRawResult

protected void setRawResult(T t)

Метод, который result-bearing CountedCompleters могут по желанию использовать для поддержания данных о результатах. По умолчанию ничего не делает. Переопределения не рекомендуются. Однако, если этот метод переопределён для обновления существующих объектов или полей, то он должен в общем случае быть потокобезопасным.

Определено в:

setRawResult в классе ForkJoinTask<T>

Параметры:

t - значение