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