diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index f9c8598..7d88489 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -34,6 +34,7 @@ #include "optimizer/planner.h" #include "storage/spin.h" #include "tcop/tcopprot.h" +#include "utils/dsa.h" #include "utils/memutils.h" #include "utils/snapmgr.h" @@ -47,6 +48,7 @@ #define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xE000000000000003) #define PARALLEL_KEY_TUPLE_QUEUE UINT64CONST(0xE000000000000004) #define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xE000000000000005) +#define PARALLEL_KEY_AREA UINT64CONST(0xE000000000000006) #define PARALLEL_TUPLE_QUEUE_SIZE 65536 @@ -345,6 +347,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers) int param_len; int instrumentation_len = 0; int instrument_offset = 0; + char *area_space; /* Allocate object for return value. */ pei = palloc0(sizeof(ParallelExecutorInfo)); @@ -413,6 +416,10 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers) shm_toc_estimate_keys(&pcxt->estimator, 1); } + /* Estimate space for DSA area. */ + shm_toc_estimate_chunk(&pcxt->estimator, dsa_minimum_size()); + shm_toc_estimate_keys(&pcxt->estimator, 1); + /* Everyone's had a chance to ask for space, so now create the DSM. */ InitializeParallelDSM(pcxt); @@ -466,6 +473,20 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers) pei->instrumentation = instrumentation; } + /* Create a DSA area that can be used by the leader and all workers. */ + area_space = shm_toc_allocate(pcxt->toc, dsa_minimum_size()); + shm_toc_insert(pcxt->toc, PARALLEL_KEY_AREA, area_space); + pei->area = dsa_create_in_place(area_space, dsa_minimum_size(), + LWTRANCHE_PARALLEL_EXEC_AREA, + "parallel query memory area", + pcxt->seg); + + /* + * Make the area available to executor nodes running in the leader. See + * also ParallelQueryMain which makes it available to workers. + */ + estate->es_query_area = pei->area; + /* * Give parallel-aware nodes a chance to initialize their shared data. * This also initializes the elements of instrumentation->ps_instrument, @@ -571,6 +592,11 @@ ExecParallelFinish(ParallelExecutorInfo *pei) void ExecParallelCleanup(ParallelExecutorInfo *pei) { + if (pei->area != NULL) + { + dsa_detach(pei->area); + pei->area = NULL; + } if (pei->pcxt != NULL) { DestroyParallelContext(pei->pcxt); @@ -728,6 +754,8 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) QueryDesc *queryDesc; SharedExecutorInstrumentation *instrumentation; int instrument_options = 0; + void *area_space; + dsa_area *area; /* Set up DestReceiver, SharedExecutorInstrumentation, and QueryDesc. */ receiver = ExecParallelGetReceiver(seg, toc); @@ -739,8 +767,13 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) /* Prepare to track buffer usage during query execution. */ InstrStartParallelQuery(); + /* Attach to the dynamic shared memory area. */ + area_space = shm_toc_lookup(toc, PARALLEL_KEY_AREA); + area = dsa_attach_in_place(area_space, seg); + /* Start up the executor, have it run the plan, and then shut it down. */ ExecutorStart(queryDesc, 0); + queryDesc->planstate->state->es_query_area = area; ExecParallelInitializeWorker(queryDesc->planstate, toc); ExecutorRun(queryDesc, ForwardScanDirection, 0L); ExecutorFinish(queryDesc); @@ -758,6 +791,7 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) ExecutorEnd(queryDesc); /* Cleanup. */ + dsa_detach(area); FreeQueryDesc(queryDesc); (*receiver->rDestroy) (receiver); } diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h index f4c6d37..4b57474 100644 --- a/src/include/executor/execParallel.h +++ b/src/include/executor/execParallel.h @@ -17,6 +17,7 @@ #include "nodes/execnodes.h" #include "nodes/parsenodes.h" #include "nodes/plannodes.h" +#include "utils/dsa.h" typedef struct SharedExecutorInstrumentation SharedExecutorInstrumentation; @@ -27,6 +28,7 @@ typedef struct ParallelExecutorInfo BufferUsage *buffer_usage; SharedExecutorInstrumentation *instrumentation; shm_mq_handle **tqueue; + dsa_area *area; bool finished; } ParallelExecutorInfo; diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 8004d85..0837c42 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -20,6 +20,7 @@ #include "lib/pairingheap.h" #include "nodes/params.h" #include "nodes/plannodes.h" +#include "utils/dsa.h" #include "utils/hsearch.h" #include "utils/reltrigger.h" #include "utils/sortsupport.h" @@ -422,6 +423,9 @@ typedef struct EState HeapTuple *es_epqTuple; /* array of EPQ substitute tuples */ bool *es_epqTupleSet; /* true if EPQ tuple is provided */ bool *es_epqScanDone; /* true if EPQ tuple has been fetched */ + + /* The per-query shared memory area to use for parallel execution. */ + dsa_area *es_query_area; } EState; diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h index 9a2d869..951e421 100644 --- a/src/include/storage/lwlock.h +++ b/src/include/storage/lwlock.h @@ -235,6 +235,7 @@ typedef enum BuiltinTrancheIds LWTRANCHE_BUFFER_MAPPING, LWTRANCHE_LOCK_MANAGER, LWTRANCHE_PREDICATE_LOCK_MANAGER, + LWTRANCHE_PARALLEL_EXEC_AREA, LWTRANCHE_FIRST_USER_DEFINED } BuiltinTrancheIds;