PostgreSQL数据库SI Message——Cache同步

本是古典何须时尚 2022-09-14 00:17 199阅读 0赞

在PostgreSQL中，每个进程都有属于自己的Cache。换句话说，同一个系统表在不同的进程中都有对应的Cache来缓存它的元组（对于RelCache来说缓存的是一个RelationData结构）。**同一个系统表的元组可能同时被多个进程的Cache所缓存，当其中某个Cache中的一个元组被删除或更新时，需要通知其他进程对其Cache进行同步。在PostgreSQL的实现中，会记录下已被删除的无效元组，并通过SI Message方式（即共享消息队列方式）在进程之间传递这一消息**。收到无效消息的进程将同步地把无效元组（或RelationData结构）从自己的Cache中删除。  
为了实现SI Message这一功能，PostgreSQL在共享内存中开辟了shmInvalBuffer记录系统中所发出的所有无效消息以及所有进程处理无消息的进度。shmInvalBuffer是一个全局变量，其数据类型为SISeg。

/* 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			nextThreshold;	/* # of messages to call SICleanupQueue */
    	int			lastBackend;	/* index of last active procState entry, +1 */
    	int			maxBackends;	/* size of procState array */
    	slock_t		msgnumLock;		/* spinlock protecting maxMsgNum */
    	/* Circular buffer holding shared-inval messages */
    	SharedInvalidationMessage buffer[MAXNUMMESSAGES];
    	/* Per-backend invalidation state info (has MaxBackends entries). */
    	ProcState	procState[FLEXIBLE_ARRAY_MEMBER];
    } SISeg;
    static SISeg *shmInvalBuffer;	/* pointer to the shared inval buffer */

在shmInvalBuffer中，无效消息存储在由Buffer字段指定的定长数组中（其长度MAXNUMMESSAGES预定义为4096）。该数组中每个元素存储一个无效消息，也可以称该数组为无效消息队列。**无效消息队列实际是一个环状结构**，最初数组为空时，新来的无效消息从前向后依次存放在数组的元素中，当数组被放满之后，新的无效消息将回到Buffer数组的头部开始插入。minMsgNum字段记录Buffer中还未被所有进程处理的无效消息编号中的最小值，maxMsgNum字段记录下一个可以用于存放新无效消息的数组元素下标。实际上，minMsgNum指出了Buffer中还没有被所有进程处理的无效消息的下界，而maxMsgNum则指出了上界，即编号比minMsgNum小的无效消息是已经被所有进程处理完的，而编号大于等于maxMsgNum的无效消息是还没有产生的，而两者之间的无效消息则是至少还有一个进程没有对其进行处理。因此在无效消息队列构成的环中，除了minMsgNum和maxMsgNum之间的位置之外，其他位置都可以用来存放新增加的无效消息。

typedef union
    { 
    	int8		id;				/* type field --- must be first */
    	SharedInvalCatcacheMsg cc;
    	SharedInvalCatalogMsg cat;
    	SharedInvalRelcacheMsg rc;
    	SharedInvalSmgrMsg sm;
    	SharedInvalRelmapMsg rm;
    	SharedInvalSnapshotMsg sn;
    } SharedInvalidationMessage;

PostgreSQL在shmInvalBuffer中用一个ProcState数组（procState字段）来存储正在读取无效消息的进程的读取进度，该数组的大小与系统允许的最大进程数MaxBackends有关，在默认情况下这个数组的大小为100（系统的默认最大进程数为100，可在postgresql.conf中修改）。**ProcState记录了PID为procPid的进程读取无效消息的状态，其中nextMsgNum的值介于shmInvalBuffer的minMsgNum值和maxMsgNum值之间**。

/* 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 */
    	PGPROC	   *proc;			/* PGPROC of backend */
    	/* nextMsgNum is meaningless if procPid == 0 or resetState is true. */
    	int			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 BackendId and it is convenient to * copy the value to and from local memory when MyBackendId is set. It's * meaningless in an active ProcState entry. */
    	LocalTransactionId nextLXID;
    } ProcState;

## 初始化 ##

SInvalShmemSize函数计算SInalShmem的大小为SISeg的大小加上其成员ProcState\*MaxBackends的大小。

/* SInvalShmemSize --- return shared-memory space needed */
    Size SInvalShmemSize(void) { 
    	Size		size;
    	size = offsetof(SISeg, procState);
    	size = add_size(size, mul_size(sizeof(ProcState), MaxBackends));
    	return size;
    }
    /* CreateSharedInvalidationState Create and initialize the SI message buffer */
    void CreateSharedInvalidationState(void) { 
    	int			i;
    	bool		found;
    	/* Allocate space in shared memory */
    	shmInvalBuffer = (SISeg *) ShmemInitStruct("shmInvalBuffer", SInvalShmemSize(), &found);
    	if (found) return;
    	/* Clear message counters, save size of procState array, init spinlock */
    	shmInvalBuffer->minMsgNum = 0;
    	shmInvalBuffer->maxMsgNum = 0;
    	shmInvalBuffer->nextThreshold = CLEANUP_MIN;
    	shmInvalBuffer->lastBackend = 0;
    	shmInvalBuffer->maxBackends = MaxBackends;
    	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 < shmInvalBuffer->maxBackends; i++){ 
    		shmInvalBuffer->procState[i].procPid = 0;	/* inactive */
    		shmInvalBuffer->procState[i].proc = NULL;
    		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;
    	}
    }

void SharedInvalBackendInit(bool sendOnly) { 
    	int			index;
    	ProcState  *stateP = NULL;
    	SISeg	   *segP = shmInvalBuffer;
    	/* This can run in parallel with read operations, but not with write operations, since SIInsertDataEntries relies on lastBackend to set hasMessages appropriately. */
    	LWLockAcquire(SInvalWriteLock, LW_EXCLUSIVE);
    	/* Look for a free entry in the procState array */
    	for (index = 0; index < segP->lastBackend; index++){ 
    		if (segP->procState[index].procPid == 0)	/* inactive slot? */{ 
    			stateP = &segP->procState[index];
    			break;
    		}
    	}
    	if (stateP == NULL){ 
    		if (segP->lastBackend < segP->maxBackends){ 
    			stateP = &segP->procState[segP->lastBackend];
    			Assert(stateP->procPid == 0);
    			segP->lastBackend++;
    		}else{ 
    			/* out of procState slots: MaxBackends exceeded -- report normally */
    			MyBackendId = InvalidBackendId;
    			LWLockRelease(SInvalWriteLock);
    			ereport(FATAL,(errcode(ERRCODE_TOO_MANY_CONNECTIONS),errmsg("sorry, too many clients already")));
    		}
    	}
    	MyBackendId = (stateP - &segP->procState[0]) + 1;
    	/* Advertise assigned backend ID in MyProc */
    	MyProc->backendId = MyBackendId;
    	/* Fetch next local transaction ID into local memory */
    	nextLocalTransactionId = stateP->nextLXID;
    	/* mark myself active, with all extant messages already read */
    	stateP->procPid = MyProcPid;
    	stateP->proc = MyProc;
    	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));
    	elog(DEBUG4, "my backend ID is %d", MyBackendId);
    }