Класс Phaser

Регистрация. В отличие от других барьеров, число зарегистрированных для синхронизации на Phaser участников может меняться со временем. Задачи можно регистрировать в любой момент (с помощью методов register(), bulkRegister(int) или конструкторов, задающих начальное число участников) и при желании отменять их регистрацию при прибытии (с помощью arriveAndDeregister()). Как и в большинстве базовых средств синхронизации, регистрация и отмена регистрации влияют только на внутренние счетчики; они не ведут дополнительного внутреннего учета, поэтому задачи не могут узнать, зарегистрированы ли они. (Однако такой учет можно добавить, создав подкласс этого класса.)

Синхронизация. Как и CyclicBarrier, Phaser можно ожидать многократно. Метод arriveAndAwaitAdvance() действует аналогично CyclicBarrier.await. Каждому поколению Phaser соответствует номер фазы. Нумерация фаз начинается с нуля и увеличивается, когда все участники прибывают к Phaser, после чего при достижении Integer.MAX_VALUE цикл начинается снова с нуля. Номера фаз позволяют независимо управлять действиями при прибытии к Phaser и при ожидании остальных с помощью двух видов методов, которые может вызывать любой зарегистрированный участник:

• Прибытие. Методы arrive() и arriveAndDeregister() регистрируют прибытие. Эти методы не блокируют выполнение, а возвращают соответствующий номер фазы прибытия, то есть номер фазы Phaser, к которой относится прибытие. Когда прибывает последний участник для данной фазы, выполняется необязательное действие, и фаза увеличивается. Эти действия выполняет участник, вызвавший переход фазы; они определяются переопределением метода onAdvance(int, int), который также управляет завершением. Переопределение этого метода похоже на передачу действия барьера объекту CyclicBarrier, но предоставляет больше гибкости.
• Ожидание. Метод awaitAdvance(int) принимает аргумент с номером фазы прибытия и возвращает управление, когда Phaser переходит к другой фазе (или уже находится в ней). В отличие от аналогичных конструкций с использованием CyclicBarrier, метод awaitAdvance продолжает ожидание, даже если ожидающий поток прерван. Также доступны версии с обработкой прерываний и тайм-аутом, однако исключения, возникающие при прерываемом ожидании задач или ожидании с тайм-аутом, не изменяют состояние Phaser. При необходимости можно выполнить соответствующие действия по восстановлению в обработчиках этих исключений, часто после вызова forceTermination. Phaser также можно использовать задачам, выполняющимся в ForkJoinPool. Прогресс гарантирован, если уровень параллелизма пула позволяет разместить максимальное число одновременно заблокированных участников.

Завершение. Phaser может перейти в состояние завершения, которое можно проверить с помощью метода isTerminated(). После завершения все методы синхронизации немедленно возвращают управление, не дожидаясь перехода фазы, о чем свидетельствует отрицательное возвращаемое значение. Аналогично, попытки регистрации после завершения не дают эффекта. Завершение происходит, когда вызов onAdvance возвращает true. Реализация по умолчанию возвращает true, если отмена регистрации привела к тому, что число зарегистрированных участников стало равным нулю. Как показано ниже, когда Phaser управляет действиями с фиксированным числом итераций, часто удобно переопределить этот метод, чтобы завершение происходило при достижении текущим номером фазы заданного порога. Для немедленного освобождения ожидающих потоков и возможности их завершения также доступен метод forceTermination().

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

В дереве иерархически организованных Phaser регистрация и отмена регистрации дочерних Phaser у родительского выполняются автоматически. Когда число зарегистрированных участников дочернего Phaser становится ненулевым (как задается в конструкторе Phaser(Phaser,int), методах register() или bulkRegister(int)), дочерний Phaser регистрируется у родительского. Когда число зарегистрированных участников становится равным нулю в результате вызова arriveAndDeregister(), дочерний Phaser отменяет регистрацию у родительского.

Мониторинг. Хотя вызывать методы синхронизации могут только зарегистрированные участники, текущее состояние Phaser может отслеживать любой вызывающий код. В любой момент всего имеется getRegisteredParties() участников, из которых getArrivedParties() прибыли в текущей фазе (getPhase()). Когда прибывают остальные (getUnarrivedParties()) участники, фаза увеличивается. Значения, возвращаемые этими методами, могут отражать промежуточные состояния и поэтому обычно не подходят для управления синхронизацией. Метод toString() возвращает снимки этих состояний в форме, удобной для неформального мониторинга.

Эффекты согласованности памяти: действия, выполненные до вызова любого варианта метода прибытия, происходят до соответствующего перехода фазы и действий onAdvance (если они есть), которые, в свою очередь, происходят до действий после перехода фазы.

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

Phaser можно использовать вместо CountDownLatch для управления однократным действием, обслуживающим переменное число участников. Типичный способ — сначала зарегистрироваться в методе настройки, затем запустить все действия и отменить регистрацию, как показано ниже:

void runTasks(List<Runnable> tasks) {
  Phaser startingGate = new Phaser(1); // "1" to register self
  // create and start threads
  for (Runnable task : tasks) {
    startingGate.register();
    new Thread(() -> {
      startingGate.arriveAndAwaitAdvance();
      task.run();
    }).start();
  }

  // deregister self to allow threads to proceed
  startingGate.arriveAndDeregister();
}

Чтобы набор потоков многократно выполнял действия заданное число итераций, можно переопределить onAdvance:

void startTasks(List<Runnable> tasks, int iterations) {
  Phaser phaser = new Phaser() {
    protected boolean onAdvance(int phase, int registeredParties) {
      return phase >= iterations - 1 || registeredParties == 0;
    }
  };
  phaser.register();
  for (Runnable task : tasks) {
    phaser.register();
    new Thread(() -> {
      do {
        task.run();
        phaser.arriveAndAwaitAdvance();
      } while (!phaser.isTerminated());
    }).start();
  }
  // allow threads to proceed; don't wait for them
  phaser.arriveAndDeregister();
}

  // ...
  phaser.register();
  while (!phaser.isTerminated())
    phaser.arriveAndAwaitAdvance();

В ситуациях, когда вы уверены, что номер фазы никогда не перейдет через границу Integer.MAX_VALUE, можно использовать связанные конструкции для ожидания определенных номеров фаз. Например:

void awaitPhase(Phaser phaser, int phase) {
  int p = phaser.register(); // assumes caller not already registered
  while (p < phase) {
    if (phaser.isTerminated())
      // ... deal with unexpected termination
    else
      p = phaser.arriveAndAwaitAdvance();
  }
  phaser.arriveAndDeregister();
}

Чтобы создать набор задач n с помощью дерева Phaser, можно использовать код следующего вида, предполагая, что у класса Task есть конструктор, принимающий Phaser, в котором он регистрируется при создании. После вызова build(new Task[n], 0, n, new Phaser()) эти задачи можно запустить, например, отправив их в пул:

void build(Task[] tasks, int lo, int hi, Phaser ph) {
  if (hi - lo > TASKS_PER_PHASER) {
    for (int i = lo; i < hi; i += TASKS_PER_PHASER) {
      int j = Math.min(i + TASKS_PER_PHASER, hi);
      build(tasks, i, j, new Phaser(ph));
    }
  } else {
    for (int i = lo; i < hi; ++i)
      tasks[i] = new Task(ph);
      // assumes new Task(ph) performs ph.register()
  }
}

Примечания по реализации: Эта реализация ограничивает максимальное число участников значением 65535. Попытки зарегистрировать дополнительных участников приводят к IllegalStateException. Однако для поддержки сколь угодно больших групп участников можно и следует создавать иерархически организованные Phaser.