Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitbfc78d7

Browse files
committed
Rewrite interaction of parallel mode with parallel executor support.
In the previous coding, before returning from ExecutorRun, we'd shutdown all parallel workers. This was dead wrong if ExecutorRun wascalled with a non-zero tuple count; it had the effect of truncatingthe query output. To fix, give ExecutePlan control over whether toenter parallel mode, and have it refuse to do so if the tuple countis non-zero. Rewrite the Gather logic so that it can cope with beingcalled outside parallel mode.Commit7aea8e4 is largely to blamefor this problem, though this patch modifies some subsequently-committedcode which relied on the guarantees it purported to make.
1 parent816e336 commitbfc78d7

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
staticvoidExecPostprocessPlan(EState*estate);
7777
staticvoidExecEndPlan(PlanState*planstate,EState*estate);
7878
staticvoidExecutePlan(EState*estate,PlanState*planstate,
79+
booluse_parallel_mode,
7980
CmdTypeoperation,
8081
boolsendTuples,
8182
longnumberTuples,
@@ -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
staticvoid
15301519
ExecutePlan(EState*estate,
15311520
PlanState*planstate,
1521+
booluse_parallel_mode,
15321522
CmdTypeoperation,
15331523
boolsendTuples,
15341524
longnumberTuples,
@@ -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+
boolgot_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
returnslot;
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
booldone= 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
externParallelExecutorInfo*ExecInitParallelPlan(PlanState*planstate,
3333
EState*estate,intnworkers);
3434
externvoidExecParallelFinish(ParallelExecutorInfo*pei);
35+
externvoidExecParallelCleanup(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
typedefstructGatherState
19621962
{
19631963
PlanStateps;/* its first field is NodeTag */
1964+
boolinitialized;
19641965
structParallelExecutorInfo*pei;
19651966
structTupleQueueFunnel*funnel;
1966-
boolneed_to_scan_workers;
19671967
boolneed_to_scan_locally;
19681968
}GatherState;
19691969

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp