PostgreSQL数据库并发事务——PROCARRAY创建

小鱼儿 2022-09-09 09:17 209阅读 0赞

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]

### ProcArray创建 ###

这里首先介绍小tick `#define offsetof(type, field) ((long) &((type *)0)->field)`，通过该宏定义获取到结构体ProcArrayStruct第一个成员到最后一个成员之前的内存大小。这时ProcArrayStruct的大小为`offsetof(ProcArrayStruct, pgprocnos) + (MaxBackends + max_prepared_xacts) * sizeof(int)`。  
在Hot Standby过程需要KnowAssignedXids数据结构，其需求的内存大小为`sizeof(TransactionId) * ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS) + sizeof(bool) * ((PGPROC_MAX_CACHED_SUBXIDS + 1) * PROCARRAY_MAXPROCS)`。

postmaster启动的时候初始化共享PGPROC array

static TransactionId *KnownAssignedXids;
    static bool *KnownAssignedXidsValid;
    void CreateSharedProcArray(void) { 
    	bool		found;
    	procArray = (ProcArrayStruct *)ShmemInitStruct("Proc Array",add_size(offsetof(ProcArrayStruct, pgprocnos),mul_size(sizeof(int),PROCARRAY_MAXPROCS)),&found);
    	if (!found){ 
    		/* We're the first - initialize. */
    		procArray->numProcs = 0;
    		procArray->maxProcs = PROCARRAY_MAXPROCS;
    		procArray->maxKnownAssignedXids = TOTAL_MAX_CACHED_SUBXIDS;
    		procArray->numKnownAssignedXids = 0;
    		procArray->tailKnownAssignedXids = 0;
    		procArray->headKnownAssignedXids = 0;
    		SpinLockInit(&procArray->known_assigned_xids_lck);
    		procArray->lastOverflowedXid = InvalidTransactionId;
    		procArray->replication_slot_xmin = InvalidTransactionId;
    		procArray->replication_slot_catalog_xmin = InvalidTransactionId;
    	}
    	allProcs = ProcGlobal->allProcs;
    	allPgXact = ProcGlobal->allPgXact;
    	/* Create or attach to the KnownAssignedXids arrays too, if needed */
    	if (EnableHotStandby) { 
    		KnownAssignedXids = (TransactionId *)ShmemInitStruct("KnownAssignedXids",mul_size(sizeof(TransactionId),TOTAL_MAX_CACHED_SUBXIDS),&found);
    		KnownAssignedXidsValid = (bool *)ShmemInitStruct("KnownAssignedXidsValid",mul_size(sizeof(bool), TOTAL_MAX_CACHED_SUBXIDS),&found);
    	}
    	/* Register and initialize fields of ProcLWLockTranche */
    	LWLockRegisterTranche(LWTRANCHE_PROC, "proc");
    }

### ProcArray操作 ###

ProcArrayAdd函数向共享array中加入指定的PGPROC，需要先将ProcArray使用PGPROC\*进行排序。

void ProcArrayAdd(PGPROC *proc) { 
    	ProcArrayStruct *arrayP = procArray;
    	int			index;
    	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
    	if (arrayP->numProcs >= arrayP->maxProcs) { 
    		/* Oops, no room. */
    		LWLockRelease(ProcArrayLock);
    		ereport(FATAL,(errcode(ERRCODE_TOO_MANY_CONNECTIONS),errmsg("sorry, too many clients already")));
    	}
    	/* Keep the procs array sorted by (PGPROC *) so that we can utilize locality of references much better. This is useful while traversing the ProcArray because there is an increased likelihood of finding the next PGPROC structure in the cache. * Since the occurrence of adding/removing a proc is much lower than the access to the ProcArray itself, the overhead should be marginal */
    	for (index = 0; index < arrayP->numProcs; index++)
    	{ 
    		/* If we are the first PGPROC or if we have found our right position in the array, break */
    		if ((arrayP->pgprocnos[index] == -1) || (arrayP->pgprocnos[index] > proc->pgprocno))
    			break;
    	}
    	memmove(&arrayP->pgprocnos[index + 1], &arrayP->pgprocnos[index],(arrayP->numProcs - index) * sizeof(int));
    	arrayP->pgprocnos[index] = proc->pgprocno;
    	arrayP->numProcs++;
    	LWLockRelease(ProcArrayLock);
    }

### 添加操作调用 ###

InitPostgres(postinit.c) --> InitProcessPhase2使MyProc在共享ProcArray可见 --> ProcArrayAdd(MyProc)

void InitProcessPhase2(void) { 
    	Assert(MyProc != NULL);
    	/* Add our PGPROC to the PGPROC array in shared memory. */
    	ProcArrayAdd(MyProc);
    	/* Arrange to clean that up at backend exit. */
    	on_shmem_exit(RemoveProcFromArray, 0);
    }

EndPrepare(twophase.c) --> MarkAsPrepared(gxact, false)  
RecoverPreparedTransactions(twophase.c) --> MarkAsPrepared(gxact, true)  
MarkAsPrepared标记GXACT为fully 有效加入共享ProcArray --> ProcArrayAdd(&ProcGlobal->allProcs\[gxact->pgprocno\])

static void MarkAsPrepared(GlobalTransaction gxact, bool lock_held) { 
    	/* Lock here may be overkill, but I'm not convinced of that ... */
    	if (!lock_held) LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);
    	Assert(!gxact->valid);
    	gxact->valid = true;
    	if (!lock_held) LWLockRelease(TwoPhaseStateLock);
    	/* Put it into the global ProcArray so TransactionIdIsInProgress considers the XID as still running. */
    	ProcArrayAdd(&ProcGlobal->allProcs[gxact->pgprocno]);
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6IKl5Y-U6I-M_size_20_color_FFFFFF_t_70_g_se_x_16_pic_center]: /images/20220829/4cb5ce9c093d4a96876d68dffe17c7a6.png