YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

/*-------------------------------------------------------------------------

 *

 * barrier.c

 *    Barriers for synchronizing cooperating processes.

 *

 * Portions Copyright (c) 1996-2018, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994, Regents of the University of California

 *

 * From Wikipedia[1]: "In parallel computing, a barrier is a type of

 * synchronization method.  A barrier for a group of threads or processes in

 * the source code means any thread/process must stop at this point and cannot

 * proceed until all other threads/processes reach this barrier."

 *

 * This implementation of barriers allows for static sets of participants

 * known up front, or dynamic sets of participants which processes can join or

 * leave at any time.  In the dynamic case, a phase number can be used to

 * track progress through a parallel algorithm, and may be necessary to

 * synchronize with the current phase of a multi-phase algorithm when a new

 * participant joins.  In the static case, the phase number is used

 * internally, but it isn't strictly necessary for client code to access it

 * because the phase can only advance when the declared number of participants

 * reaches the barrier, so client code should be in no doubt about the current

 * phase of computation at all times.

 *

 * Consider a parallel algorithm that involves separate phases of computation

 * A, B and C where the output of each phase is needed before the next phase

 * can begin.

 *

 * In the case of a static barrier initialized with 4 participants, each

 * participant works on phase A, then calls BarrierArriveAndWait to wait until

 * all 4 participants have reached that point.  When BarrierArriveAndWait

 * returns control, each participant can work on B, and so on.  Because the

 * barrier knows how many participants to expect, the phases of computation

 * don't need labels or numbers, since each process's program counter implies

 * the current phase.  Even if some of the processes are slow to start up and

 * begin running phase A, the other participants are expecting them and will

 * patiently wait at the barrier.  The code could be written as follows:

 *

 *     perform_a();

 *     BarrierArriveAndWait(&barrier, ...);

 *     perform_b();

 *     BarrierArriveAndWait(&barrier, ...);

 *     perform_c();

 *     BarrierArriveAndWait(&barrier, ...);

 *

 * If the number of participants is not known up front, then a dynamic barrier

 * is needed and the number should be set to zero at initialization.  New

 * complications arise because the number necessarily changes over time as

 * participants attach and detach, and therefore phases B, C or even the end

 * of processing may be reached before any given participant has started

 * running and attached.  Therefore the client code must perform an initial

 * test of the phase number after attaching, because it needs to find out

 * which phase of the algorithm has been reached by any participants that are

 * already attached in order to synchronize with that work.  Once the program

 * counter or some other representation of current progress is synchronized

 * with the barrier's phase, normal control flow can be used just as in the

 * static case.  Our example could be written using a switch statement with

 * cases that fall-through, as follows:

 *

 *     phase = BarrierAttach(&barrier);

 *     switch (phase)

 *     {

 *     case PHASE_A:

 *         perform_a();

 *         BarrierArriveAndWait(&barrier, ...);

 *     case PHASE_B:

 *         perform_b();

 *         BarrierArriveAndWait(&barrier, ...);

 *     case PHASE_C:

 *         perform_c();

 *         BarrierArriveAndWait(&barrier, ...);

 *     }

 *     BarrierDetach(&barrier);

 *

 * Static barriers behave similarly to POSIX's pthread_barrier_t.  Dynamic

 * barriers behave similarly to Java's java.util.concurrent.Phaser.

 *

 * [1] https://en.wikipedia.org/wiki/Barrier_(computer_science)

 *

 * IDENTIFICATION

 *    src/backend/storage/ipc/barrier.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "storage/barrier.h"

static inline bool BarrierDetachImpl(Barrier *barrier, bool arrive);

/*

 * Initialize this barrier.  To use a static party size, provide the number of

 * participants to wait for at each phase indicating that that number of

 * backends is implicitly attached.  To use a dynamic party size, specify zero

 * here and then use BarrierAttach() and

 * BarrierDetach()/BarrierArriveAndDetach() to register and deregister

 * participants explicitly.

 */

void

BarrierInit(Barrier *barrier, int participants)

{

  SpinLockInit(&barrier->mutex);

  barrier->participants = participants;

  barrier->arrived = 0;

  barrier->phase = 0;

  barrier->elected = 0;

  barrier->static_party = participants > 0;

  ConditionVariableInit(&barrier->condition_variable);

}

/*

 * Arrive at this barrier, wait for all other attached participants to arrive

 * too and then return.  Increments the current phase.  The caller must be

 * attached.

 *

 * While waiting, pg_stat_activity shows a wait_event_class and wait_event

 * controlled by the wait_event_info passed in, which should be a value from

 * from one of the WaitEventXXX enums defined in pgstat.h.

 *

 * Return true in one arbitrarily chosen participant.  Return false in all

 * others.  The return code can be used to elect one participant to execute a

 * phase of work that must be done serially while other participants wait.

 */

bool

BarrierArriveAndWait(Barrier *barrier, uint32 wait_event_info)

