Класс CountedCompleter<T>

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

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

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

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

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

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

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

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

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

Поиск. Дерево CountedCompleter может выполнять поиск значения или свойства в разных частях структуры данных и сообщать результат в 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();
  }
}

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

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();