From: Robert Haas Date: Fri, 16 Oct 2015 15:56:02 +0000 (-0400) Subject: Rewrite interaction of parallel mode with parallel executor support. X-Git-Tag: REL9_6_BETA1~1198 X-Git-Url: https://granicus.if.org/sourcecode?a=commitdiff_plain;h=bfc78d7196eb28cd4e3d6c24f7e607bacecf1129;p=postgresql Rewrite interaction of parallel mode with parallel executor support. In the previous coding, before returning from ExecutorRun, we'd shut down all parallel workers. This was dead wrong if ExecutorRun was called with a non-zero tuple count; it had the effect of truncating the query output. To fix, give ExecutePlan control over whether to enter parallel mode, and have it refuse to do so if the tuple count is non-zero. Rewrite the Gather logic so that it can cope with being called outside parallel mode. Commit 7aea8e4f2daa4b39ca9d1309a0c4aadb0f7ed81b is largely to blame for this problem, though this patch modifies some subsequently-committed code which relied on the guarantees it purported to make. --- diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 37b7bbd413..a55022e0a8 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -76,6 +76,7 @@ static void CheckValidRowMarkRel(Relation rel, RowMarkType markType); static void ExecPostprocessPlan(EState *estate); static void ExecEndPlan(PlanState *planstate, EState *estate); static void ExecutePlan(EState *estate, PlanState *planstate, + bool use_parallel_mode, CmdType operation, bool sendTuples, long numberTuples, @@ -243,11 +244,6 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags) if (!(eflags & (EXEC_FLAG_SKIP_TRIGGERS | EXEC_FLAG_EXPLAIN_ONLY))) AfterTriggerBeginQuery(); - /* Enter parallel mode, if required by the query. */ - if (queryDesc->plannedstmt->parallelModeNeeded && - !(eflags & EXEC_FLAG_EXPLAIN_ONLY)) - EnterParallelMode(); - MemoryContextSwitchTo(oldcontext); } @@ -341,15 +337,13 @@ standard_ExecutorRun(QueryDesc *queryDesc, if (!ScanDirectionIsNoMovement(direction)) ExecutePlan(estate, queryDesc->planstate, + queryDesc->plannedstmt->parallelModeNeeded, operation, sendTuples, count, direction, dest); - /* Allow nodes to release or shut down resources. */ - (void) ExecShutdownNode(queryDesc->planstate); - /* * shutdown tuple receiver, if we started it */ @@ -482,11 +476,6 @@ standard_ExecutorEnd(QueryDesc *queryDesc) */ MemoryContextSwitchTo(oldcontext); - /* Exit parallel mode, if it was required by the query. */ - if (queryDesc->plannedstmt->parallelModeNeeded && - !(estate->es_top_eflags & EXEC_FLAG_EXPLAIN_ONLY)) - ExitParallelMode(); - /* * Release EState and per-query memory context. This should release * everything the executor has allocated. @@ -1529,6 +1518,7 @@ ExecEndPlan(PlanState *planstate, EState *estate) static void ExecutePlan(EState *estate, PlanState *planstate, + bool use_parallel_mode, CmdType operation, bool sendTuples, long numberTuples, @@ -1548,6 +1538,20 @@ ExecutePlan(EState *estate, */ estate->es_direction = direction; + /* + * If a tuple count was supplied, we must force the plan to run without + * parallelism, because we might exit early. + */ + if (numberTuples != 0) + use_parallel_mode = false; + + /* + * If a tuple count was supplied, we must force the plan to run without + * parallelism, because we might exit early. + */ + if (use_parallel_mode) + EnterParallelMode(); + /* * Loop until we've processed the proper number of tuples from the plan. */ @@ -1566,7 +1570,11 @@ ExecutePlan(EState *estate, * process so we just end the loop... */ if (TupIsNull(slot)) + { + /* Allow nodes to release or shut down resources. */ + (void) ExecShutdownNode(planstate); break; + } /* * If we have a junk filter, then project a new tuple with the junk @@ -1603,6 +1611,9 @@ ExecutePlan(EState *estate, if (numberTuples && numberTuples == current_tuple_count) break; } + + if (use_parallel_mode) + ExitParallelMode(); } diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index e6930c1d51..3bb820692d 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -442,6 +442,23 @@ ExecParallelFinish(ParallelExecutorInfo *pei) pei->instrumentation); } +/* + * Clean up whatever ParallelExecutreInfo resources still exist after + * ExecParallelFinish. We separate these routines because someone might + * want to examine the contents of the DSM after ExecParallelFinish and + * before calling this routine. + */ +void +ExecParallelCleanup(ParallelExecutorInfo *pei) +{ + if (pei->pcxt != NULL) + { + DestroyParallelContext(pei->pcxt); + pei->pcxt = NULL; + } + pfree(pei); +} + /* * Create a DestReceiver to write tuples we produce to the shm_mq designated * for that purpose. diff --git a/src/backend/executor/nodeGather.c b/src/backend/executor/nodeGather.c index c689a4d17a..7e2272f634 100644 --- a/src/backend/executor/nodeGather.c +++ b/src/backend/executor/nodeGather.c @@ -16,6 +16,7 @@ #include "postgres.h" #include "access/relscan.h" +#include "access/xact.h" #include "executor/execdebug.h" #include "executor/execParallel.h" #include "executor/nodeGather.h" @@ -45,7 +46,6 @@ ExecInitGather(Gather *node, EState *estate, int eflags) gatherstate = makeNode(GatherState); gatherstate->ps.plan = (Plan *) node; gatherstate->ps.state = estate; - gatherstate->need_to_scan_workers = false; gatherstate->need_to_scan_locally = !node->single_copy; /* @@ -106,52 +106,57 @@ ExecGather(GatherState *node) * needs to allocate large dynamic segement, so it is better to do if it * is really needed. */ - if (!node->pei) + if (!node->initialized) { EState *estate = node->ps.state; - - /* Initialize the workers required to execute Gather node. */ - node->pei = ExecInitParallelPlan(node->ps.lefttree, - estate, - ((Gather *) (node->ps.plan))->num_workers); + Gather *gather = (Gather *) node->ps.plan; /* - * Register backend workers. If the required number of workers are not - * available then we perform the scan with available workers and if - * there are no more workers available, then the Gather node will just - * scan locally. + * Sometimes we might have to run without parallelism; but if + * parallel mode is active then we can try to fire up some workers. */ - LaunchParallelWorkers(node->pei->pcxt); - - node->funnel = CreateTupleQueueFunnel(); - - for (i = 0; i < node->pei->pcxt->nworkers; ++i) + if (gather->num_workers > 0 && IsInParallelMode()) { - if (node->pei->pcxt->worker[i].bgwhandle) + bool got_any_worker = false; + + /* Initialize the workers required to execute Gather node. */ + node->pei = ExecInitParallelPlan(node->ps.lefttree, + estate, + gather->num_workers); + + /* + * Register backend workers. We might not get as many as we + * requested, or indeed any at all. + */ + LaunchParallelWorkers(node->pei->pcxt); + + /* Set up a tuple queue to collect the results. */ + node->funnel = CreateTupleQueueFunnel(); + for (i = 0; i < node->pei->pcxt->nworkers; ++i) { - shm_mq_set_handle(node->pei->tqueue[i], - node->pei->pcxt->worker[i].bgwhandle); - RegisterTupleQueueOnFunnel(node->funnel, node->pei->tqueue[i]); - node->need_to_scan_workers = true; + if (node->pei->pcxt->worker[i].bgwhandle) + { + shm_mq_set_handle(node->pei->tqueue[i], + node->pei->pcxt->worker[i].bgwhandle); + RegisterTupleQueueOnFunnel(node->funnel, + node->pei->tqueue[i]); + got_any_worker = true; + } } + + /* No workers? Then never mind. */ + if (!got_any_worker) + ExecShutdownGather(node); } - /* If no workers are available, we must always scan locally. */ - if (!node->need_to_scan_workers) - node->need_to_scan_locally = true; + /* Run plan locally if no workers or not single-copy. */ + node->need_to_scan_locally = (node->funnel == NULL) + || !gather->single_copy; + node->initialized = true; } slot = gather_getnext(node); - if (TupIsNull(slot)) - { - /* - * Destroy the parallel context once we complete fetching all the - * tuples. Otherwise, the DSM and workers will stick around for the - * lifetime of the entire statement. - */ - ExecShutdownGather(node); - } return slot; } @@ -194,10 +199,9 @@ gather_getnext(GatherState *gatherstate) */ slot = gatherstate->ps.ps_ProjInfo->pi_slot; - while (gatherstate->need_to_scan_workers || - gatherstate->need_to_scan_locally) + while (gatherstate->funnel != NULL || gatherstate->need_to_scan_locally) { - if (gatherstate->need_to_scan_workers) + if (gatherstate->funnel != NULL) { bool done = false; @@ -206,7 +210,7 @@ gather_getnext(GatherState *gatherstate) gatherstate->need_to_scan_locally, &done); if (done) - gatherstate->need_to_scan_workers = false; + ExecShutdownGather(gatherstate); if (HeapTupleIsValid(tup)) { @@ -247,30 +251,20 @@ gather_getnext(GatherState *gatherstate) void ExecShutdownGather(GatherState *node) { - Gather *gather; - - if (node->pei == NULL || node->pei->pcxt == NULL) - return; - - /* - * Ensure all workers have finished before destroying the parallel context - * to ensure a clean exit. - */ - if (node->funnel) + /* Shut down tuple queue funnel before shutting down workers. */ + if (node->funnel != NULL) { DestroyTupleQueueFunnel(node->funnel); node->funnel = NULL; } - ExecParallelFinish(node->pei); - - /* destroy parallel context. */ - DestroyParallelContext(node->pei->pcxt); - node->pei->pcxt = NULL; - - gather = (Gather *) node->ps.plan; - node->need_to_scan_locally = !gather->single_copy; - node->need_to_scan_workers = false; + /* Now shut down the workers. */ + if (node->pei != NULL) + { + ExecParallelFinish(node->pei); + ExecParallelCleanup(node->pei); + node->pei = NULL; + } } /* ---------------------------------------------------------------- @@ -295,5 +289,7 @@ ExecReScanGather(GatherState *node) */ ExecShutdownGather(node); + node->initialized = false; + ExecReScan(node->ps.lefttree); } diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h index 4fc797ad98..505500e76b 100644 --- a/src/include/executor/execParallel.h +++ b/src/include/executor/execParallel.h @@ -32,5 +32,6 @@ typedef struct ParallelExecutorInfo extern ParallelExecutorInfo *ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers); extern void ExecParallelFinish(ParallelExecutorInfo *pei); +extern void ExecParallelCleanup(ParallelExecutorInfo *pei); #endif /* EXECPARALLEL_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 23670e1ff9..4fcdcc4067 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1961,9 +1961,9 @@ typedef struct UniqueState typedef struct GatherState { PlanState ps; /* its first field is NodeTag */ + bool initialized; struct ParallelExecutorInfo *pei; struct TupleQueueFunnel *funnel; - bool need_to_scan_workers; bool need_to_scan_locally; } GatherState;