test/he3pg
1
0
Fork 0
mirror of https://gitee.com/he3db/he3pg.git synced 2024-11-29 18:58:35 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

!183 parallel replay panic

Browse Source 
Merge pull request !183 from shenzhengntu/dev_performance
This commit is contained in:
shipixian 2023-06-28 02:33:47 +00:00 committed by Gitee
commit f86c2a823f
No known key found for this signature in database
GPG Key ID: 173E9B9CA92EEF8F
4 changed files with 50 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
initdata/masterconf/postgresql.conf
View File

@ -93,7 +93,7 @@ max_connections = 100 # (change requires restart)
# - Authentication - # - Authentication -
#authentication_timeout = 1min # 1s-600s #authentication_timeout = 1min # 1s-600s
#password_encryption = scram-sha-256 # scram-sha-256 or md5
#db_user_namespace = off #db_user_namespace = off
# GSSAPI using Kerberos # GSSAPI using Kerberos

2
initdata/pushconf/postgresql.conf
View File

@ -93,7 +93,7 @@ max_connections = 100 # (change requires restart)
# - Authentication - # - Authentication -
#authentication_timeout = 1min # 1s-600s #authentication_timeout = 1min # 1s-600s
password_encryption = md5 # scram-sha-256 or md5
#db_user_namespace = off #db_user_namespace = off
# GSSAPI using Kerberos # GSSAPI using Kerberos

88
src/backend/access/transam/pagehashqueue.c
View File

