From: Kyotaro Horiguchi Date: Mon, 23 Mar 2020 15:50:33 +0900 Subject: [PATCH v36 4/4] Propagage pending sync information to parallel workers Parallel workers needs to know about pending syncs since it performs some WAL-related operations. This patch reconstruct pending sync hash on workers then set the correct newness flag on relcache creation. --- src/backend/catalog/storage.c | 141 ++++++++++++++++++++++++++-- src/backend/executor/execParallel.c | 28 ++++++ src/backend/utils/cache/relcache.c | 17 ++++ src/include/catalog/storage.h | 2 + 4 files changed, 178 insertions(+), 10 deletions(-) diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c index 56be8a2d52..5bbd0f5c3c 100644 --- a/src/backend/catalog/storage.c +++ b/src/backend/catalog/storage.c @@ -19,6 +19,7 @@ #include "postgres.h" +#include "access/parallel.h" #include "access/visibilitymap.h" #include "access/xact.h" #include "access/xlog.h" @@ -74,6 +75,26 @@ typedef struct PendingRelSync static PendingRelDelete *pendingDeletes = NULL; /* head of linked list */ HTAB *pendingSyncHash = NULL; + +/* + * create_pendingsync_hash - helper function to create pending sync hash + */ +static void +create_pendingsync_hash(void) +{ + HASHCTL ctl; + + Assert(pendingSyncHash == NULL); + + ctl.keysize = sizeof(RelFileNode); + ctl.entrysize = sizeof(PendingRelSync); + ctl.hcxt = TopTransactionContext; + pendingSyncHash = + hash_create("pending sync hash", + 16, &ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); +} + + /* * RelationCreateStorage * Create physical storage for a relation. @@ -137,17 +158,12 @@ RelationCreateStorage(RelFileNode rnode, char relpersistence) /* we sync only permanent relations */ Assert(backend == InvalidBackendId); + /* we don't assume create storage on parallel workers*/ + Assert(!IsParallelWorker()); + + /* create the hash if not yet */ if (!pendingSyncHash) - { - HASHCTL ctl; - - ctl.keysize = sizeof(RelFileNode); - ctl.entrysize = sizeof(PendingRelSync); - ctl.hcxt = TopTransactionContext; - pendingSyncHash = - hash_create("pending sync hash", - 16, &ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); - } + create_pendingsync_hash(); pending = hash_search(pendingSyncHash, &rnode, HASH_ENTER, &found); Assert(!found); @@ -493,6 +509,111 @@ RelFileNodeSkippingWAL(RelFileNode rnode) return true; } +/* + * SerializeSkippingWALRelFileNods + * + * RelationNeedsWAL and RelFileNodeSkippingWAL must offer the correct answer to + * parallel workers. This function is used to serialize RelFileNodes of pending + * syncs so that workers know about pending syncs. Since workers are not + * assumed to create or truncate of relfilenodes or perform transactional + * operations, only rnodes are required. + */ +int +SerializePendingSyncs(RelFileNode **parray) +{ + HTAB *tmphash; + HASHCTL ctl; + HASH_SEQ_STATUS scan; + PendingRelSync *sync; + PendingRelDelete *delete; + RelFileNode *src; + RelFileNode *dest; + int nrnodes = 0; + + /* Don't call from parallel workers */ + Assert(!IsParallelWorker()); + + if (XLogIsNeeded() || !pendingSyncHash) + return 0; /* No pending syncs */ + + /* Create temporary hash to collect active relfilenodes */ + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(RelFileNode); + ctl.entrysize = sizeof(RelFileNode); + ctl.hcxt = CurrentMemoryContext; + tmphash = hash_create("tmp relfilenodes", 16, &ctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + + /* collect all rnodes from pending syncs */ + hash_seq_init(&scan, pendingSyncHash); + while ((sync = (PendingRelSync *) hash_seq_search(&scan))) + { + (void) hash_search(tmphash, &sync->rnode, HASH_ENTER, NULL); + nrnodes++; + } + + /* remove deleted rnodes */ + for (delete = pendingDeletes; delete != NULL; delete = delete->next) + { + if (delete->atCommit) + { + bool found; + + (void) hash_search(tmphash, (void *) &delete->relnode, + HASH_REMOVE, &found); + if (found) + nrnodes--; + } + } + + Assert(nrnodes >= 0); + + if (nrnodes == 0) + return 0; + + /* Create and fill the array. It contains terminating (0,0,0) node */ + *parray = palloc(sizeof(RelFileNode) * (nrnodes + 1)); + dest = &(*parray)[0]; + + hash_seq_init(&scan, tmphash); + while ((src = (RelFileNode *) hash_seq_search(&scan))) + *dest++ = *src; + + hash_destroy(tmphash); + + /* set terminator */ + MemSet(dest, 0, sizeof(RelFileNode)); + + return (nrnodes + 1) * sizeof(RelFileNode); +} + +/* + * SerializeSkippingWALRelFileNods + */ +void +DeserializePendingSyncs(RelFileNode *array) +{ + RelFileNode *rnode; + + Assert(pendingSyncHash == NULL); + Assert(IsParallelWorker()); + + create_pendingsync_hash(); + + for (rnode = array ; rnode->relNode != 0 ; rnode++) + { + PendingRelSync *pending; + + Assert(rnode->spcNode != 0 && rnode->dbNode != 0); + + pending = hash_search(pendingSyncHash, rnode, HASH_ENTER, NULL); + + pending->is_truncated = false; /* not cared */ + } + + return; +} + /* * smgrDoPendingDeletes() -- Take care of relation deletes at end of xact. * diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index a753d6efa0..5964c02b8a 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -23,6 +23,7 @@ #include "postgres.h" +#include "catalog/storage.h" #include "executor/execParallel.h" #include "executor/executor.h" #include "executor/nodeAppend.h" @@ -62,6 +63,7 @@ #define PARALLEL_KEY_DSA UINT64CONST(0xE000000000000007) #define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xE000000000000008) #define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009) +#define PARALLEL_KEY_PENDING_SYNCS UINT64CONST(0xE00000000000000A) #define PARALLEL_TUPLE_QUEUE_SIZE 65536 @@ -583,6 +585,9 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, Size dsa_minsize = dsa_minimum_size(); char *query_string; int query_len; + int pendingsync_list_size = 0; + RelFileNode *pendingsync_buf; + char *pendingsync_space; /* * Force any initplan outputs that we're going to pass to workers to be @@ -684,6 +689,14 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, } } + /* Estimate space pending-sync relfilenodes */ + pendingsync_list_size = SerializePendingSyncs(&pendingsync_buf); + if (pendingsync_list_size > 0) + { + shm_toc_estimate_chunk(&pcxt->estimator, pendingsync_list_size); + shm_toc_estimate_keys(&pcxt->estimator, 1); + } + /* Estimate space for DSA area. */ shm_toc_estimate_chunk(&pcxt->estimator, dsa_minsize); shm_toc_estimate_keys(&pcxt->estimator, 1); @@ -769,6 +782,15 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, } } + /* Copy pending sync list if any */ + if (pendingsync_list_size > 0) + { + pendingsync_space = shm_toc_allocate(pcxt->toc, pendingsync_list_size); + memcpy(pendingsync_space, pendingsync_buf, pendingsync_list_size); + shm_toc_insert(pcxt->toc, PARALLEL_KEY_PENDING_SYNCS, + pendingsync_space); + } + /* * Create a DSA area that can be used by the leader and all workers. * (However, if we failed to create a DSM and are using private memory @@ -1332,6 +1354,7 @@ void ParallelQueryMain(dsm_segment *seg, shm_toc *toc) { FixedParallelExecutorState *fpes; + RelFileNode *pendingsyncs; BufferUsage *buffer_usage; DestReceiver *receiver; QueryDesc *queryDesc; @@ -1345,6 +1368,11 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) /* Get fixed-size state. */ fpes = shm_toc_lookup(toc, PARALLEL_KEY_EXECUTOR_FIXED, false); + /* Get pending-sync information */ + pendingsyncs = shm_toc_lookup(toc, PARALLEL_KEY_PENDING_SYNCS, true); + if (pendingsyncs) + DeserializePendingSyncs(pendingsyncs); + /* Set up DestReceiver, SharedExecutorInstrumentation, and QueryDesc. */ receiver = ExecParallelGetReceiver(seg, toc); instrumentation = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, true); diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 9ee9dc8cc0..fd96b0593a 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -33,6 +33,7 @@ #include "access/htup_details.h" #include "access/multixact.h" #include "access/nbtree.h" +#include "access/parallel.h" #include "access/reloptions.h" #include "access/sysattr.h" #include "access/table.h" @@ -1241,6 +1242,22 @@ RelationBuildDesc(Oid targetRelId, bool insertIt) if (insertIt) RelationCacheInsert(relation, true); + /* + * Restore new-ness flags on parallel workers. + * + * Parallel workers needs WAL-skipping information. Fill in the flag using + * pending sync information. It is restored at worker start. + * + * We assume that parallel worker doesn't perform transactional operations, + * so just set rd_firstRelfilenodeSubid to 1 for relations with new + * relfilenodes. + */ + if (IsParallelWorker()) + { + if (RelFileNodeSkippingWAL(relation->rd_node)) + relation->rd_firstRelfilenodeSubid = 1; + } + /* It's fully valid */ relation->rd_isvalid = true; diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h index bd37bf311c..37d96729c2 100644 --- a/src/include/catalog/storage.h +++ b/src/include/catalog/storage.h @@ -30,6 +30,8 @@ extern void RelationTruncate(Relation rel, BlockNumber nblocks); extern void RelationCopyStorage(SMgrRelation src, SMgrRelation dst, ForkNumber forkNum, char relpersistence); extern bool RelFileNodeSkippingWAL(RelFileNode rnode); +extern int SerializePendingSyncs(RelFileNode **parray); +extern void DeserializePendingSyncs(RelFileNode *parray); /* * These functions used to be in storage/smgr/smgr.c, which explains the -- 2.18.2 ----Next_Part(Mon_Mar_23_17_20_27_2020_544)----