Класс Phaser

Переиспользуемый барьер синхронизации, похожий по функциональности на CyclicBarrier и CountDownLatch, но поддерживающий более гибкое использование.

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

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

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

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

   // allow threads to start and deregister self
   phaser.arriveAndDeregister();
 }

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

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

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

Конструкторы

Phaser

public Phaser()

Создаёт новый фазер без первоначально зарегистрированных сторон, без родителя и начальным номером фазы 0. Любой поток, использующий этот фазер, должен сначала зарегистрироваться в нём.

Phaser

public Phaser(int parties)

Создаёт новый фазер с заданным количеством зарегистрированных, ещё не прибывших сторон, без родителя и начальным номером фазы 0.

Параметры:

parties - количество сторон, необходимое для перехода к следующей фазе

Исключения:

IllegalArgumentException - если количество сторон меньше нуля или больше максимального поддерживаемого количества сторон

Phaser

public Phaser(Phaser parent)

Эквивалентно Phaser(parent, 0).

Параметры:

parent - родительский фазер

Phaser

public Phaser(Phaser parent,
              int parties)

Создаёт новый фазер с заданным родителем и количеством зарегистрированных, ещё не прибывших сторон. Когда заданный родитель не null и заданное количество сторон больше нуля, этот дочерний фазер регистрируется у своего родителя.

Параметры:

parent - родительский фазер

parties - количество сторон, необходимых для перехода к следующей фазе

Исключения:

IllegalArgumentException - если количество сторон меньше нуля или больше максимального поддерживаемого количества сторон

Методы

register

public int register()

Добавляет новую, ещё не прибывшую сторону в этот фазер. Если в данный момент выполняется вызов onAdvance(int, int), этот метод может ожидать его завершения перед возвращением. Если у этого фазера есть родитель, и у этого фазера ранее не было зарегистрированных сторон, этот дочерний фазер также регистрируется у своего родителя. Если этот фазер завершён, попытка регистрации не имеет эффекта и возвращается отрицательное значение.

Возвращает:

номер фазы прибытия, к которой относится эта регистрация. Если это значение отрицательное, то этот фазер завершён, в этом случае регистрация не имеет эффекта.

Исключения:

IllegalStateException - если попытка регистрации превышает максимальное поддерживаемое количество сторон

bulkRegister

public int bulkRegister(int parties)

Добавляет заданное количество новых, ещё не прибывших сторон в этот фазер. Если в данный момент выполняется вызов onAdvance(int, int), этот метод может ожидать его завершения перед возвращением. Если у этого фазера есть родитель, и заданное количество сторон больше нуля, и у этого фазера ранее не было зарегистрированных сторон, этот дочерний фазер также регистрируется у своего родителя. Если этот фазер завершён, попытка регистрации не имеет эффекта и возвращается отрицательное значение.

Параметры:

parties - количество дополнительных сторон, необходимых для перехода к следующей фазе

Возвращает:

номер фазы прибытия, к которой относится эта регистрация. Если это значение отрицательное, то этот фазер завершён, в этом случае регистрация не имеет эффекта.

Исключения:

IllegalStateException - если попытка регистрации превышает максимальное поддерживаемое количество сторон

IllegalArgumentException - если parties < 0

arrive

public int arrive()

Возвращает:

номер фазы прибытия или отрицательное значение при завершении

Исключения:

IllegalStateException - если не завершён и количество не прибывших сторон станет отрицательным

arriveAndDeregister

public int arriveAndDeregister()

Возвращает:

номер фазы прибытия или отрицательное значение при завершении

Исключения:

IllegalStateException - если не завершён и количество зарегистрированных или не прибывших сторон станет отрицательным

arriveAndAwaitAdvance

public int arriveAndAwaitAdvance()

Возвращает:

номер фазы прибытия или (отрицательное) текущую фазу при завершении

Исключения:

IllegalStateException - если не завершён и количество не прибывших сторон станет отрицательным

awaitAdvance

public int awaitAdvance(int phase)

Ожидает перехода фазы этого фазера из заданного значения фазы, возвращаясь немедленно, если текущая фаза не равна заданному значению фазы или этот фазер завершён.

Параметры:

phase - номер фазы прибытия или отрицательное значение при завершении; этот аргумент обычно является значением, возвращённым предыдущим вызовом arrive или arriveAndDeregister.

