NotificationsYou must be signed in to change notification settings
Fork5
Star26

Commitdf4685f

committed

Minor optimizations based on ParallelContext having nworkers_launched.

Originally, we didn't have nworkers_launched, so code that used parallelcontexts had to be preprared for the possibility that not all of theworkers requested actually got launched. But now we can count on knowingthe number of workers that were successfully launched, which can shaveoff a few cycles and simplify some code slightly.Amit Kapila, reviewed by Haribabu Kommi, per a suggestion from PeterGeoghegan.

1 parent546cd0d commitdf4685fCopy full SHA for df4685f

File tree

3 files changed

+12

-16

lines changed

src/backend
- access/transam
  - parallel.c
- executor
  - execParallel.c
  - nodeGather.c

3 files changed

+12

-16

lines changed

`‎src/backend/access/transam/parallel.c`

Lines changed: 4 additions & 4 deletions

Original file line number	Diff line number	Diff line change
`@@ -520,7 +520,7 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt)`
`520`	`520`	`*/`
`521`	`521`	`CHECK_FOR_INTERRUPTS();`
`522`	`522`
`523`		`-for (i=0;i<pcxt->nworkers;++i)`
	`523`	`+for (i=0;i<pcxt->nworkers_launched;++i)`
`524`	`524`	`{`
`525`	`525`	`if (pcxt->worker[i].error_mqh!=NULL)`
`526`	`526`	`{`
`@@ -560,7 +560,7 @@ WaitForParallelWorkersToExit(ParallelContext *pcxt)`
`560`	`560`	`inti;`
`561`	`561`
`562`	`562`	`/* Wait until the workers actually die. */`
`563`		`-for (i=0;i<pcxt->nworkers;++i)`
	`563`	`+for (i=0;i<pcxt->nworkers_launched;++i)`
`564`	`564`	`{`
`565`	`565`	`BgwHandleStatusstatus;`
`566`	`566`
`@@ -610,7 +610,7 @@ DestroyParallelContext(ParallelContext *pcxt)`
`610`	`610`	`/* Kill each worker in turn, and forget their error queues. */`
`611`	`611`	`if (pcxt->worker!=NULL)`
`612`	`612`	`{`
`613`		`-for (i=0;i<pcxt->nworkers;++i)`
	`613`	`+for (i=0;i<pcxt->nworkers_launched;++i)`
`614`	`614`	`{`
`615`	`615`	`if (pcxt->worker[i].error_mqh!=NULL)`
`616`	`616`	`{`
`@@ -708,7 +708,7 @@ HandleParallelMessages(void)`
`708`	`708`	`if (pcxt->worker==NULL)`
`709`	`709`	`continue;`
`710`	`710`
`711`		`-for (i=0;i<pcxt->nworkers;++i)`
	`711`	`+for (i=0;i<pcxt->nworkers_launched;++i)`
`712`	`712`	`{`
`713`	`713`	`/*`
`714`	`714`	`* Read as many messages as we can from each worker, but stop when`

`‎src/backend/executor/execParallel.c`

Lines changed: 1 addition & 1 deletion

Original file line number	Diff line number	Diff line change
`@@ -522,7 +522,7 @@ ExecParallelFinish(ParallelExecutorInfo *pei)`
`522`	`522`	`WaitForParallelWorkersToFinish(pei->pcxt);`
`523`	`523`
`524`	`524`	`/* Next, accumulate buffer usage. */`
`525`		`-for (i=0;i<pei->pcxt->nworkers;++i)`
	`525`	`+for (i=0;i<pei->pcxt->nworkers_launched;++i)`
`526`	`526`	`InstrAccumParallelQuery(&pei->buffer_usage[i]);`
`527`	`527`
`528`	`528`	`/* Finally, accumulate instrumentation, if any. */`

`‎src/backend/executor/nodeGather.c`

Lines changed: 7 additions & 11 deletions

Original file line number	Diff line number	Diff line change
`@@ -153,7 +153,6 @@ ExecGather(GatherState *node)`
`153`	`153`	`if (gather->num_workers>0&&IsInParallelMode())`
`154`	`154`	`{`
`155`	`155`	`ParallelContext*pcxt;`
`156`		`-boolgot_any_worker= false;`
`157`	`156`
`158`	`157`	`/* Initialize the workers required to execute Gather node. */`
`159`	`158`	`if (!node->pei)`
`@@ -169,29 +168,26 @@ ExecGather(GatherState *node)`
`169`	`168`	`LaunchParallelWorkers(pcxt);`
`170`	`169`
`171`	`170`	`/* Set up tuple queue readers to read the results. */`
`172`		`-if (pcxt->nworkers>0)`
	`171`	`+if (pcxt->nworkers_launched>0)`
`173`	`172`	`{`
`174`	`173`	`node->nreaders=0;`
`175`	`174`	`node->reader=`
`176`		`-palloc(pcxt->nworkerssizeof(TupleQueueReader));`
	`175`	`+palloc(pcxt->nworkers_launchedsizeof(TupleQueueReader));`
`177`	`176`
`178`		`-for (i=0;i<pcxt->nworkers;++i)`
	`177`	`+for (i=0;i<pcxt->nworkers_launched;++i)`
`179`	`178`	`{`
`180`		`-if (pcxt->worker[i].bgwhandle==NULL)`
`181`		`-continue;`
`182`		`-`
`183`	`179`	`shm_mq_set_handle(node->pei->tqueue[i],`
`184`	`180`	`pcxt->worker[i].bgwhandle);`
`185`	`181`	`node->reader[node->nreaders++]=`
`186`	`182`	`CreateTupleQueueReader(node->pei->tqueue[i],`
`187`	`183`	`fslot->tts_tupleDescriptor);`
`188`		`-got_any_worker= true;`
`189`	`184`	`}`
`190`	`185`	`}`
`191`		`-`
`192`		`-/* No workers? Then never mind. */`
`193`		`-if (!got_any_worker)`
	`186`	`+else`
	`187`	`+{`
	`188`	`+/* No workers? Then never mind. */`
`194`	`189`	`ExecShutdownGatherWorkers(node);`
	`190`	`+}`
`195`	`191`	`}`
`196`	`192`
`197`	`193`	`/* Run plan locally if no workers or not single-copy. */`

0 commit comments

Comments

(0)

Movatterモバイル変換

Navigation Menu

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Commitdf4685f

File tree

3 files changed

3 files changed

`‎src/backend/access/transam/parallel.c`

`‎src/backend/executor/execParallel.c`

`‎src/backend/executor/nodeGather.c`

0 commit comments