sinvaladt.c File Reference

#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/spin.h"

Include dependency graph for sinvaladt.c:

Go to the source code of this file.

Data Structures
struct	ProcState
struct	SISeg

Macros
#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
#define	NumProcStateSlots   (MaxBackends + NUM_AUXILIARY_PROCS)

Typedefs
typedef struct ProcState	ProcState
typedef struct SISeg	SISeg

Functions
static void	CleanupInvalidationState (int status, Datum arg)
Size	SharedInvalShmemSize (void)
void	SharedInvalShmemInit (void)
void	SharedInvalBackendInit (bool sendOnly)
void	SIInsertDataEntries (const SharedInvalidationMessage *data, int n)
int	SIGetDataEntries (SharedInvalidationMessage *data, int datasize)
void	SICleanupQueue (bool callerHasWriteLock, int minFree)
LocalTransactionId	GetNextLocalTransactionId (void)

Variables
static SISeg *	shmInvalBuffer
static LocalTransactionId	nextLocalTransactionId

Macro Definition Documentation

CLEANUP_MIN

#define CLEANUP_MIN   (MAXNUMMESSAGES / 2)

Definition at line 131 of file sinvaladt.c.

CLEANUP_QUANTUM

#define CLEANUP_QUANTUM   (MAXNUMMESSAGES / 16)

Definition at line 132 of file sinvaladt.c.

MAXNUMMESSAGES

#define MAXNUMMESSAGES   4096

Definition at line 129 of file sinvaladt.c.

MSGNUMWRAPAROUND

#define MSGNUMWRAPAROUND   (MAXNUMMESSAGES * 262144)

Definition at line 130 of file sinvaladt.c.

NumProcStateSlots

#define NumProcStateSlots   (MaxBackends + NUM_AUXILIARY_PROCS)

Definition at line 204 of file sinvaladt.c.

SIG_THRESHOLD

#define SIG_THRESHOLD   (MAXNUMMESSAGES / 2)

Definition at line 133 of file sinvaladt.c.

WRITE_QUANTUM

#define WRITE_QUANTUM   64

Definition at line 134 of file sinvaladt.c.

Typedef Documentation

ProcState

typedef struct ProcState ProcState

SISeg

typedef struct SISeg SISeg

Function Documentation

CleanupInvalidationState()

static void CleanupInvalidationState ( int  status, Datum  arg  )

static

Definition at line 328 of file sinvaladt.c.

{
    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);
}

References arg, Assert, DatumGetPointer(), elog, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, nextLocalTransactionId, and PANIC.

Referenced by SharedInvalBackendInit().

GetNextLocalTransactionId()

LocalTransactionId GetNextLocalTransactionId

(

void

)

Definition at line 701 of file sinvaladt.c.

{
    LocalTransactionId result;
 
    /* loop to avoid returning InvalidLocalTransactionId at wraparound */
    do
    {
        result = nextLocalTransactionId++;
    } while (!LocalTransactionIdIsValid(result));
 
    return result;
}

References LocalTransactionIdIsValid, and nextLocalTransactionId.

Referenced by InitRecoveryTransactionEnvironment(), and StartTransaction().

SharedInvalBackendInit()

void SharedInvalBackendInit

(

bool

sendOnly

)

Definition at line 272 of file sinvaladt.c.

{
    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 = 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));
}