Возвращает:

следующий номер фазы прибытия или аргумент, если он отрицательный, или (отрицательное) текущую фазу при завершении

awaitAdvanceInterruptibly

public int awaitAdvanceInterruptibly(int phase)
                              throws InterruptedException

Ожидает перехода фазы этого фазера из заданного значения фазы, бросая InterruptedException при прерывании ожидания или возвращаясь немедленно, если текущая фаза не равна заданному значению фазы или этот фазер завершён.

Параметры:

phase - номер фазы прибытия или отрицательное значение при завершении; этот аргумент обычно является значением, возвращённым предыдущим вызовом arrive или arriveAndDeregister.

Возвращает:

следующий номер фазы прибытия или аргумент, если он отрицательный, или (отрицательное) текущую фазу при завершении

Исключения:

InterruptedException - если поток прерван во время ожидания

awaitAdvanceInterruptibly

public int awaitAdvanceInterruptibly(int phase,
                                     long timeout,
                                     TimeUnit unit)
                              throws InterruptedException,
                                     TimeoutException

Ожидает перехода фазы этого фазера из заданного значения фазы или истечения заданного таймаута, бросая InterruptedException при прерывании ожидания или возвращаясь немедленно, если текущая фаза не равна заданному значению фазы или этот фазер завершён.

Параметры:

phase - номер фазы прибытия или отрицательное значение при завершении; этот аргумент обычно является значением, возвращённым предыдущим вызовом arrive или arriveAndDeregister.

timeout - время ожидания перед отказом, в единицах unit

unit - TimeUnit, определяющий, как интерпретировать параметр timeout

Возвращает:

следующий номер фазы прибытия или аргумент, если он отрицательный, или (отрицательное) текущую фазу при завершении

Исключения:

InterruptedException - если поток прерван во время ожидания

TimeoutException - если ожидание истекло

forceTermination

public void forceTermination()

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

getPhase

public final int getPhase()

Возвращает текущий номер фазы. Максимальный номер фазы – Integer.MAX_VALUE, после чего он сбрасывается до нуля. При завершении номер фазы отрицательный, в этом случае текущую фазу до завершения можно получить с помощью getPhase() + Integer.MIN_VALUE.

Возвращает:

номер фазы или отрицательное значение при завершении

getRegisteredParties

public int getRegisteredParties()

Возвращает количество сторон, зарегистрированных в этом фазере.

Возвращает:

количество сторон

getArrivedParties

public int getArrivedParties()

Возвращает количество зарегистрированных сторон, которые прибыли на текущую фазу этого фазера. Если этот фазер завершён, возвращаемое значение не имеет смысла и является произвольным.

Возвращает:

количество прибывших сторон

getUnarrivedParties

public int getUnarrivedParties()

Возвращает количество зарегистрированных сторон, которые ещё не прибыли на текущую фазу этого фазера. Если этот фазер завершён, возвращаемое значение не имеет смысла и является произвольным.

Возвращает:

количество не прибывших сторон

getParent

public Phaser getParent()

Возвращает родителя этого фазера или null , если его нет.

Возвращает:

родителя этого фазера или null , если его нет

getRoot

public Phaser getRoot()

Возвращает корневого предка этого фазера, который совпадает с этим фазером, если у него нет родителя.

Возвращает:

корневого предка этого фазера

isTerminated

public boolean isTerminated()

Возвращает true , если этот фазер завершён.

Возвращает:

true , если этот фазер завершён

onAdvance

protected boolean onAdvance(int phase,
                            int registeredParties)

Phaser phaser = new Phaser() {
   protected boolean onAdvance(int phase, int parties) { return false; }
 }

Параметры:

phase - текущий номер фазы при входе в этот метод, перед продвижением фазера

registeredParties - текущее количество зарегистрированных сторон

Возвращает:

true если этот фазер должен завершиться

toString

public String toString()

Возвращает строку, идентифицирующую этот фазер, а также его состояние. Состояние, в скобках, включает строку "phase = " за которой следует номер фазы, "parties = " за которым следует количество зарегистрированных сторон и "arrived = " за которым следует количество прибывших сторон.

Переопределяет:

toString в классе Object

Возвращает:

строку, идентифицирующую этот фазер, а также его состояние