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

 *

 * sinvaladt.c

 *    POSTGRES shared cache invalidation data manager.

 *

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

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

 *

 *

 * IDENTIFICATION

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

 *

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

 */

#include "postgres.h"

#include <signal.h>

#include <unistd.h>

#include "miscadmin.h"

#include "storage/ipc.h"

#include "storage/proc.h"

#include "storage/procnumber.h"

#include "storage/procsignal.h"

#include "storage/shmem.h"

#include "storage/sinvaladt.h"

#include "storage/subsystems.h"

/*

 * Conceptually, the shared cache invalidation messages are stored in an

 * infinite array, where maxMsgNum is the next array subscript to store a

 * submitted message in, minMsgNum is the smallest array subscript containing

 * a message not yet read by all backends, and we always have maxMsgNum >=

 * minMsgNum.  (They are equal when there are no messages pending.)  For each

 * active backend, there is a nextMsgNum pointer indicating the next message it

 * needs to read; we have maxMsgNum >= nextMsgNum >= minMsgNum for every

 * backend.

 *

 * (In the current implementation, minMsgNum is a lower bound for the

 * per-process nextMsgNum values, but it isn't rigorously kept equal to the

 * smallest nextMsgNum --- it may lag behind.  We only update it when

 * SICleanupQueue is called, and we try not to do that often.)

 *

 * In reality, the messages are stored in a circular buffer of MAXNUMMESSAGES

 * entries.  We translate MsgNum values into circular-buffer indexes by

 * computing MsgNum % MAXNUMMESSAGES (this should be fast as long as

 * MAXNUMMESSAGES is a constant and a power of 2).  As long as maxMsgNum

 * doesn't exceed minMsgNum by more than MAXNUMMESSAGES, we have enough space

 * in the buffer.  If the buffer does overflow, we recover by setting the

 * "reset" flag for each backend that has fallen too far behind.  A backend

 * that is in "reset" state is ignored while determining minMsgNum.  When

 * it does finally attempt to receive inval messages, it must discard all

 * its invalidatable state, since it won't know what it missed.

 *

 * To reduce the probability of needing resets, we send a "catchup" interrupt

 * to any backend that seems to be falling unreasonably far behind.  The

 * normal behavior is that at most one such interrupt is in flight at a time;

 * when a backend completes processing a catchup interrupt, it executes

 * SICleanupQueue, which will signal the next-furthest-behind backend if

 * needed.  This avoids undue contention from multiple backends all trying

 * to catch up at once.  However, the furthest-back backend might be stuck

 * in a state where it can't catch up.  Eventually it will get reset, so it

 * won't cause any more problems for anyone but itself.  But we don't want

 * to find that a bunch of other backends are now too close to the reset

 * threshold to be saved.  So SICleanupQueue is designed to occasionally

 * send extra catchup interrupts as the queue gets fuller, to backends that

 * are far behind and haven't gotten one yet.  As long as there aren't a lot

 * of "stuck" backends, we won't need a lot of extra interrupts, since ones

 * that aren't stuck will propagate their interrupts to the next guy.

 *

 * We would have problems if the MsgNum values overflow an integer, so

 * whenever minMsgNum exceeds MSGNUMWRAPAROUND, we subtract MSGNUMWRAPAROUND

 * from all the MsgNum variables simultaneously.  MSGNUMWRAPAROUND can be

 * large so that we don't need to do this often.  It must be a multiple of

 * MAXNUMMESSAGES so that the existing circular-buffer entries don't need

 * to be moved when we do it.

 *

 * Access to the shared sinval array is protected by two locks, SInvalReadLock

 * and SInvalWriteLock.  Readers take SInvalReadLock in shared mode; this

 * authorizes them to modify their own ProcState but not to modify or even

 * look at anyone else's.  When we need to perform array-wide updates,

 * such as in SICleanupQueue, we take SInvalReadLock in exclusive mode to

 * lock out all readers.  Writers take SInvalWriteLock (always in exclusive

 * mode) to serialize adding messages to the queue.  Note that a writer

 * can operate in parallel with one or more readers, because the writer

 * has no need to touch anyone's ProcState, except in the infrequent cases

 * when SICleanupQueue is needed.  The only point of overlap is that

 * the writer wants to change maxMsgNum while readers need to read it.

 * We deal with that by making maxMsgNum an atomic variable.  (The exact rule

 * is that you need to use a barrier-providing accessor to read maxMsgNum if

 * you are not holding SInvalWriteLock, and you need a barrier-providing

 * accessor to write maxMsgNum unless you are holding both locks.)  The

 * barriers ensure that messages written to the array are actually there

 * before maxMsgNum is increased, and that readers will see that data after

 * fetching maxMsgNum.

 */

/*

 * Configurable parameters.

 *

 * MAXNUMMESSAGES: max number of shared-inval messages we can buffer.

 * Must be a power of 2 for speed.

 *

 * MSGNUMWRAPAROUND: how often to reduce MsgNum variables to avoid overflow.

 * Must be a multiple of MAXNUMMESSAGES.  Should be large.

 *

 * CLEANUP_MIN: the minimum number of messages that must be in the buffer

 * before we bother to call SICleanupQueue.

 *

 * CLEANUP_QUANTUM: how often (in messages) to call SICleanupQueue once

 * we exceed CLEANUP_MIN.  Should be a power of 2 for speed.

 *

 * SIG_THRESHOLD: the minimum number of messages a backend must have fallen

 * behind before we'll send it PROCSIG_CATCHUP_INTERRUPT.

 *

 * WRITE_QUANTUM: the max number of messages to push into the buffer per

 * iteration of SIInsertDataEntries.  Noncritical but should be less than

 * CLEANUP_QUANTUM, because we only consider calling SICleanupQueue once

 * per iteration.

 */

#define MAXNUMMESSAGES 4096

#define MSGNUMWRAPAROUND (MAXNUMMESSAGES * 262144)

#define CLEANUP_MIN (MAXNUMMESSAGES / 2)

#define CLEANUP_QUANTUM (MAXNUMMESSAGES / 16)

#define SIG_THRESHOLD (MAXNUMMESSAGES / 2)

#define WRITE_QUANTUM 64

/* Per-backend state in shared invalidation structure */

typedef struct ProcState

{

  /* procPid is zero in an inactive ProcState array entry. */

  pid_t   procPid;    /* PID of backend, for signaling */

  /* nextMsgNum is meaningless if procPid == 0 or resetState is true. */

  uint32    nextMsgNum;   /* next message number to read */

  bool    resetState;   /* backend needs to reset its state */

  bool    signaled;   /* backend has been sent catchup signal */

  bool    hasMessages;  /* backend has unread messages */

  /*

   * Backend only sends invalidations, never receives them. This only makes

   * sense for Startup process during recovery because it doesn't maintain a

   * relcache, yet it fires inval messages to allow query backends to see

   * schema changes.

   */

  bool    sendOnly;   /* backend only sends, never receives */

  /*

   * Next LocalTransactionId to use for each idle backend slot.  We keep

   * this here because it is indexed by ProcNumber and it is convenient to

   * copy the value to and from local memory when MyProcNumber is set. It's

   * meaningless in an active ProcState entry.

   */

  LocalTransactionId nextLXID;

} ProcState;

/* Shared cache invalidation memory segment */

typedef struct SISeg

{

  /*

   * General state information

   */

  uint32    minMsgNum;    /* oldest message still needed */

  pg_atomic_uint32 maxMsgNum; /* next message number to be assigned */

  uint32    nextThreshold;  /* # of messages to call SICleanupQueue */

  /*

   * Circular buffer holding shared-inval messages

   */

  SharedInvalidationMessage buffer[MAXNUMMESSAGES];

  /*

   * Per-backend invalidation state info.

   *

   * 'procState' has NumProcStateSlots entries, and is indexed by pgprocno.

   * 'numProcs' is the number of slots currently in use, and 'pgprocnos' is

   * a dense array of their indexes, to speed up scanning all in-use slots.

   *

   * 'pgprocnos' is largely redundant with ProcArrayStruct->pgprocnos, but

   * having our separate copy avoids contention on ProcArrayLock, and allows

   * us to track only the processes that participate in shared cache

   * invalidations.

   */

  int     numProcs;

  int      *pgprocnos;

  ProcState procState[FLEXIBLE_ARRAY_MEMBER];

} SISeg;

/*

 * We reserve a slot for each possible ProcNumber, plus one for each

 * possible auxiliary process type.  (This scheme assumes there is not

 * more than one of any auxiliary process type at a time, except for

 * IO workers.)

 */

