agora inbox for [email protected]
help / color / mirror / Atom feedFrom: Kyotaro Horiguchi <[email protected]>
Subject: [PATCH v36 4/4] Propagage pending sync information to parallel workers
Date: Mon, 23 Mar 2020 15:50:33 +0900
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)----
view thread (2+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected]
Subject: Re: [PATCH v36 4/4] Propagage pending sync information to parallel workers
In-Reply-To: <no-message-id-199323@localhost>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox