Location via proxy:   [ UP ]  
[Report a bug]   [Manage cookies]                
Skip to content

Commit bfc78d7

Browse files
committed
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 7aea8e4 is largely to blame for this problem, though this patch modifies some subsequently-committed code which relied on the guarantees it purported to make.
1 parent 816e336 commit bfc78d7

File tree

5 files changed

+95
-70
lines changed

5 files changed

+95
-70
lines changed

src/backend/executor/execMain.c

Lines changed: 24 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ static void CheckValidRowMarkRel(Relation rel, RowMarkType markType);
7676
static void ExecPostprocessPlan(EState *estate);
7777
static void ExecEndPlan(PlanState *planstate, EState *estate);
7878
static void ExecutePlan(EState *estate, PlanState *planstate,
79+
bool use_parallel_mode,
7980
CmdType operation,
8081
bool sendTuples,
8182
long numberTuples,
@@ -243,11 +244,6 @@ standard_ExecutorStart(QueryDesc *queryDesc, int eflags)
243244
if (!(eflags & (EXEC_FLAG_SKIP_TRIGGERS | EXEC_FLAG_EXPLAIN_ONLY)))
244245
AfterTriggerBeginQuery();
245246

246-
/* Enter parallel mode, if required by the query. */
247-
if (queryDesc->plannedstmt->parallelModeNeeded &&
248-
!(eflags & EXEC_FLAG_EXPLAIN_ONLY))
249-
EnterParallelMode();
250-
251247
MemoryContextSwitchTo(oldcontext);
252248
}
253249