References CleanupInvalidationState(), elog, ERROR, fb(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyProcNumber, MyProcPid, nextLocalTransactionId, SISeg::numProcs, NumProcStateSlots, on_shmem_exit(), PANIC, SISeg::pgprocnos, PointerGetDatum(), and shmInvalBuffer.

Referenced by InitPostgres(), and InitRecoveryTransactionEnvironment().

SharedInvalShmemInit()

void SharedInvalShmemInit

(

void

)

Definition at line 234 of file sinvaladt.c.

{
    int         i;
    bool        found;
 
    /* Allocate space in shared memory */
    shmInvalBuffer = (SISeg *)
        ShmemInitStruct("shmInvalBuffer", SharedInvalShmemSize(), &found);
    if (found)
        return;
 
    /* Clear message counters, save size of procState array, init spinlock */
    shmInvalBuffer->minMsgNum = 0;
    shmInvalBuffer->maxMsgNum = 0;
    shmInvalBuffer->nextThreshold = CLEANUP_MIN;
    SpinLockInit(&shmInvalBuffer->msgnumLock);
 
    /* 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];
}

References CLEANUP_MIN, ProcState::hasMessages, i, InvalidLocalTransactionId, SISeg::maxMsgNum, SISeg::minMsgNum, SISeg::msgnumLock, ProcState::nextLXID, ProcState::nextMsgNum, SISeg::nextThreshold, SISeg::numProcs, NumProcStateSlots, SISeg::pgprocnos, ProcState::procPid, SISeg::procState, ProcState::resetState, SharedInvalShmemSize(), ShmemInitStruct(), shmInvalBuffer, ProcState::signaled, and SpinLockInit.

Referenced by CreateOrAttachShmemStructs().

SharedInvalShmemSize()

Size SharedInvalShmemSize

(

void

)

Definition at line 218 of file sinvaladt.c.

{
    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 */
 
    return size;
}

References add_size(), fb(), mul_size(), and NumProcStateSlots.

Referenced by CalculateShmemSize(), and SharedInvalShmemInit().

SICleanupQueue()

void SICleanupQueue	(	bool	callerHasWriteLock,
		int	minFree
	)

Definition at line 577 of file sinvaladt.c.

{
    SISeg      *segP = shmInvalBuffer;
    int         min,
                minsig,
                lowbound,
                numMsgs,
                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 = segP->maxMsgNum;
    minsig = min - SIG_THRESHOLD;
    lowbound = min - MAXNUMMESSAGES + minFree;
 
    for (i = 0; i < segP->numProcs; i++)
    {
        ProcState  *stateP = &segP->procState[segP->pgprocnos[i]];
        int         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;
        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 = 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);
    }
}

References Assert, CLEANUP_MIN, CLEANUP_QUANTUM, DEBUG4, elog, fb(), i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXNUMMESSAGES, MSGNUMWRAPAROUND, ProcState::nextMsgNum, PROCSIG_CATCHUP_INTERRUPT, SendProcSignal(), shmInvalBuffer, and SIG_THRESHOLD.

Referenced by ReceiveSharedInvalidMessages(), and SIInsertDataEntries().

SIGetDataEntries()

int SIGetDataEntries	(	SharedInvalidationMessage *	data,
		int	datasize
	)

Definition at line 473 of file sinvaladt.c.

{
    SISeg      *segP;
    ProcState  *stateP;
    int         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 spinlock */
    SpinLockAcquire(&segP->msgnumLock);
    max = segP->maxMsgNum;
    SpinLockRelease(&segP->msgnumLock);
 
    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;
}

References data, fb(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MAXNUMMESSAGES, MyProcNumber, SISeg::procState, shmInvalBuffer, SpinLockAcquire, and SpinLockRelease.

Referenced by ReceiveSharedInvalidMessages().

SIInsertDataEntries()

void SIInsertDataEntries	(	const SharedInvalidationMessage *	data,
		int	n
	)

Definition at line 370 of file sinvaladt.c.

{
    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);
        int         numMsgs;
        int         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 = 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 = segP->maxMsgNum;
        while (nthistime-- > 0)
        {
            segP->buffer[max % MAXNUMMESSAGES] = *data++;
            max++;
        }
 
        /* Update current value of maxMsgNum using spinlock */
        SpinLockAcquire(&segP->msgnumLock);
        segP->maxMsgNum = max;
        SpinLockRelease(&segP->msgnumLock);
 
        /*
         * 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);
    }
}

References data, fb(), ProcState::hasMessages, i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAXNUMMESSAGES, Min, shmInvalBuffer, SICleanupQueue(), SpinLockAcquire, SpinLockRelease, and WRITE_QUANTUM.

Referenced by SendSharedInvalidMessages().

Variable Documentation

nextLocalTransactionId

LocalTransactionId nextLocalTransactionId

static

Definition at line 209 of file sinvaladt.c.

Referenced by CleanupInvalidationState(), GetNextLocalTransactionId(), and SharedInvalBackendInit().

shmInvalBuffer

SISeg* shmInvalBuffer

static

Definition at line 206 of file sinvaladt.c.

Referenced by SharedInvalBackendInit(), SharedInvalShmemInit(), SICleanupQueue(), SIGetDataEntries(), and SIInsertDataEntries().