#define NumProcStateSlots (MaxBackends + NUM_AUXILIARY_PROCS)

static SISeg *shmInvalBuffer; /* pointer to the shared inval buffer */

static void SharedInvalShmemRequest(void *arg);

static void SharedInvalShmemInit(void *arg);

const ShmemCallbacks SharedInvalShmemCallbacks = {

  .request_fn = SharedInvalShmemRequest,

  .init_fn = SharedInvalShmemInit,

};

static LocalTransactionId nextLocalTransactionId;

static void CleanupInvalidationState(int status, Datum arg);

/*

 * SharedInvalShmemRequest

 *    Register shared memory needs for the SI message buffer

 */

static void

SharedInvalShmemRequest(void *arg)

{

  Size    size;

  size = offsetof(SISeg, procState);

  size = add_size(size, mul_size(sizeof(ProcState), NumProcStateSlots)); /* procState */

  size = add_size(size, mul_size(sizeof(int), NumProcStateSlots)); /* pgprocnos */

  ShmemRequestStruct(.name = "shmInvalBuffer",

             .size = size,

             .ptr = (void **) &shmInvalBuffer,

    );

}

static void

SharedInvalShmemInit(void *arg)

{

  int     i;

  /* Clear message counters */

  shmInvalBuffer->minMsgNum = 0;

  pg_atomic_init_u32(&shmInvalBuffer->maxMsgNum, 0);

  shmInvalBuffer->nextThreshold = CLEANUP_MIN;

  /* The buffer[] array is initially all unused, so we need not fill it */

  /* Mark all backends inactive, and initialize nextLXID */

  for (i = 0; i < NumProcStateSlots; i++)

  {

    shmInvalBuffer->procState[i].procPid = 0; /* inactive */

    shmInvalBuffer->procState[i].nextMsgNum = 0;  /* meaningless */

    shmInvalBuffer->procState[i].resetState = false;

    shmInvalBuffer->procState[i].signaled = false;

    shmInvalBuffer->procState[i].hasMessages = false;

    shmInvalBuffer->procState[i].nextLXID = InvalidLocalTransactionId;

  }

  shmInvalBuffer->numProcs = 0;

  shmInvalBuffer->pgprocnos = (int *) &shmInvalBuffer->procState[i];

}

/*

 * SharedInvalBackendInit

 *    Initialize a new backend to operate on the sinval buffer

 */

void

SharedInvalBackendInit(bool sendOnly)

{

  ProcState  *stateP;

  pid_t   oldPid;

  SISeg    *segP = shmInvalBuffer;

  if (MyProcNumber < 0)

    elog(ERROR, "MyProcNumber not set");

  if (MyProcNumber >= NumProcStateSlots)

    elog(PANIC, "unexpected MyProcNumber %d in SharedInvalBackendInit (max %d)",

       MyProcNumber, NumProcStateSlots);

  stateP = &segP->procState[MyProcNumber];

  /*

   * This can run in parallel with read operations, but not with write

   * operations, since SIInsertDataEntries relies on the pgprocnos array to

   * set hasMessages appropriately.

   */

  LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);

  oldPid = stateP->procPid;

  if (oldPid != 0)

  {

    LWLockRelease(SInvalWriteLock);

    elog(ERROR, "sinval slot for backend %d is already in use by process %d",

       MyProcNumber, (int) oldPid);

  }

  shmInvalBuffer->pgprocnos[shmInvalBuffer->numProcs++] = MyProcNumber;

  /* Fetch next local transaction ID into local memory */

  nextLocalTransactionId = stateP->nextLXID;

  /* mark myself active, with all extant messages already read */

  stateP->procPid = MyProcPid;

  stateP->nextMsgNum = pg_atomic_read_u32(&segP->maxMsgNum);

  stateP->resetState = false;

  stateP->signaled = false;

  stateP->hasMessages = false;

  stateP->sendOnly = sendOnly;

  LWLockRelease(SInvalWriteLock);

  /* register exit routine to mark my entry inactive at exit */

  on_shmem_exit(CleanupInvalidationState, PointerGetDatum(segP));

}

/*

 * CleanupInvalidationState

 *    Mark the current backend as no longer active.

 *

 * This function is called via on_shmem_exit() during backend shutdown.

 *

 * arg is really of type "SISeg*".

 */

static void

