Класс Phaser

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

Синхронизация. Как и CyclicBarrier, фазу можно многократно ожидать. Метод 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() возвращает снимки этих запросов состояния в удобной для неформального мониторинга форме.

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

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

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