Класс CountedCompleter<T>

A ForkJoinTask with a completion action performed when triggered and there are no remaining pending actions. CountedCompleters are in general more robust in the presence of subtask stalls and blockage than are other forms of ForkJoinTasks, but are less intuitive to program. Uses of CountedCompleter are similar to those of other completion based components (such as CompletionHandler) except that multiple pending completions may be necessary to trigger the completion action onCompletion(CountedCompleter), not just one. Unless initialized otherwise, the pending count starts at zero, but may be (atomically) changed using methods setPendingCount(int), addToPendingCount(int), and compareAndSetPendingCount(int, int). Upon invocation of tryComplete(), if the pending action count is nonzero, it is decremented; otherwise, the completion action is performed, and if this completer itself has a completer, the process is continued with its completer. As is the case with related synchronization components such as Phaser and Semaphore, these methods affect only internal counts; they do not establish any further internal bookkeeping. In particular, the identities of pending tasks are not maintained. As illustrated below, you can create subclasses that do record some or all pending tasks or their results when needed. As illustrated below, utility methods supporting customization of completion traversals are also provided. However, because CountedCompleters provide only basic synchronization mechanisms, it may be useful to create further abstract subclasses that maintain linkages, fields, and additional support methods appropriate for a set of related usages.

A concrete CountedCompleter class must define method compute(), that should in most cases (as illustrated below), invoke tryComplete() once before returning. The class may also optionally override method onCompletion(CountedCompleter) to perform an action upon normal completion, and method onExceptionalCompletion(Throwable, CountedCompleter) to perform an action upon any exception.

CountedCompleters most often do not bear results, in which case they are normally declared as CountedCompleter<Void>, and will always return null as a result value. In other cases, you should override method getRawResult() to provide a result from join(), invoke(), and related methods. In general, this method should return the value of a field (or a function of one or more fields) of the CountedCompleter object that holds the result upon completion. Method setRawResult(T) by default plays no role in CountedCompleters. It is possible, but rarely applicable, to override this method to maintain other objects or fields holding result data.

A CountedCompleter that does not itself have a completer (i.e., one for which getCompleter() returns null) can be used as a regular ForkJoinTask with this added functionality. However, any completer that in turn has another completer serves only as an internal helper for other computations, so its own task status (as reported in methods such as ForkJoinTask.isDone()) is arbitrary; this status changes only upon explicit invocations of complete(T), ForkJoinTask.cancel(boolean), ForkJoinTask.completeExceptionally(Throwable) or upon exceptional completion of method compute. Upon any exceptional completion, the exception may be relayed to a task's completer (and its completer, and so on), if one exists and it has not otherwise already completed. Similarly, cancelling an internal CountedCompleter has only a local effect on that completer, so is not often useful.

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

Параллельное рекурсивное разбиение. 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();