CleanupInvalidationState(int status, Datum arg)

{

  SISeg    *segP = (SISeg *) DatumGetPointer(arg);

  ProcState  *stateP;

  int     i;

  Assert(segP);

  LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);

  stateP = &segP->procState[MyProcNumber];

  /* Update next local transaction ID for next holder of this proc number */

  stateP->nextLXID = nextLocalTransactionId;

  /* Mark myself inactive */

  stateP->procPid = 0;

  stateP->nextMsgNum = 0;

  stateP->resetState = false;

  stateP->signaled = false;

  for (i = segP->numProcs - 1; i >= 0; i--)

  {

    if (segP->pgprocnos[i] == MyProcNumber)

    {

      if (i != segP->numProcs - 1)

        segP->pgprocnos[i] = segP->pgprocnos[segP->numProcs - 1];

      break;

    }

  }

  if (i < 0)

    elog(PANIC, "could not find entry in sinval array");

  segP->numProcs--;

  LWLockRelease(SInvalWriteLock);

}

/*

 * SIInsertDataEntries

 *    Add new invalidation message(s) to the buffer.

 */

void

SIInsertDataEntries(const SharedInvalidationMessage *data, int n)

{

  SISeg    *segP = shmInvalBuffer;

  /*

   * N can be arbitrarily large.  We divide the work into groups of no more

   * than WRITE_QUANTUM messages, to be sure that we don't hold the lock for

   * an unreasonably long time.  (This is not so much because we care about

   * letting in other writers, as that some just-caught-up backend might be

   * trying to do SICleanupQueue to pass on its signal, and we don't want it

   * to have to wait a long time.)  Also, we need to consider calling

   * SICleanupQueue every so often.

   */

  while (n > 0)

  {

    int     nthistime = Min(n, WRITE_QUANTUM);

    uint32    numMsgs;

    uint32    max;

    int     i;

    n -= nthistime;

    LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);

    /*

     * If the buffer is full, we *must* acquire some space.  Clean the

     * queue and reset anyone who is preventing space from being freed.

     * Otherwise, clean the queue only when it's exceeded the next

     * fullness threshold.  We have to loop and recheck the buffer state

     * after any call of SICleanupQueue.

     */

    for (;;)

    {

      numMsgs = pg_atomic_read_u32(&segP->maxMsgNum) - segP->minMsgNum;

      if (numMsgs + nthistime > MAXNUMMESSAGES ||

        numMsgs >= segP->nextThreshold)

        SICleanupQueue(true, nthistime);

      else

        break;

    }

    /*

     * Insert new message(s) into proper slot of circular buffer

     */

    max = pg_atomic_read_u32(&segP->maxMsgNum);

    while (nthistime-- > 0)

    {

      segP->buffer[max % MAXNUMMESSAGES] = *data++;

      max++;

    }

    /* Update current value of maxMsgNum using barrier */

    pg_atomic_write_membarrier_u32(&segP->maxMsgNum, max);

    /*

     * Now that the maxMsgNum change is globally visible, we give everyone

     * a swift kick to make sure they read the newly added messages.

     * Releasing SInvalWriteLock will enforce a full memory barrier, so

     * these (unlocked) changes will be committed to memory before we exit

     * the function.

     */

    for (i = 0; i < segP->numProcs; i++)

    {

      ProcState  *stateP = &segP->procState[segP->pgprocnos[i]];

      stateP->hasMessages = true;

    }

    LWLockRelease(SInvalWriteLock);

  }

}

/*

 * SIGetDataEntries

 *    get next SI message(s) for current backend, if there are any

 *

 * Possible return values:

 *  0:   no SI message available

 *  n>0: next n SI messages have been extracted into data[]

 * -1:   SI reset message extracted

 *

 * If the return value is less than the array size "datasize", the caller

 * can assume that there are no more SI messages after the one(s) returned.

 * Otherwise, another call is needed to collect more messages.

 *

 * NB: this can run in parallel with other instances of SIGetDataEntries

 * executing on behalf of other backends, since each instance will modify only

 * fields of its own backend's ProcState, and no instance will look at fields

 * of other backends' ProcStates.  We express this by grabbing SInvalReadLock

 * in shared mode.  Note that this is not exactly the normal (read-only)

 * interpretation of a shared lock! Look closely at the interactions before

 * allowing SInvalReadLock to be grabbed in shared mode for any other reason!

 *

 * NB: this can also run in parallel with SIInsertDataEntries.  It is not

 * guaranteed that we will return any messages added after the routine is

 * entered.

 *

 * Note: we assume that "datasize" is not so large that it might be important

 * to break our hold on SInvalReadLock into segments.

 */

