Класс Phaser

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

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

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

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

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

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

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