@@ -341,15 +337,13 @@ standard_ExecutorRun(QueryDesc *queryDesc,
341337
if (!ScanDirectionIsNoMovement(direction))
342338
ExecutePlan(estate,
343339
queryDesc->planstate,
340+
queryDesc->plannedstmt->parallelModeNeeded,
344341
operation,
345342
sendTuples,
346343
count,
347344
direction,
348345
dest);
349346

350-
/* Allow nodes to release or shut down resources. */
351-
(void) ExecShutdownNode(queryDesc->planstate);
352-
353347
/*
354348
* shutdown tuple receiver, if we started it
355349
*/
@@ -482,11 +476,6 @@ standard_ExecutorEnd(QueryDesc *queryDesc)
482476
*/
483477
MemoryContextSwitchTo(oldcontext);
484478

485-
/* Exit parallel mode, if it was required by the query. */
486-
if (queryDesc->plannedstmt->parallelModeNeeded &&
487-
!(estate->es_top_eflags & EXEC_FLAG_EXPLAIN_ONLY))
488-
ExitParallelMode();
489-
490479
/*
491480
* Release EState and per-query memory context. This should release
492481
* everything the executor has allocated.
@@ -1529,6 +1518,7 @@ ExecEndPlan(PlanState *planstate, EState *estate)
15291518
static void
15301519
ExecutePlan(EState *estate,
15311520
PlanState *planstate,
1521+
bool use_parallel_mode,
15321522
CmdType operation,
15331523
bool sendTuples,
15341524
long numberTuples,
@@ -1548,6 +1538,20 @@ ExecutePlan(EState *estate,
15481538
*/
15491539
estate->es_direction = direction;
15501540

1541+
/*
1542+
* If a tuple count was supplied, we must force the plan to run without
1543+
* parallelism, because we might exit early.
1544+
*/
1545+
if (numberTuples != 0)
1546+
use_parallel_mode = false;
1547+
1548+
/*
1549+
* If a tuple count was supplied, we must force the plan to run without
1550+
* parallelism, because we might exit early.
1551+
*/
1552+
if (use_parallel_mode)
1553+
EnterParallelMode();
1554+
15511555
/*
15521556
* Loop until we've processed the proper number of tuples from the plan.
15531557
*/
@@ -1566,7 +1570,11 @@ ExecutePlan(EState *estate,
15661570
* process so we just end the loop...
15671571
*/
15681572
if (TupIsNull(slot))
1573+
{
1574+
/* Allow nodes to release or shut down resources. */
1575+
(void) ExecShutdownNode(planstate);
15691576
break;
1577+
}
15701578

15711579
/*
15721580
* If we have a junk filter, then project a new tuple with the junk
@@ -1603,6 +1611,9 @@ ExecutePlan(EState *estate,
16031611
if (numberTuples && numberTuples == current_tuple_count)
16041612
break;
16051613
}
1614+
1615+
if (use_parallel_mode)
1616+
ExitParallelMode();
16061617
}
16071618

16081619

src/backend/executor/execParallel.c

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -442,6 +442,23 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
442442
pei->instrumentation);
443443
}
444444

445+
/*
446+
* Clean up whatever ParallelExecutreInfo resources still exist after
447+
* ExecParallelFinish. We separate these routines because someone might
448+
* want to examine the contents of the DSM after ExecParallelFinish and
449+
* before calling this routine.
450+
*/
451+
void
452+
ExecParallelCleanup(ParallelExecutorInfo *pei)
453+
{
454+
if (pei->pcxt != NULL)
455+
{
456+
DestroyParallelContext(pei->pcxt);
457+
pei->pcxt = NULL;
458+
}
459+
pfree(pei);
460+
}
461+
445462
/*
446463
* Create a DestReceiver to write tuples we produce to the shm_mq designated
447464
* for that purpose.

src/backend/executor/nodeGather.c

Lines changed: 52 additions & 56 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
#include "postgres.h"
1717

1818
#include "access/relscan.h"
19+
#include "access/xact.h"
1920
#include "executor/execdebug.h"
2021
#include "executor/execParallel.h"
2122
#include "executor/nodeGather.h"
@@ -45,7 +46,6 @@ ExecInitGather(Gather *node, EState *estate, int eflags)
4546
gatherstate = makeNode(GatherState);
4647
gatherstate->ps.plan = (Plan *) node;
4748
gatherstate->ps.state = estate;
48-
gatherstate->need_to_scan_workers = false;
4949
gatherstate->need_to_scan_locally = !node->single_copy;
5050

5151
/*
@@ -106,52 +106,57 @@ ExecGather(GatherState *node)
106106
* needs to allocate large dynamic segement, so it is better to do if it
107107
* is really needed.
108108
*/
109-
if (!node->pei)
109+
if (!node->initialized)
110110
{
111111
EState *estate = node->ps.state;
112-
113-
/* Initialize the workers required to execute Gather node. */
114-
node->pei = ExecInitParallelPlan(node->ps.lefttree,
115-
estate,
116-
((Gather *) (node->ps.plan))->num_workers);
112+
Gather *gather = (Gather *) node->ps.plan;
117113

118114
/*
119-
* Register backend workers. If the required number of workers are not
120-
* available then we perform the scan with available workers and if
121-
* there are no more workers available, then the Gather node will just
122-
* scan locally.
115+
* Sometimes we might have to run without parallelism; but if
116+
* parallel mode is active then we can try to fire up some workers.
123117
*/
124-
LaunchParallelWorkers(node->pei->pcxt);
125-
126-
node->funnel = CreateTupleQueueFunnel();
127-
128-
for (i = 0; i < node->pei->pcxt->nworkers; ++i)
118+
if (gather->num_workers > 0 && IsInParallelMode())
129119
{
130-
if (node->pei->pcxt->worker[i].bgwhandle)
120+
bool got_any_worker = false;
121+
122+
/* Initialize the workers required to execute Gather node. */
123+
node->pei = ExecInitParallelPlan(node->ps.lefttree,
124+
estate,
125+
gather->num_workers);
126+
127+
/*
128+
* Register backend workers. We might not get as many as we
129+
* requested, or indeed any at all.
130+
*/
131+
LaunchParallelWorkers(node->pei->pcxt);
132+
133+
/* Set up a tuple queue to collect the results. */
134+
node->funnel = CreateTupleQueueFunnel();
135+
for (i = 0; i < node->pei->pcxt->nworkers; ++i)
131136
{
132-
shm_mq_set_handle(node->pei->tqueue[i],
133-
node->pei->pcxt->worker[i].bgwhandle);
134-
RegisterTupleQueueOnFunnel(node->funnel, node->pei->tqueue[i]);
135-
node->need_to_scan_workers = true;
137+
if (node->pei->pcxt->worker[i].bgwhandle)
138+
{
139+
shm_mq_set_handle(node->pei->tqueue[i],
140+
node->pei->pcxt->worker[i].bgwhandle);
141+
RegisterTupleQueueOnFunnel(node->funnel,
142+
node->pei->tqueue[i]);
143+
got_any_worker = true;
144+
}
136145
}
146+
147+
/* No workers? Then never mind. */
148+
if (!got_any_worker)
149+
ExecShutdownGather(node);
137150
}
138151

139-
/* If no workers are available, we must always scan locally. */
140-
if (!node->need_to_scan_workers)
141-
node->need_to_scan_locally = true;
152+
/* Run plan locally if no workers or not single-copy. */
153+
node->need_to_scan_locally = (node->funnel == NULL)
154+
|| !gather->single_copy;
155+
node->initialized = true;
142156
}
143157

144158
slot = gather_getnext(node);
145159

146-
if (TupIsNull(slot))
147-
{
148-
/*
149-
* Destroy the parallel context once we complete fetching all the
150-
* tuples. Otherwise, the DSM and workers will stick around for the
151-
* lifetime of the entire statement.
152-
*/
153-
ExecShutdownGather(node);
154-
}
155160
return slot;
156161
}
157162

@@ -194,10 +199,9 @@ gather_getnext(GatherState *gatherstate)
194199
*/
195200
slot = gatherstate->ps.ps_ProjInfo->pi_slot;
196201

197-
while (gatherstate->need_to_scan_workers ||
198-
gatherstate->need_to_scan_locally)
202+
while (gatherstate->funnel != NULL || gatherstate->need_to_scan_locally)
199203
{
200-
if (gatherstate->need_to_scan_workers)
204+
if (gatherstate->funnel != NULL)
201205
{
202206
bool done = false;
203207

@@ -206,7 +210,7 @@ gather_getnext(GatherState *gatherstate)
206210
gatherstate->need_to_scan_locally,
207211
&done);
208212
if (done)
209-
gatherstate->need_to_scan_workers = false;
213+
ExecShutdownGather(gatherstate);
210214

211215
if (HeapTupleIsValid(tup))
212216
{
@@ -247,30 +251,20 @@ gather_getnext(GatherState *gatherstate)
247251
void
248252
ExecShutdownGather(GatherState *node)
249253
{
250-
Gather *gather;
251-
252-
if (node->pei == NULL || node->pei->pcxt == NULL)
253-
return;
254-
255-
/*
256-
* Ensure all workers have finished before destroying the parallel context
257-
* to ensure a clean exit.
258-
*/
259-
if (node->funnel)
254+
/* Shut down tuple queue funnel before shutting down workers. */
255+
if (node->funnel != NULL)
260256
{
261257
DestroyTupleQueueFunnel(node->funnel);
262258
node->funnel = NULL;
263259
}
264260

265-
ExecParallelFinish(node->pei);
266-
267-
/* destroy parallel context. */
268-
DestroyParallelContext(node->pei->pcxt);
269-
node->pei->pcxt = NULL;
270-
271-
gather = (Gather *) node->ps.plan;
272-
node->need_to_scan_locally = !gather->single_copy;
273-
node->need_to_scan_workers = false;
261+
/* Now shut down the workers. */
262+
if (node->pei != NULL)
263+
{
264+
ExecParallelFinish(node->pei);
265+
ExecParallelCleanup(node->pei);
266+
node->pei = NULL;
267+
}
274268
}
275269

276270
/* ----------------------------------------------------------------
@@ -295,5 +289,7 @@ ExecReScanGather(GatherState *node)
295289
*/
296290
ExecShutdownGather(node);
297291

292+
node->initialized = false;
293+
298294
ExecReScan(node->ps.lefttree);
299295
}

src/include/executor/execParallel.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,5 +32,6 @@ typedef struct ParallelExecutorInfo
3232
extern ParallelExecutorInfo *ExecInitParallelPlan(PlanState *planstate,
3333
EState *estate, int nworkers);
3434
extern void ExecParallelFinish(ParallelExecutorInfo *pei);
35+
extern void ExecParallelCleanup(ParallelExecutorInfo *pei);
3536

3637
#endif /* EXECPARALLEL_H */

src/include/nodes/execnodes.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1961,9 +1961,9 @@ typedef struct UniqueState
19611961
typedef struct GatherState
19621962
{
19631963
PlanState ps; /* its first field is NodeTag */
1964+
bool initialized;
19641965
struct ParallelExecutorInfo *pei;
19651966
struct TupleQueueFunnel *funnel;
1966-
bool need_to_scan_workers;
19671967
bool need_to_scan_locally;
19681968
} GatherState;
19691969

0 commit comments

Comments
 (0)