int

SIGetDataEntries(SharedInvalidationMessage *data, int datasize)

{

  SISeg    *segP;

  ProcState  *stateP;

  uint32    max;

  int     n;

  segP = shmInvalBuffer;

  stateP = &segP->procState[MyProcNumber];

  /*

   * Before starting to take locks, do a quick, unlocked test to see whether

   * there can possibly be anything to read.  On a multiprocessor system,

   * it's possible that this load could migrate backwards and occur before

   * we actually enter this function, so we might miss a sinval message that

   * was just added by some other processor.  But they can't migrate

   * backwards over a preceding lock acquisition, so it should be OK.  If we

   * haven't acquired a lock preventing against further relevant

   * invalidations, any such occurrence is not much different than if the

   * invalidation had arrived slightly later in the first place.

   */

  if (!stateP->hasMessages)

    return 0;

  LWLockAcquire(SInvalReadLock, LW_SHARED);

  /*

   * We must reset hasMessages before determining how many messages we're

   * going to read.  That way, if new messages arrive after we have

   * determined how many we're reading, the flag will get reset and we'll

   * notice those messages part-way through.

   *

   * Note that, if we don't end up reading all of the messages, we had

   * better be certain to reset this flag before exiting!

   */

  stateP->hasMessages = false;

  /* Fetch current value of maxMsgNum using barrier */

  max = pg_atomic_read_membarrier_u32(&segP->maxMsgNum);

  if (stateP->resetState)

  {

    /*

     * Force reset.  We can say we have dealt with any messages added

     * since the reset, as well; and that means we should clear the

     * signaled flag, too.

     */

    stateP->nextMsgNum = max;

    stateP->resetState = false;

    stateP->signaled = false;

    LWLockRelease(SInvalReadLock);

    return -1;

  }

  /*

   * Retrieve messages and advance backend's counter, until data array is

   * full or there are no more messages.

   *

   * There may be other backends that haven't read the message(s), so we

   * cannot delete them here.  SICleanupQueue() will eventually remove them

   * from the queue.

   */

  n = 0;

  while (n < datasize && stateP->nextMsgNum < max)

  {

    data[n++] = segP->buffer[stateP->nextMsgNum % MAXNUMMESSAGES];

    stateP->nextMsgNum++;

  }

  /*

   * If we have caught up completely, reset our "signaled" flag so that

   * we'll get another signal if we fall behind again.

   *

   * If we haven't caught up completely, reset the hasMessages flag so that

   * we see the remaining messages next time.

   */

  if (stateP->nextMsgNum >= max)

    stateP->signaled = false;

  else

    stateP->hasMessages = true;

  LWLockRelease(SInvalReadLock);

  return n;

}

/*

 * SICleanupQueue

 *    Remove messages that have been consumed by all active backends

 *

 * callerHasWriteLock is true if caller is holding SInvalWriteLock.

 * minFree is the minimum number of message slots to make free.

 *

 * Possible side effects of this routine include marking one or more

 * backends as "reset" in the array, and sending PROCSIG_CATCHUP_INTERRUPT

 * to some backend that seems to be getting too far behind.  We signal at

 * most one backend at a time, for reasons explained at the top of the file.

 *

 * Caution: because we transiently release write lock when we have to signal

 * some other backend, it is NOT guaranteed that there are still minFree

 * free message slots at exit.  Caller must recheck and perhaps retry.

 */

void

SICleanupQueue(bool callerHasWriteLock, int minFree)

