open-dsi
/
postgres
mirror of https://github.com/postgres/postgres
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Modify interactions between sinval.c and sinvaladt.c. The code that actually

Browse Source 
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.

Per my proposal dated 25-Jun-2007 and followup discussion.
REL8_5_ALPHA1_BRANCH
Alvaro Herrera 18 years ago
parent a3f66eac01
commit ec6550c6c0
6 changed files with 173 additions and 190 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 4
      src/backend/storage/ipc/ipci.c
  2. 58
      src/backend/storage/ipc/sinval.c
  3. 198
      src/backend/storage/ipc/sinvaladt.c
  4. 6
      src/backend/utils/init/postinit.c
  5. 6
      src/include/storage/sinval.h
  6. 91
      src/include/storage/sinvaladt.h

4
src/backend/storage/ipc/ipci.c
Unescape View File

@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/ipc/ipci.c,v 1.94 2008/01/01 19:45:51 momjian Exp $
* $PostgreSQL: pgsql/src/backend/storage/ipc/ipci.c,v 1.95 2008/03/16 19:47:33 alvherre Exp $
*
*-------------------------------------------------------------------------
*/
@ -30,7 +30,7 @@
#include "storage/pg_shmem.h"
#include "storage/pmsignal.h"
#include "storage/procarray.h"
#include "storage/sinval.h"
#include "storage/sinvaladt.h"
#include "storage/spin.h"

58
src/backend/storage/ipc/sinval.c
Unescape View File

@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/ipc/sinval.c,v 1.83 2008/01/01 19:45:51 momjian Exp $
* $PostgreSQL: pgsql/src/backend/storage/ipc/sinval.c,v 1.84 2008/03/16 19:47:33 alvherre Exp $
*
*-------------------------------------------------------------------------
*/
@ -48,38 +48,6 @@ static volatile int catchupInterruptOccurred = 0;
static void ProcessCatchupEvent(void);
/****************************************************************************/
/* CreateSharedInvalidationState() Initialize SI buffer */
/* */
/* should be called only by the POSTMASTER */
/****************************************************************************/
void
CreateSharedInvalidationState(void)
{
/* SInvalLock must be initialized already, during LWLock init */
SIBufferInit();
}
/*
* InitBackendSharedInvalidationState
* Initialize new backend's state info in buffer segment.
*/
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")));
}
/*
* SendSharedInvalidMessage
* Add a shared-cache-invalidation message to the global SI message queue.
@ -89,9 +57,7 @@ SendSharedInvalidMessage(SharedInvalidationMessage *msg)
{
bool insertOK;
LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
insertOK = SIInsertDataEntry(shmInvalBuffer, msg);
LWLockRelease(SInvalLock);
insertOK = SIInsertDataEntry(msg);
if (!insertOK)
elog(DEBUG4, "SI buffer overflow");
}
@ -123,19 +89,7 @@ ReceiveSharedInvalidMessages(
*/
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);
getResult = SIGetDataEntry(MyBackendId, &data);
if (getResult == 0)
break; /* nothing more to do */
@ -155,11 +109,7 @@ ReceiveSharedInvalidMessages(
/* If we got any messages, try to release dead messages */
if (gotMessage)
{
LWLockAcquire(SInvalLock, LW_EXCLUSIVE);
SIDelExpiredDataEntries(shmInvalBuffer);
LWLockRelease(SInvalLock);
}
SIDelExpiredDataEntries(false);
}

198
src/backend/storage/ipc/sinvaladt.c
Unescape View File

