Класс 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 самостоятельно продолжают вычисления, другим задачам не нужно блокироваться в ожидании их выполнения.

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

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

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

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