{

  SISeg    *segP = shmInvalBuffer;

  uint32    min,

        minsig,

        lowbound,

        numMsgs;

  int     i;

  ProcState  *needSig = NULL;

  /* Lock out all writers and readers */

  if (!callerHasWriteLock)

    LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);

  LWLockAcquire(SInvalReadLock, LW_EXCLUSIVE);

  /*

   * Recompute minMsgNum = minimum of all backends' nextMsgNum, identify the

   * furthest-back backend that needs signaling (if any), and reset any

   * backends that are too far back.  Note that because we ignore sendOnly

   * backends here it is possible for them to keep sending messages without

   * a problem even when they are the only active backend.

   */

  min = pg_atomic_read_u32(&segP->maxMsgNum);

  /* clamp at zero to avoid underflow */

  if (min > SIG_THRESHOLD)

    minsig = min - SIG_THRESHOLD;

  else

    minsig = 0;

  /* clamp at zero to avoid underflow */

  if (min + minFree > MAXNUMMESSAGES)

    lowbound = min + minFree - MAXNUMMESSAGES;

  else

    lowbound = 0;

  for (i = 0; i < segP->numProcs; i++)

  {

    ProcState  *stateP = &segP->procState[segP->pgprocnos[i]];

    uint32    n = stateP->nextMsgNum;

    /* Ignore if already in reset state */

    Assert(stateP->procPid != 0);

    if (stateP->resetState || stateP->sendOnly)

      continue;

    /*

     * If we must free some space and this backend is preventing it, force

     * him into reset state and then ignore until he catches up.

     */

    if (n < lowbound)

    {

      stateP->resetState = true;

      /* no point in signaling him ... */

      continue;

    }

    /* Track the global minimum nextMsgNum */

    if (n < min)

      min = n;

    /* Also see who's furthest back of the unsignaled backends */

    if (n < minsig && !stateP->signaled)

    {

      minsig = n;

      needSig = stateP;

    }

  }

  segP->minMsgNum = min;

  /*

   * When minMsgNum gets really large, decrement all message counters so as

   * to forestall overflow of the counters.  This happens seldom enough that

   * folding it into the previous loop would be a loser.

   */

  if (min >= MSGNUMWRAPAROUND)

  {

    segP->minMsgNum -= MSGNUMWRAPAROUND;

    pg_atomic_fetch_sub_u32(&segP->maxMsgNum, MSGNUMWRAPAROUND);

    for (i = 0; i < segP->numProcs; i++)

      segP->procState[segP->pgprocnos[i]].nextMsgNum -= MSGNUMWRAPAROUND;

  }

  /*

   * Determine how many messages are still in the queue, and set the

   * threshold at which we should repeat SICleanupQueue().

   */

  numMsgs = pg_atomic_read_u32(&segP->maxMsgNum) - segP->minMsgNum;

  if (numMsgs < CLEANUP_MIN)

    segP->nextThreshold = CLEANUP_MIN;

  else

    segP->nextThreshold = (numMsgs / CLEANUP_QUANTUM + 1) * CLEANUP_QUANTUM;

  /*

   * Lastly, signal anyone who needs a catchup interrupt.  Since

   * SendProcSignal() might not be fast, we don't want to hold locks while

   * executing it.

   */

  if (needSig)

  {

    pid_t   his_pid = needSig->procPid;

    ProcNumber  his_procNumber = (needSig - &segP->procState[0]);

    needSig->signaled = true;

    LWLockRelease(SInvalReadLock);

    LWLockRelease(SInvalWriteLock);

    elog(DEBUG4, "sending sinval catchup signal to PID %d", (int) his_pid);

    SendProcSignal(his_pid, PROCSIG_CATCHUP_INTERRUPT, his_procNumber);

    if (callerHasWriteLock)

      LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);

  }

  else

  {

    LWLockRelease(SInvalReadLock);

    if (!callerHasWriteLock)

      LWLockRelease(SInvalWriteLock);

  }

}

/*

 * GetNextLocalTransactionId --- allocate a new LocalTransactionId

 *

 * We split VirtualTransactionIds into two parts so that it is possible

 * to allocate a new one without any contention for shared memory, except

 * for a bit of additional overhead during backend startup/shutdown.

 * The high-order part of a VirtualTransactionId is a ProcNumber, and the

 * low-order part is a LocalTransactionId, which we assign from a local

 * counter.  To avoid the risk of a VirtualTransactionId being reused

 * within a short interval, successive procs occupying the same PGPROC slot

 * should use a consecutive sequence of local IDs, which is implemented

 * by copying nextLocalTransactionId as seen above.

 */

LocalTransactionId

GetNextLocalTransactionId(void)

{

  LocalTransactionId result;

  /* loop to avoid returning InvalidLocalTransactionId at wraparound */

  do

  {

    result = nextLocalTransactionId++;

  } while (!LocalTransactionIdIsValid(result));

  return result;

}