@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.66 2008/01/01 19:45:51 momjian Exp $
* $PostgreSQL: pgsql/src/backend/storage/ipc/sinvaladt.c,v 1.67 2008/03/16 19:47:33 alvherre Exp $
*
*-------------------------------------------------------------------------
*/
@ -24,7 +24,82 @@
#include "storage/sinvaladt.h"
SISeg *shmInvalBuffer;
/*
* 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;
@ -49,13 +124,12 @@ SInvalShmemSize(void)
}
/*
* SIBufferInit
* Create and initialize a new SI message buffer
* SharedInvalBufferInit
* Create and initialize the SI message buffer
*/
void
SIBufferInit(void)
CreateSharedInvalidationState(void)
{
SISeg *segP;
Size size;
int i;
bool found;
@ -64,49 +138,43 @@ SIBufferInit(void)
size = offsetof(SISeg, procState);
size = add_size(size, mul_size(sizeof(ProcState), MaxBackends));
shmInvalBuffer = segP = (SISeg *)
shmInvalBuffer = (SISeg *)
ShmemInitStruct("shmInvalBuffer", size, &found);
if (found)
return;
segP->nextLXID = ShmemAlloc(sizeof(LocalTransactionId) * MaxBackends);
shmInvalBuffer->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;
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 < segP->maxBackends; i++)
for (i = 0; i < shmInvalBuffer->maxBackends; i++)
{
segP->procState[i].nextMsgNum = -1; /* inactive */
segP->procState[i].resetState = false;
segP->nextLXID[i] = InvalidLocalTransactionId;
shmInvalBuffer->procState[i].nextMsgNum = -1; /* inactive */
shmInvalBuffer->procState[i].resetState = false;
shmInvalBuffer->nextLXID[i] = InvalidLocalTransactionId;
}
}
/*
* SIBackendInit
* SharedInvalBackendInit
* 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)
void
SharedInvalBackendInit(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++)
@ -128,9 +196,14 @@ SIBackendInit(SISeg *segP)
}
else
{
/* out of procState slots */
/*
* out of procState slots: MaxBackends exceeded -- report normally
*/
MyBackendId = InvalidBackendId;
return 0;
LWLockRelease(SInvalLock);
ereport(FATAL,
(errcode(ERRCODE_TOO_MANY_CONNECTIONS),
errmsg("sorry, too many clients already")));
}
}
@ -153,10 +226,10 @@ SIBackendInit(SISeg *segP)
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));
return 1;
}
/*
@ -210,9 +283,16 @@ CleanupInvalidationState(int status, Datum arg)
* Returns true for normal successful insertion, false if had to reset.
*/
bool
SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data)
SIInsertDataEntry(SharedInvalidationMessage *data)
{
int numMsgs = segP->maxMsgNum - segP->minMsgNum;
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)
@ -222,12 +302,13 @@ SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data)
* messages but not yet have done SIDelExpiredDataEntries() to advance
* minMsgNum. So, make sure minMsgNum is up-to-date.
*/
SIDelExpiredDataEntries(segP);
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;
}
}
@ -246,7 +327,7 @@ SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data)
IsUnderPostmaster)
{
elog(DEBUG4, "SI table is 70%% full, signaling postmaster");
SendPostmasterSignal(PMSIGNAL_WAKEN_CHILDREN);
signal_postmaster = true;
}
/*
@ -255,6 +336,11 @@ SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data)
segP->buffer[segP->maxMsgNum % MAXNUMMESSAGES] = *data;
segP->maxMsgNum++;
LWLockRelease(SInvalLock);
if (signal_postmaster)
SendPostmasterSignal(PMSIGNAL_WAKEN_CHILDREN);
return true;
}
@ -293,14 +379,23 @@ SISetProcStateInvalid(SISeg *segP)
* -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().
* 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(SISeg *segP, int backendId,
SharedInvalidationMessage *data)
SIGetDataEntry(int backendId, SharedInvalidationMessage *data)
{
ProcState *stateP = &segP->procState[backendId - 1];
ProcState *stateP;
SISeg *segP;
LWLockAcquire(SInvalLock, LW_SHARED);
segP = shmInvalBuffer;
stateP = &segP->procState[backendId - 1];
if (stateP->resetState)
{
@ -310,11 +405,15 @@ SIGetDataEntry(SISeg *segP, int backendId,
*/
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.
@ -327,6 +426,8 @@ SIGetDataEntry(SISeg *segP, int backendId,
* delete it here. SIDelExpiredDataEntries() should be called to remove
* dead messages.
*/
LWLockRelease(SInvalLock);
return 1; /* got a message */
}
@ -335,15 +436,23 @@ SIGetDataEntry(SISeg *segP, int backendId,
* Remove messages that have been consumed by all active backends
*/
void
SIDelExpiredDataEntries(SISeg *segP)
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 */
@ -372,6 +481,9 @@ SIDelExpiredDataEntries(SISeg *segP)
segP->procState[i].nextMsgNum -= MSGNUMWRAPAROUND;
}
}
if (!locked)
LWLockRelease(SInvalLock);
}