@ -470,24 +470,29 @@ static HTAB *PageCountHash = NULL;
static uint32_t curLatchPos = 0; static uint32_t curLatchPos = 0;
typedef struct { typedef struct {
pg_atomic_uint32 gpushpos; slock_t mutex;
pg_atomic_uint32 ready; volatile uint32 gpushpos;
pg_atomic_uint32 gpos; volatile bool ready;
volatile uint32 gpos;
pg_atomic_uint32 latchPos; pg_atomic_uint32 latchPos;
pg_atomic_uint32 taskNum; pg_atomic_uint32 taskNum;
pg_atomic_uint32 modifyNum; uint32 modifyNum;
Latch pageFlushWakeupLatch[PARALLEL_NUM]; Latch pageFlushWakeupLatch[PARALLEL_NUM];
PageValue*gtag[G_QUEUE_LEN]; PageValue*gtag[G_QUEUE_LEN];
}PageHashQueueShmemStruct; }PageHashQueueShmemStruct;
static PageHashQueueShmemStruct *PageHashQueueShmem; static PageHashQueueShmemStruct *PageHashQueueShmem;
void pushSlaveReplayQueue(int pageNum) { void pushSlaveReplayQueue(int pageNum) {
if (pg_atomic_read_u32(&PageHashQueueShmem->gpos) != 0 && pg_atomic_read_u32(&PageHashQueueShmem->ready) == 0) {
pg_atomic_exchange_u32(&PageHashQueueShmem->ready,1); if (PageHashQueueShmem->gpos != 0 && PageHashQueueShmem->ready == false) {
SpinLockAcquire(&PageHashQueueShmem->mutex);
PageHashQueueShmem->ready = true;
SpinLockRelease(&PageHashQueueShmem->mutex);
WakeupFlushWork(); WakeupFlushWork();
} }
while(pageNum > CompletedTaskNum()) { while(pageNum > CompletedTaskNum()) {
usleep(10); pg_usleep(1000L);
} }
cleanMap(); cleanMap();
} }
@ -547,12 +552,15 @@ PageHashQueueShmemInit(void)
if (!found) if (!found)
{ {
pg_atomic_init_u32(&PageHashQueueShmem->ready,0); SpinLockInit(&PageHashQueueShmem->mutex);
pg_atomic_init_u32(&PageHashQueueShmem->gpushpos,0); SpinLockAcquire(&PageHashQueueShmem->mutex);
pg_atomic_init_u32(&PageHashQueueShmem->latchPos,0); PageHashQueueShmem->ready = false;
PageHashQueueShmem->gpushpos = 0;
SpinLockRelease(&PageHashQueueShmem->mutex);
PageHashQueueShmem->gpos = 0;
pg_atomic_init_u32(&PageHashQueueShmem->taskNum,0); pg_atomic_init_u32(&PageHashQueueShmem->taskNum,0);
pg_atomic_init_u32(&PageHashQueueShmem->gpos, 0); pg_atomic_init_u32(&PageHashQueueShmem->latchPos, 0);
pg_atomic_init_u32(&PageHashQueueShmem->modifyNum,0); PageHashQueueShmem->modifyNum = 0;
for (int i = 0;i<PARALLEL_NUM;i++) { for (int i = 0;i<PARALLEL_NUM;i++) {
InitSharedLatch(&PageHashQueueShmem->pageFlushWakeupLatch[i]); InitSharedLatch(&PageHashQueueShmem->pageFlushWakeupLatch[i]);
} }
@ -567,7 +575,6 @@ Size PageHashMapSize(void) {
return RedoStartPointSize() + hash_estimate_size(G_QUEUE_LEN,sizeof(PageValue)); return RedoStartPointSize() + hash_estimate_size(G_QUEUE_LEN,sizeof(PageValue));
} }
void void
InitBufferPoolHashMap(void) InitBufferPoolHashMap(void)
{ {
@ -605,11 +612,12 @@ uint32_t addFileKey(BufferTag*onePage) {
&found); &found);
if (found == false) { if (found == false) {
result->num = 0; result->num = 0;
uint32_t gpos = pg_atomic_fetch_add_u32(&PageHashQueueShmem->gpos,1); uint32_t gpos = PageHashQueueShmem->gpos++;
PageHashQueueShmem->gtag[gpos] = result; PageHashQueueShmem->gtag[gpos] = result;
} }
result->num++; result->num++;
return pg_atomic_fetch_add_u32(&PageHashQueueShmem->modifyNum,1); PageHashQueueShmem->modifyNum++;
return PageHashQueueShmem->modifyNum;
} }
void cleanMap(void) { void cleanMap(void) {
@ -624,11 +632,13 @@ void cleanMap(void) {
HASH_REMOVE, NULL) == NULL) HASH_REMOVE, NULL) == NULL)
elog(ERROR, "hash table corrupted"); elog(ERROR, "hash table corrupted");
} }
pg_atomic_init_u32(&PageHashQueueShmem->ready,0); SpinLockAcquire(&PageHashQueueShmem->mutex);
pg_atomic_init_u32(&PageHashQueueShmem->gpos, 0); PageHashQueueShmem->ready = false;
pg_atomic_init_u32(&PageHashQueueShmem->gpushpos,0); PageHashQueueShmem->gpushpos = 0;
SpinLockRelease(&PageHashQueueShmem->mutex);
PageHashQueueShmem->gpos = 0;
pg_atomic_init_u32(&PageHashQueueShmem->taskNum,0); pg_atomic_init_u32(&PageHashQueueShmem->taskNum,0);
pg_atomic_init_u32(&PageHashQueueShmem->modifyNum,0); PageHashQueueShmem->modifyNum = 0;
} }
uint32_t hashMapSize(void) { uint32_t hashMapSize(void) {
@ -640,35 +650,31 @@ static int cmp(const void* a,const void* b) {
} }
void SortPageQueue(void) { void SortPageQueue(void) {
if (pg_atomic_read_u32(&PageHashQueueShmem->gpos) != 0 && pg_atomic_read_u32(&PageHashQueueShmem->ready) == 0) { if (PageHashQueueShmem->gpos != 0 && PageHashQueueShmem->ready == false) {
qsort(PageHashQueueShmem->gtag,pg_atomic_read_u32(&PageHashQueueShmem->gpos),sizeof(PageValue*),cmp); qsort(PageHashQueueShmem->gtag,PageHashQueueShmem->gpos,sizeof(PageValue*),cmp);
pg_atomic_exchange_u32(&PageHashQueueShmem->ready,1); SpinLockAcquire(&PageHashQueueShmem->mutex);
PageHashQueueShmem->ready = true;
SpinLockRelease(&PageHashQueueShmem->mutex);
WakeupFlushWork(); WakeupFlushWork();
return; return;
} }
} }
BufferTag* QueuePushPage(void) { BufferTag* QueuePushPage(void) {
uint32_t gpushpos = pg_atomic_read_u32(&PageHashQueueShmem->gpushpos); uint32_t gpushpos;
uint32_t gpos = pg_atomic_read_u32(&PageHashQueueShmem->gpos); bool hasData = false;
uint32_t ready = pg_atomic_read_u32(&PageHashQueueShmem->ready); if (PageHashQueueShmem->ready == true) {
if (ready == 0 || SpinLockAcquire(&PageHashQueueShmem->mutex);
gpushpos >= gpos){ if (PageHashQueueShmem->ready == true && PageHashQueueShmem->gpushpos < PageHashQueueShmem->gpos) {
return NULL; hasData = true;
} gpushpos = PageHashQueueShmem->gpushpos++;
gpushpos = pg_atomic_fetch_add_u32(&PageHashQueueShmem->gpushpos,1);
uint32_t ReGpos = pg_atomic_read_u32(&PageHashQueueShmem->gpos);
ready = pg_atomic_read_u32(&PageHashQueueShmem->ready);
/* only ready 1 to 0 then 0 to 1 will produce a serious problem,
reason functions of pushTikv trigger the next batch of data processing
cost time less than (pg_atomic_read_u32 and pg_atomic_fetch_add_u32) */
if (ready!= 0 && gpushpos < gpos) {
return &(PageHashQueueShmem->gtag[gpushpos]->tag);
} else {
if (ReGpos!=0 && gpushpos < gpos) {
elog(PANIC,"QueuePushPage gpushpos %d < gpos %d",gpushpos,gpos);
} }
SpinLockRelease(&PageHashQueueShmem->mutex);
}
if (hasData == false) {
return NULL; return NULL;
} else {
return &(PageHashQueueShmem->gtag[gpushpos]->tag);
} }
} }

2
src/backend/utils/misc/postgresql.conf.sample
View File

@ -93,7 +93,7 @@
# - Authentication - # - Authentication -
#authentication_timeout = 1min # 1s-600s #authentication_timeout = 1min # 1s-600s
#password_encryption = scram-sha-256 # scram-sha-256 or md5
#db_user_namespace = off #db_user_namespace = off
# GSSAPI using Kerberos # GSSAPI using Kerberos