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

Commit 166b61a

Browse files
committed
Avoid aggregating worker instrumentation multiple times.
Amit Kapila, per design ideas from me.
1 parent adeee97 commit 166b61a

File tree

3 files changed

+16
-11
lines changed

3 files changed

+16
-11
lines changed

src/backend/executor/execParallel.c

+13-5
Original file line numberDiff line numberDiff line change
@@ -277,13 +277,15 @@ ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize)
277277
}
278278

279279
/*
280-
* Re-initialize the response queues for backend workers to return tuples
281-
* to the main backend and start the workers.
280+
* Re-initialize the parallel executor info such that it can be reused by
281+
* workers.
282282
*/
283-
shm_mq_handle **
284-
ExecParallelReinitializeTupleQueues(ParallelContext *pcxt)
283+
void
284+
ExecParallelReinitialize(ParallelExecutorInfo *pei)
285285
{
286-
return ExecParallelSetupTupleQueues(pcxt, true);
286+
ReinitializeParallelDSM(pei->pcxt);
287+
pei->tqueue = ExecParallelSetupTupleQueues(pei->pcxt, true);
288+
pei->finished = false;
287289
}
288290

289291
/*
@@ -308,6 +310,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers)
308310

309311
/* Allocate object for return value. */
310312
pei = palloc0(sizeof(ParallelExecutorInfo));
313+
pei->finished = false;
311314
pei->planstate = planstate;
312315

313316
/* Fix up and serialize plan to be sent to workers. */
@@ -469,6 +472,9 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
469472
{
470473
int i;
471474

475+
if (pei->finished)
476+
return;
477+
472478
/* First, wait for the workers to finish. */
473479
WaitForParallelWorkersToFinish(pei->pcxt);
474480

@@ -480,6 +486,8 @@ ExecParallelFinish(ParallelExecutorInfo *pei)
480486
if (pei->instrumentation)
481487
ExecParallelRetrieveInstrumentation(pei->planstate,
482488
pei->instrumentation);
489+
490+
pei->finished = true;
483491
}
484492

485493
/*

src/backend/executor/nodeGather.c

+1-5
Original file line numberDiff line numberDiff line change
@@ -456,11 +456,7 @@ ExecReScanGather(GatherState *node)
456456
node->initialized = false;
457457

458458
if (node->pei)
459-
{
460-
ReinitializeParallelDSM(node->pei->pcxt);
461-
node->pei->tqueue =
462-
ExecParallelReinitializeTupleQueues(node->pei->pcxt);
463-
}
459+
ExecParallelReinitialize(node->pei);
464460

465461
ExecReScan(node->ps.lefttree);
466462
}

src/include/executor/execParallel.h

+2-1
Original file line numberDiff line numberDiff line change
@@ -27,12 +27,13 @@ typedef struct ParallelExecutorInfo
2727
BufferUsage *buffer_usage;
2828
SharedExecutorInstrumentation *instrumentation;
2929
shm_mq_handle **tqueue;
30+
bool finished;
3031
} ParallelExecutorInfo;
3132

3233
extern ParallelExecutorInfo *ExecInitParallelPlan(PlanState *planstate,
3334
EState *estate, int nworkers);
3435
extern void ExecParallelFinish(ParallelExecutorInfo *pei);
3536
extern void ExecParallelCleanup(ParallelExecutorInfo *pei);
36-
extern shm_mq_handle **ExecParallelReinitializeTupleQueues(ParallelContext *pcxt);
37+
extern void ExecParallelReinitialize(ParallelExecutorInfo *pei);
3738

3839
#endif /* EXECPARALLEL_H */

0 commit comments

Comments
 (0)