6
src/backend/utils/init/postinit.c
Unescape View File

@ -8,7 +8,7 @@
*
*
* IDENTIFICATION
* $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.180 2008/01/01 19:45:53 momjian Exp $
* $PostgreSQL: pgsql/src/backend/utils/init/postinit.c,v 1.181 2008/03/16 19:47:34 alvherre Exp $
*
*
*-------------------------------------------------------------------------
@ -36,7 +36,7 @@
#include "storage/ipc.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "storage/sinval.h"
#include "storage/sinvaladt.h"
#include "storage/smgr.h"
#include "utils/acl.h"
#include "utils/flatfiles.h"
@ -411,7 +411,7 @@ InitPostgres(const char *in_dbname, Oid dboid, const char *username,
*/
MyBackendId = InvalidBackendId;
InitBackendSharedInvalidationState();
SharedInvalBackendInit();
if (MyBackendId > MaxBackends || MyBackendId <= 0)
elog(FATAL, "bad backend id: %d", MyBackendId);

6
src/include/storage/sinval.h
Unescape View File

@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* $PostgreSQL: pgsql/src/include/storage/sinval.h,v 1.46 2008/01/01 19:45:59 momjian Exp $
* $PostgreSQL: pgsql/src/include/storage/sinval.h,v 1.47 2008/03/16 19:47:34 alvherre Exp $
*
*-------------------------------------------------------------------------
*/
@ -83,10 +83,6 @@ typedef union
} SharedInvalidationMessage;
extern Size SInvalShmemSize(void);
extern void CreateSharedInvalidationState(void);
extern void InitBackendSharedInvalidationState(void);
extern void SendSharedInvalidMessage(SharedInvalidationMessage *msg);
extern void ReceiveSharedInvalidMessages(
void (*invalFunction) (SharedInvalidationMessage *msg),

91
src/include/storage/sinvaladt.h
Unescape View File

@ -7,7 +7,7 @@
* Portions Copyright (c) 1996-2008, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* $PostgreSQL: pgsql/src/include/storage/sinvaladt.h,v 1.45 2008/01/01 19:45:59 momjian Exp $
* $PostgreSQL: pgsql/src/include/storage/sinvaladt.h,v 1.46 2008/03/16 19:47:34 alvherre Exp $
*
*-------------------------------------------------------------------------
*/
@ -16,55 +16,16 @@
#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
{
@ -73,53 +34,17 @@ typedef struct ProcState
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 Size SInvalShmemSize(void);
extern void CreateSharedInvalidationState(void);
extern void SharedInvalBackendInit(void);
extern bool SIInsertDataEntry(SISeg *segP, SharedInvalidationMessage *data);
extern int SIGetDataEntry(SISeg *segP, int backendId,
SharedInvalidationMessage *data);
extern void SIDelExpiredDataEntries(SISeg *segP);
extern bool SIInsertDataEntry(SharedInvalidationMessage *data);
extern int SIGetDataEntry(int backendId, SharedInvalidationMessage *data);
extern void SIDelExpiredDataEntries(bool locked);
extern LocalTransactionId GetNextLocalTransactionId(void);

Write Preview
Loading…
Cancel
Save