I just modified the interactions in sinval.c and sinvaladt.c per
http://thread.gmane.org/gmane.comp.db.postgresql.devel.patches/18820/focus=18822
It looks a lot saner this way: the code that actually deals with the
queue, including locking etc, is all in sinvaladt.c. This means that
the struct definition of the queue, and the queue pointer, are now
internal "implementation details" inside sinvaladt.c.
One side effect of this change is that the call to SendPostmasterSignal
now occurs after the lock has been released. ISTM this is a good idea
on general principles (no syscalls in lwlocked code), but I'm wondering
if I created a thundering hoard problem that did not exist before.
All tests pass.
As a test of Review Board, I uploaded the patch to it:
http://reviewdemo.postgresql.org/r/19/
--
Alvaro Herrera http://www.CommandPrompt.com/
The PostgreSQL Company - Command Prompt, Inc.
Index: src/backend/storage/ipc/sinval.c
===================================================================
RCS file: /home/alvherre/Code/cvs/pgsql/src/backend/storage/ipc/sinval.c,v
retrieving revision 1.83
diff -c -p -r1.83 sinval.c
*** src/backend/storage/ipc/sinval.c 1 Jan 2008 19:45:51 -0000 1.83
--- src/backend/storage/ipc/sinval.c 29 Feb 2008 17:14:26 -0000
*************** static void ProcessCatchupEvent(void);
*** 56,62 ****
void
CreateSharedInvalidationState(void)
{
- /* SInvalLock must be initialized already, during LWLock init */
SIBufferInit();
}
--- 56,61 ----
*************** CreateSharedInvalidationState(void)
*** 67,83 ****
void
InitBackendSharedInvalidationState(void)
{
! int flag;
!
! LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
! flag = SIBackendInit(shmInvalBuffer);
! LWLockRelease(SInvalLock);
! if (flag < 0) /* unexpected problem */
! elog(FATAL, "shared cache invalidation initialization failed");
! if (flag == 0) /* expected problem: MaxBackends exceeded */
! ereport(FATAL,
! (errcode(ERRCODE_TOO_MANY_CONNECTIONS),
! errmsg("sorry, too many clients already")));
}
/*
--- 66,72 ----
void
InitBackendSharedInvalidationState(void)
{
! SIBackendInit();
}
/*
*************** SendSharedInvalidMessage(SharedInvalidat
*** 89,97 ****
{
bool insertOK;
! LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
! insertOK = SIInsertDataEntry(shmInvalBuffer, msg);
! LWLockRelease(SInvalLock);
if (!insertOK)
elog(DEBUG4, "SI buffer overflow");
}
--- 78,84 ----
{
bool insertOK;
! insertOK = SIInsertDataEntry(msg);
if (!insertOK)
elog(DEBUG4, "SI buffer overflow");
}
*************** ReceiveSharedInvalidMessages(
*** 123,141 ****
*/
catchupInterruptOccurred = 0;
! /*
! * We can run SIGetDataEntry in parallel with other backends running
! * SIGetDataEntry for themselves, 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
! * SInvalLock 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 SInvalLock to be grabbed in shared
! * mode for any other reason!
! */
! LWLockAcquire(SInvalLock, LW_SHARED);
! getResult = SIGetDataEntry(shmInvalBuffer, MyBackendId, &data);
! LWLockRelease(SInvalLock);
if (getResult == 0)
break; /* nothing more to do */
--- 110,116 ----
*/
catchupInterruptOccurred = 0;
! getResult = SIGetDataEntry(MyBackendId, &data);
if (getResult == 0)
break; /* nothing more to do */
*************** ReceiveSharedInvalidMessages(
*** 155,165 ****
/* If we got any messages, try to release dead messages */
if (gotMessage)
! {
! LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
! SIDelExpiredDataEntries(shmInvalBuffer);
! LWLockRelease(SInvalLock);
! }
}
--- 130,136 ----
/* If we got any messages, try to release dead messages */
if (gotMessage)
! SIDelExpiredDataEntries(false);
}
Index: src/backend/storage/ipc/sinvaladt.c
===================================================================
RCS file: /home/alvherre/Code/cvs/pgsql/src/backend/storage/ipc/sinvaladt.c,v
retrieving revision 1.66
diff -c -p -r1.66 sinvaladt.c
*** src/backend/storage/ipc/sinvaladt.c 1 Jan 2008 19:45:51 -0000 1.66
--- src/backend/storage/ipc/sinvaladt.c 29 Feb 2008 17:11:20 -0000
***************
*** 24,30 ****
#include "storage/sinvaladt.h"
! SISeg *shmInvalBuffer;
static LocalTransactionId nextLocalTransactionId;
--- 24,105 ----
#include "storage/sinvaladt.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 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 reset it to empty and
! * force each backend to "reset", ie, discard all its invalidatable state.
! *
! * 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.
! */
!
!
! /*
! * 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.
! */
!
! #define MAXNUMMESSAGES 4096
! #define MSGNUMWRAPAROUND (MAXNUMMESSAGES * 4096)
!
!
! /* Shared cache invalidation memory segment */
! typedef struct SISeg
! {
! /*
! * General state information
! */
! int minMsgNum; /* oldest message still needed */
! int maxMsgNum; /* next message number to be assigned */
! int lastBackend; /* index of last active procState entry, +1 */
! int maxBackends; /* size of procState array */
! int freeBackends; /* number of empty procState slots */
!
! /*
! * Next LocalTransactionId to use for each idle backend slot. We keep
! * this here because it is indexed by BackendId and it is convenient to
! * copy the value to and from local memory when MyBackendId is set.
! */
! LocalTransactionId *nextLXID; /* array of maxBackends entries */
!
! /*
! * Circular buffer holding shared-inval messages
! */
! SharedInvalidationMessage buffer[MAXNUMMESSAGES];
!
! /*
! * Per-backend state info.
! *
! * We declare procState as 1 entry because C wants a fixed-size array, but
! * actually it is maxBackends entries long.
! */
! ProcState procState[1]; /* reflects the invalidation state */
! } SISeg;
!
! static SISeg *shmInvalBuffer; /* pointer to the shared inval buffer */
!
static LocalTransactionId nextLocalTransactionId;
*************** SInvalShmemSize(void)
*** 55,61 ****
void
SIBufferInit(void)
{
- SISeg *segP;
Size size;
int i;
bool found;
--- 130,135 ----
*************** SIBufferInit(void)
*** 64,112 ****
size = offsetof(SISeg, procState);
size = add_size(size, mul_size(sizeof(ProcState), MaxBackends));
! shmInvalBuffer = segP = (SISeg *)
ShmemInitStruct("shmInvalBuffer", size, &found);
if (found)
return;
! segP->nextLXID = ShmemAlloc(sizeof(LocalTransactionId) * MaxBackends);
/* Clear message counters, save size of procState array */
! segP->minMsgNum = 0;
! segP->maxMsgNum = 0;
! segP->lastBackend = 0;
! segP->maxBackends = MaxBackends;
! segP->freeBackends = MaxBackends;
/* The buffer[] array is initially all unused, so we need not fill it */
/* Mark all backends inactive, and initialize nextLXID */
! for (i = 0; i < segP->maxBackends; i++)
{
! segP->procState[i].nextMsgNum = -1; /* inactive */
! segP->procState[i].resetState = false;
! segP->nextLXID[i] = InvalidLocalTransactionId;
}
}
/*
* SIBackendInit
* Initialize a new backend to operate on the sinval buffer
- *
- * Returns:
- * >0 A-OK
- * 0 Failed to find a free procState slot (ie, MaxBackends exceeded)
- * <0 Some other failure (not currently used)
- *
- * NB: this routine, and all following ones, must be executed with the
- * SInvalLock lock held, since there may be multiple backends trying
- * to access the buffer.
*/
! int
! SIBackendInit(SISeg *segP)
{
int index;
ProcState *stateP = NULL;
/* Look for a free entry in the procState array */
for (index = 0; index < segP->lastBackend; index++)
--- 138,180 ----
size = offsetof(SISeg, procState);
size = add_size(size, mul_size(sizeof(ProcState), MaxBackends));
! shmInvalBuffer = (SISeg *)
ShmemInitStruct("shmInvalBuffer", size, &found);
if (found)
return;
! shmInvalBuffer->nextLXID = ShmemAlloc(sizeof(LocalTransactionId) * MaxBackends);
/* Clear message counters, save size of procState array */
! shmInvalBuffer->minMsgNum = 0;
! shmInvalBuffer->maxMsgNum = 0;
! shmInvalBuffer->lastBackend = 0;
! shmInvalBuffer->maxBackends = MaxBackends;
! shmInvalBuffer->freeBackends = MaxBackends;
/* The buffer[] array is initially all unused, so we need not fill it */
/* Mark all backends inactive, and initialize nextLXID */
! for (i = 0; i < shmInvalBuffer->maxBackends; i++)
{
! shmInvalBuffer->procState[i].nextMsgNum = -1; /* inactive */
! shmInvalBuffer->procState[i].resetState = false;
! shmInvalBuffer->nextLXID[i] = InvalidLocalTransactionId;
}
}
/*
* SIBackendInit
* Initialize a new backend to operate on the sinval buffer
*/
! void
! SIBackendInit(void)
{
int index;
ProcState *stateP = NULL;
+ SISeg *segP = shmInvalBuffer;
+
+ LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
/* Look for a free entry in the procState array */
for (index = 0; index < segP->lastBackend; index++)
*************** SIBackendInit(SISeg *segP)
*** 128,136 ****
}
else
{
! /* out of procState slots */
MyBackendId = InvalidBackendId;
! return 0;
}
}
--- 196,209 ----
}
else
{
! /*
! * out of procState slots: MaxBackends exceeded -- report normally
! */
MyBackendId = InvalidBackendId;
! LWLockRelease(SInvalLock);
! ereport(FATAL,
! (errcode(ERRCODE_TOO_MANY_CONNECTIONS),
! errmsg("sorry, too many clients already")));
}
}
*************** SIBackendInit(SISeg *segP)
*** 153,162 ****
stateP->nextMsgNum = segP->maxMsgNum;
stateP->resetState = false;
/* register exit routine to mark my entry inactive at exit */
on_shmem_exit(CleanupInvalidationState, PointerGetDatum(segP));
-
- return 1;
}
/*
--- 226,235 ----
stateP->nextMsgNum = segP->maxMsgNum;
stateP->resetState = false;
+ LWLockRelease(SInvalLock);
+
/* register exit routine to mark my entry inactive at exit */
on_shmem_exit(CleanupInvalidationState, PointerGetDatum(segP));
}
/*
*************** CleanupInvalidationState(int status, Dat
*** 210,218 ****
* Returns true for normal successful insertion, false if had to reset.
*/
bool
! SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data)
{
! int numMsgs = segP->maxMsgNum - segP->minMsgNum;
/* Is the buffer full? */
if (numMsgs >= MAXNUMMESSAGES)
--- 283,298 ----
* Returns true for normal successful insertion, false if had to reset.
*/
bool
! SIInsertDataEntry(SharedInvalidationMessage *data)
{
! int numMsgs;
! bool signal_postmaster = false;
! SISeg *segP;
!
! LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
!
! segP = shmInvalBuffer;
! numMsgs = segP->maxMsgNum - segP->minMsgNum;
/* Is the buffer full? */
if (numMsgs >= MAXNUMMESSAGES)
*************** SIInsertDataEntry(SISeg *segP, SharedInv
*** 222,233 ****
* messages but not yet have done SIDelExpiredDataEntries() to advance
* minMsgNum. So, make sure minMsgNum is up-to-date.
*/
! SIDelExpiredDataEntries(segP);
numMsgs = segP->maxMsgNum - segP->minMsgNum;
if (numMsgs >= MAXNUMMESSAGES)
{
/* Yup, it's definitely full, no choice but to reset */
SISetProcStateInvalid(segP);
return false;
}
}
--- 302,314 ----
* messages but not yet have done SIDelExpiredDataEntries() to advance
* minMsgNum. So, make sure minMsgNum is up-to-date.
*/
! SIDelExpiredDataEntries(true);
numMsgs = segP->maxMsgNum - segP->minMsgNum;
if (numMsgs >= MAXNUMMESSAGES)
{
/* Yup, it's definitely full, no choice but to reset */
SISetProcStateInvalid(segP);
+ LWLockRelease(SInvalLock);
return false;
}
}
*************** SIInsertDataEntry(SISeg *segP, SharedInv
*** 246,252 ****
IsUnderPostmaster)
{
elog(DEBUG4, "SI table is 70%% full, signaling postmaster");
! SendPostmasterSignal(PMSIGNAL_WAKEN_CHILDREN);
}
/*
--- 327,333 ----
IsUnderPostmaster)
{
elog(DEBUG4, "SI table is 70%% full, signaling postmaster");
! signal_postmaster = true;
}
/*
*************** SIInsertDataEntry(SISeg *segP, SharedInv
*** 255,260 ****
--- 336,346 ----
segP->buffer[segP->maxMsgNum % MAXNUMMESSAGES] = *data;
segP->maxMsgNum++;
+ LWLockRelease(SInvalLock);
+
+ if (signal_postmaster)
+ SendPostmasterSignal(PMSIGNAL_WAKEN_CHILDREN);
+
return true;
}
*************** SISetProcStateInvalid(SISeg *segP)
*** 293,306 ****
* -1: SI reset message extracted
*
* NB: this can run in parallel with other instances of SIGetDataEntry
! * executing on behalf of other backends. See comments in sinval.c in
! * ReceiveSharedInvalidMessages().
*/
int
! SIGetDataEntry(SISeg *segP, int backendId,
! SharedInvalidationMessage *data)
{
! ProcState *stateP = &segP->procState[backendId - 1];
if (stateP->resetState)
{
--- 379,401 ----
* -1: SI reset message extracted
*
* NB: this can run in parallel with other instances of SIGetDataEntry
! * 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 SInvalLock 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 SInvalLock to be grabbed in shared mode for any other reason!
*/
int
! SIGetDataEntry(int backendId, SharedInvalidationMessage *data)
{
! ProcState *stateP;
! SISeg *segP;
!
! LWLockAcquire(SInvalLock, LW_SHARED);
!
! segP = shmInvalBuffer;
! stateP = &segP->procState[backendId - 1];
if (stateP->resetState)
{
*************** SIGetDataEntry(SISeg *segP, int backendI
*** 310,320 ****
--- 405,419 ----
*/
stateP->resetState = false;
stateP->nextMsgNum = segP->maxMsgNum;
+ LWLockRelease(SInvalLock);
return -1;
}
if (stateP->nextMsgNum >= segP->maxMsgNum)
+ {
+ LWLockRelease(SInvalLock);
return 0; /* nothing to read */
+ }
/*
* Retrieve message and advance my counter.
*************** SIGetDataEntry(SISeg *segP, int backendI
*** 327,332 ****
--- 426,433 ----
* delete it here. SIDelExpiredDataEntries() should be called to remove
* dead messages.
*/
+
+ LWLockRelease(SInvalLock);
return 1; /* got a message */
}
*************** SIGetDataEntry(SISeg *segP, int backendI
*** 335,349 ****
* Remove messages that have been consumed by all active backends
*/
void
! SIDelExpiredDataEntries(SISeg *segP)
{
int min,
i,
h;
min = segP->maxMsgNum;
if (min == segP->minMsgNum)
return; /* fast path if no messages exist */
/* Recompute minMsgNum = minimum of all backends' nextMsgNum */
--- 436,458 ----
* Remove messages that have been consumed by all active backends
*/
void
! SIDelExpiredDataEntries(bool locked)
{
+ SISeg *segP = shmInvalBuffer;
int min,
i,
h;
+ if (!locked)
+ LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
+
min = segP->maxMsgNum;
if (min == segP->minMsgNum)
+ {
+ if (!locked)
+ LWLockRelease(SInvalLock);
return; /* fast path if no messages exist */
+ }
/* Recompute minMsgNum = minimum of all backends' nextMsgNum */
*************** SIDelExpiredDataEntries(SISeg *segP)
*** 372,377 ****
--- 481,489 ----
segP->procState[i].nextMsgNum -= MSGNUMWRAPAROUND;
}
}
+
+ if (!locked)
+ LWLockRelease(SInvalLock);
}
Index: src/include/storage/sinvaladt.h
===================================================================
RCS file: /home/alvherre/Code/cvs/pgsql/src/include/storage/sinvaladt.h,v
retrieving revision 1.45
diff -c -p -r1.45 sinvaladt.h
*** src/include/storage/sinvaladt.h 1 Jan 2008 19:45:59 -0000 1.45
--- src/include/storage/sinvaladt.h 29 Feb 2008 17:09:37 -0000
***************
*** 16,70 ****
#include "storage/sinval.h"
-
/*
* The shared cache invalidation manager is responsible for transmitting
* invalidation messages between backends. Any message sent by any backend
* must be delivered to all already-running backends before it can be
* forgotten.
! *
! * Conceptually, the 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 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 reset it to empty and
! * force each backend to "reset", ie, discard all its invalidatable state.
! *
! * 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.
! *
* The struct type SharedInvalidationMessage, defining the contents of
* a single message, is defined in sinval.h.
*/
-
- /*
- * 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.
- */
-
- #define MAXNUMMESSAGES 4096
- #define MSGNUMWRAPAROUND (MAXNUMMESSAGES * 4096)
-
-
/* Per-backend state in shared invalidation structure */
typedef struct ProcState
{
--- 16,31 ----
#include "storage/sinval.h"
/*
* The shared cache invalidation manager is responsible for transmitting
* invalidation messages between backends. Any message sent by any backend
* must be delivered to all already-running backends before it can be
* forgotten.
! *
* The struct type SharedInvalidationMessage, defining the contents of
* a single message, is defined in sinval.h.
*/
/* Per-backend state in shared invalidation structure */
typedef struct ProcState
{
*************** typedef struct ProcState
*** 73,125 ****
bool resetState; /* true, if backend has to reset its state */
} ProcState;
- /* Shared cache invalidation memory segment */
- typedef struct SISeg
- {
- /*
- * General state information
- */
- int minMsgNum; /* oldest message still needed */
- int maxMsgNum; /* next message number to be assigned */
- int lastBackend; /* index of last active procState entry, +1 */
- int maxBackends; /* size of procState array */
- int freeBackends; /* number of empty procState slots */
-
- /*
- * Next LocalTransactionId to use for each idle backend slot. We keep
- * this here because it is indexed by BackendId and it is convenient to
- * copy the value to and from local memory when MyBackendId is set.
- */
- LocalTransactionId *nextLXID; /* array of maxBackends entries */
-
- /*
- * Circular buffer holding shared-inval messages
- */
- SharedInvalidationMessage buffer[MAXNUMMESSAGES];
-
- /*
- * Per-backend state info.
- *
- * We declare procState as 1 entry because C wants a fixed-size array, but
- * actually it is maxBackends entries long.
- */
- ProcState procState[1]; /* reflects the invalidation state */
- } SISeg;
-
-
- extern SISeg *shmInvalBuffer; /* pointer to the shared inval buffer */
-
/*
* prototypes for functions in sinvaladt.c
*/
extern void SIBufferInit(void);
! extern int SIBackendInit(SISeg *segP);
! extern bool SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data);
! extern int SIGetDataEntry(SISeg *segP, int backendId,
! SharedInvalidationMessage *data);
! extern void SIDelExpiredDataEntries(SISeg *segP);
extern LocalTransactionId GetNextLocalTransactionId(void);
--- 34,49 ----
bool resetState; /* true, if backend has to reset its state */
} ProcState;
/*
* prototypes for functions in sinvaladt.c
*/
extern void SIBufferInit(void);
! extern void SIBackendInit(void);
! extern bool SIInsertDataEntry(SharedInvalidationMessage *data);
! extern int SIGetDataEntry(int backendId, SharedInvalidationMessage *data);
! extern void SIDelExpiredDataEntries(bool locked);
extern LocalTransactionId GetNextLocalTransactionId(void);
---------------------------(end of broadcast)---------------------------
TIP 9: In versions below 8.0, the planner will ignore your desire to
choose an index scan if your joining column's datatypes do not
match