{

  bool    release = false;

  bool    elected;

  int     start_phase;

  int     next_phase;

  SpinLockAcquire(&barrier->mutex);

  start_phase = barrier->phase;

  next_phase = start_phase + 1;

  ++barrier->arrived;

  if (barrier->arrived == barrier->participants)

  {

    release = true;

    barrier->arrived = 0;

    barrier->phase = next_phase;

    barrier->elected = next_phase;

  }

  SpinLockRelease(&barrier->mutex);

  /*

   * If we were the last expected participant to arrive, we can release our

   * peers and return true to indicate that this backend has been elected to

   * perform any serial work.

   */

  if (release)

  {

    ConditionVariableBroadcast(&barrier->condition_variable);

    return true;

  }

  /*

   * Otherwise we have to wait for the last participant to arrive and

   * advance the phase.

   */

  elected = false;

  ConditionVariablePrepareToSleep(&barrier->condition_variable);

  for (;;)

  {

    /*

     * We know that phase must either be start_phase, indicating that we

     * need to keep waiting, or next_phase, indicating that the last

     * participant that we were waiting for has either arrived or detached

     * so that the next phase has begun.  The phase cannot advance any

     * further than that without this backend's participation, because

     * this backend is attached.

     */

    SpinLockAcquire(&barrier->mutex);

    Assert(barrier->phase == start_phase || barrier->phase == next_phase);

    release = barrier->phase == next_phase;

    if (release && barrier->elected != next_phase)

    {

      /*

       * Usually the backend that arrives last and releases the other

       * backends is elected to return true (see above), so that it can

       * begin processing serial work while it has a CPU timeslice.

       * However, if the barrier advanced because someone detached, then

       * one of the backends that is awoken will need to be elected.

       */

      barrier->elected = barrier->phase;

      elected = true;

    }

    SpinLockRelease(&barrier->mutex);

    if (release)

      break;

    ConditionVariableSleep(&barrier->condition_variable, wait_event_info);

  }

  ConditionVariableCancelSleep();

  return elected;

}

/*

 * Arrive at this barrier, but detach rather than waiting.  Returns true if

 * the caller was the last to detach.

 */

bool

BarrierArriveAndDetach(Barrier *barrier)

{

  return BarrierDetachImpl(barrier, true);

}

/*

 * Attach to a barrier.  All waiting participants will now wait for this

 * participant to call BarrierArriveAndWait(), BarrierDetach() or

 * BarrierArriveAndDetach().  Return the current phase.

 */

int

BarrierAttach(Barrier *barrier)

{

  int     phase;

  Assert(!barrier->static_party);

  SpinLockAcquire(&barrier->mutex);

  ++barrier->participants;

  phase = barrier->phase;

  SpinLockRelease(&barrier->mutex);

  return phase;

}

/*

 * Detach from a barrier.  This may release other waiters from BarrierWait and

 * advance the phase if they were only waiting for this backend.  Return true

 * if this participant was the last to detach.

 */

bool

BarrierDetach(Barrier *barrier)

{

  return BarrierDetachImpl(barrier, false);

}

/*

 * Return the current phase of a barrier.  The caller must be attached.

 */

int

BarrierPhase(Barrier *barrier)

{

  /*

   * It is OK to read barrier->phase without locking, because it can't

   * change without us (we are attached to it), and we executed a memory

   * barrier when we either attached or participated in changing it last

   * time.

   */

  return barrier->phase;

}

/*

 * Return an instantaneous snapshot of the number of participants currently

 * attached to this barrier.  For debugging purposes only.

 */

int

BarrierParticipants(Barrier *barrier)

{

  int     participants;

  SpinLockAcquire(&barrier->mutex);

  participants = barrier->participants;

  SpinLockRelease(&barrier->mutex);

  return participants;

}

/*

 * Detach from a barrier.  If 'arrive' is true then also increment the phase

 * if there are no other participants.  If there are other participants

 * waiting, then the phase will be advanced and they'll be released if they

 * were only waiting for the caller.  Return true if this participant was the

 * last to detach.

 */

static inline bool

BarrierDetachImpl(Barrier *barrier, bool arrive)

{

  bool    release;

  bool    last;

  Assert(!barrier->static_party);

  SpinLockAcquire(&barrier->mutex);

  Assert(barrier->participants > 0);

  --barrier->participants;

  /*

   * If any other participants are waiting and we were the last participant

   * waited for, release them.  If no other participants are waiting, but

   * this is a BarrierArriveAndDetach() call, then advance the phase too.

   */

  if ((arrive || barrier->participants > 0) &&

    barrier->arrived == barrier->participants)

  {

    release = true;

    barrier->arrived = 0;

    ++barrier->phase;

  }

  else

    release = false;

  last = barrier->participants == 0;

  SpinLockRelease(&barrier->mutex);

  if (release)

    ConditionVariableBroadcast(&barrier->condition_variable);

  return last;

}