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

Commitd6a149f

Browse files
committed
Separate reinitialization of shared parallel-scan state from ExecReScan.
Previously, the parallel executor logic did reinitialization of sharedstate within the ExecReScan code for parallel-aware scan nodes. This isproblematic, because it means that the ExecReScan call has to occursynchronously (ie, during the parent Gather node's ReScan call). That isswimming very much against the tide so far as the ExecReScan machinery isconcerned; the fact that it works at all today depends on a lot of fragileassumptions, such as that no plan node between Gather and a parallel-awarescan node is parameterized. Another objection is that because ExecReScanmight be called in workers as well as the leader, hacky extra tests areneeded in some places to prevent unwanted shared-state resets.Hence, let's separate this code into two functions, a ReInitializeDSMcall and the ReScan call proper. ReInitializeDSM is called only inthe leader and is guaranteed to run before we start new workers.ReScan is returned to its traditional function of resetting only localstate, which means that ExecReScan's usual habits of delaying oreliminating child rescan calls are safe again.As with the preceding commit7df2c1f, it doesn't seem to be necessaryto make these changes in 9.6, which is a good thing because the FDW andCustomScan APIs are impacted.Discussion:https://postgr.es/m/CAA4eK1JkByysFJNh9M349u_nNjqETuEnY_y1VUc_kJiU0bxtaQ@mail.gmail.com
1 parent5816ddc commitd6a149f

22 files changed

+302
-135
lines changed

‎doc/src/sgml/custom-scan.sgml

Lines changed: 23 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -320,22 +320,39 @@ void (*InitializeDSMCustomScan) (CustomScanState *node,
320320
void *coordinate);
321321
</programlisting>
322322
Initialize the dynamic shared memory that will be required for parallel
323-
operation;<literal>coordinate</> points toan amount of allocated space
324-
equal to the return value of <function>EstimateDSMCustomScan</>.
323+
operation.<literal>coordinate</> points toa shared memory area of
324+
sizeequal to the return value of <function>EstimateDSMCustomScan</>.
325325
This callback is optional, and need only be supplied if this custom
326326
scan provider supports parallel execution.
327327
</para>
328328

329329
<para>
330330
<programlisting>
331+
void (*ReInitializeDSMCustomScan) (CustomScanState *node,
332+
ParallelContext *pcxt,
333+
void *coordinate);
334+
</programlisting>
335+
Re-initialize the dynamic shared memory required for parallel operation
336+
when the custom-scan plan node is about to be re-scanned.
337+
This callback is optional, and need only be supplied if this custom
338+
scan provider supports parallel execution.
339+
Recommended practice is that this callback reset only shared state,
340+
while the <function>ReScanCustomScan</> callback resets only local
341+
state. Currently, this callback will be called
342+
before <function>ReScanCustomScan</>, but it's best not to rely on
343+
that ordering.
344+
</para>
345+
346+
<para>
347+
<programlisting>
331348
void (*InitializeWorkerCustomScan) (CustomScanState *node,
332349
shm_toc *toc,
333350
void *coordinate);
334351
</programlisting>
335-
Initialize a parallel worker'scustom state based on the shared state
336-
set upin the leaderby <literal>InitializeDSMCustomScan</>.
337-
This callback is optional, andneeds only be supplied if this
338-
custom path supports parallel execution.
352+
Initialize a parallel worker'slocal state based on the shared state
353+
set upby the leaderduring <function>InitializeDSMCustomScan</>.
354+
This callback is optional, andneed only be supplied if this custom
355+
scan provider supports parallel execution.
339356
</para>
340357

341358
<para>

‎doc/src/sgml/fdwhandler.sgml

Lines changed: 31 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1191,12 +1191,12 @@ ImportForeignSchema (ImportForeignSchemaStmt *stmt, Oid serverOid);
11911191
<para>
11921192
A <structname>ForeignScan</> node can, optionally, support parallel
11931193
execution. A parallel <structname>ForeignScan</> will be executed
1194-
in multiple processes andshould return each rowonly once across
1194+
in multiple processes andmust return each rowexactly once across
11951195
all cooperating processes. To do this, processes can coordinate through
1196-
fixedsize chunks of dynamic shared memory. This shared memory is not
1197-
guaranteed to be mapped at the same address in every process, sopointers
1198-
may notbe used. The followingcallbacks are all optional in general,
1199-
but required if parallel execution is to be supported.
1196+
fixed-size chunks of dynamic shared memory. This shared memory is not
1197+
guaranteed to be mapped at the same address in every process, soit
1198+
must notcontain pointers. The followingfunctions are all optional,
1199+
butmost arerequired if parallel execution is to be supported.
12001200
</para>
12011201

12021202
<para>
@@ -1215,7 +1215,7 @@ IsForeignScanParallelSafe(PlannerInfo *root, RelOptInfo *rel,
12151215
</para>
12161216

12171217
<para>
1218-
If thiscallback is not defined, it is assumed that the scan must take
1218+
If thisfunction is not defined, it is assumed that the scan must take
12191219
place within the parallel leader. Note that returning true does not mean
12201220
that the scan itself can be done in parallel, only that the scan can be
12211221
performed within a parallel worker. Therefore, it can be useful to define
@@ -1230,6 +1230,9 @@ EstimateDSMForeignScan(ForeignScanState *node, ParallelContext *pcxt);
12301230
Estimate the amount of dynamic shared memory that will be required
12311231
for parallel operation. This may be higher than the amount that will
12321232
actually be used, but it must not be lower. The return value is in bytes.
1233+
This function is optional, and can be omitted if not needed; but if it
1234+
is omitted, the next three functions must be omitted as well, because
1235+
no shared memory will be allocated for the FDW's use.
12331236
</para>
12341237

12351238
<para>
@@ -1239,8 +1242,25 @@ InitializeDSMForeignScan(ForeignScanState *node, ParallelContext *pcxt,
12391242
void *coordinate);
12401243
</programlisting>
12411244
Initialize the dynamic shared memory that will be required for parallel
1242-
operation; <literal>coordinate</> points to an amount of allocated space
1243-
equal to the return value of <function>EstimateDSMForeignScan</>.
1245+
operation. <literal>coordinate</> points to a shared memory area of
1246+
size equal to the return value of <function>EstimateDSMForeignScan</>.
1247+
This function is optional, and can be omitted if not needed.
1248+
</para>
1249+
1250+
<para>
1251+
<programlisting>
1252+
void
1253+
ReInitializeDSMForeignScan(ForeignScanState *node, ParallelContext *pcxt,
1254+
void *coordinate);
1255+
</programlisting>
1256+
Re-initialize the dynamic shared memory required for parallel operation
1257+
when the foreign-scan plan node is about to be re-scanned.
1258+
This function is optional, and can be omitted if not needed.
1259+
Recommended practice is that this function reset only shared state,
1260+
while the <function>ReScanForeignScan</> function resets only local
1261+
state. Currently, this function will be called
1262+
before <function>ReScanForeignScan</>, but it's best not to rely on
1263+
that ordering.
12441264
</para>
12451265

12461266
<para>
@@ -1249,10 +1269,9 @@ void
12491269
InitializeWorkerForeignScan(ForeignScanState *node, shm_toc *toc,
12501270
void *coordinate);
12511271
</programlisting>
1252-
Initialize a parallel worker's custom state based on the shared state
1253-
set up in the leader by <literal>InitializeDSMForeignScan</>.
1254-
This callback is optional, and needs only be supplied if this
1255-
custom path supports parallel execution.
1272+
Initialize a parallel worker's local state based on the shared state
1273+
set up by the leader during <function>InitializeDSMForeignScan</>.
1274+
This function is optional, and can be omitted if not needed.
12561275
</para>
12571276

12581277
<para>

‎src/backend/access/heap/heapam.c

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -1525,25 +1525,6 @@ heap_rescan(HeapScanDesc scan,
15251525
* reinitialize scan descriptor
15261526
*/
15271527
initscan(scan,key, true);
1528-
1529-
/*
1530-
* reset parallel scan, if present
1531-
*/
1532-
if (scan->rs_parallel!=NULL)
1533-
{
1534-
ParallelHeapScanDescparallel_scan;
1535-
1536-
/*
1537-
* Caller is responsible for making sure that all workers have
1538-
* finished the scan before calling this, so it really shouldn't be
1539-
* necessary to acquire the mutex at all. We acquire it anyway, just
1540-
* to be tidy.
1541-
*/
1542-
parallel_scan=scan->rs_parallel;
1543-
SpinLockAcquire(&parallel_scan->phs_mutex);
1544-
parallel_scan->phs_cblock=parallel_scan->phs_startblock;
1545-
SpinLockRelease(&parallel_scan->phs_mutex);
1546-
}
15471528
}
15481529

15491530
/* ----------------
@@ -1640,6 +1621,25 @@ heap_parallelscan_initialize(ParallelHeapScanDesc target, Relation relation,
16401621
SerializeSnapshot(snapshot,target->phs_snapshot_data);
16411622
}
16421623

1624+
/* ----------------
1625+
*heap_parallelscan_reinitialize - reset a parallel scan
1626+
*
1627+
*Call this in the leader process. Caller is responsible for
1628+
*making sure that all workers have finished the scan beforehand.
1629+
* ----------------
1630+
*/
1631+
void
1632+
heap_parallelscan_reinitialize(ParallelHeapScanDescparallel_scan)
1633+
{
1634+
/*
1635+
* It shouldn't be necessary to acquire the mutex here, but we do it
1636+
* anyway, just to be tidy.
1637+
*/
1638+
SpinLockAcquire(&parallel_scan->phs_mutex);
1639+
parallel_scan->phs_cblock=parallel_scan->phs_startblock;
1640+
SpinLockRelease(&parallel_scan->phs_mutex);
1641+
}
1642+
16431643
/* ----------------
16441644
*heap_beginscan_parallel - join a parallel scan
16451645
*

‎src/backend/executor/execParallel.c

Lines changed: 72 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,8 @@ static bool ExecParallelInitializeDSM(PlanState *node,
109109
ExecParallelInitializeDSMContext*d);
110110
staticshm_mq_handle**ExecParallelSetupTupleQueues(ParallelContext*pcxt,
111111
boolreinitialize);
112+
staticboolExecParallelReInitializeDSM(PlanState*planstate,
113+
ParallelContext*pcxt);
112114
staticboolExecParallelRetrieveInstrumentation(PlanState*planstate,
113115
SharedExecutorInstrumentation*instrumentation);
114116

@@ -364,18 +366,6 @@ ExecParallelSetupTupleQueues(ParallelContext *pcxt, bool reinitialize)
364366
returnresponseq;
365367
}
366368

367-
/*
368-
* Re-initialize the parallel executor info such that it can be reused by
369-
* workers.
370-
*/
371-
void
372-
ExecParallelReinitialize(ParallelExecutorInfo*pei)
373-
{
374-
ReinitializeParallelDSM(pei->pcxt);
375-
pei->tqueue=ExecParallelSetupTupleQueues(pei->pcxt, true);
376-
pei->finished= false;
377-
}
378-
379369
/*
380370
* Sets up the required infrastructure for backend workers to perform
381371
* execution and return results to the main backend.
@@ -567,7 +557,7 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers)
567557
ExecParallelInitializeDSM(planstate,&d);
568558

569559
/*
570-
* Make sure that the world hasn't shifted under ourfeat. This could
560+
* Make sure that the world hasn't shifted under ourfeet. This could
571561
* probably just be an Assert(), but let's be conservative for now.
572562
*/
573563
if (e.nnodes!=d.nnodes)
@@ -577,6 +567,75 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, int nworkers)
577567
returnpei;
578568
}
579569

570+
/*
571+
* Re-initialize the parallel executor shared memory state before launching
572+
* a fresh batch of workers.
573+
*/
574+
void
575+
ExecParallelReinitialize(PlanState*planstate,
576+
ParallelExecutorInfo*pei)
577+
{
578+
/* Old workers must already be shut down */
579+
Assert(pei->finished);
580+
581+
ReinitializeParallelDSM(pei->pcxt);
582+
pei->tqueue=ExecParallelSetupTupleQueues(pei->pcxt, true);
583+
pei->finished= false;
584+
585+
/* Traverse plan tree and let each child node reset associated state. */
586+
ExecParallelReInitializeDSM(planstate,pei->pcxt);
587+
}
588+
589+
/*
590+
* Traverse plan tree to reinitialize per-node dynamic shared memory state
591+
*/
592+
staticbool
593+
ExecParallelReInitializeDSM(PlanState*planstate,
594+
ParallelContext*pcxt)
595+
{
596+
if (planstate==NULL)
597+
return false;
598+
599+
/*
600+
* Call reinitializers for DSM-using plan nodes.
601+
*/
602+
if (planstate->plan->parallel_aware)
603+
{
604+
switch (nodeTag(planstate))
605+
{
606+
caseT_SeqScanState:
607+
ExecSeqScanReInitializeDSM((SeqScanState*)planstate,
608+
pcxt);
609+
break;
610+
caseT_IndexScanState:
611+
ExecIndexScanReInitializeDSM((IndexScanState*)planstate,
612+
pcxt);
613+
break;
614+
caseT_IndexOnlyScanState:
615+
ExecIndexOnlyScanReInitializeDSM((IndexOnlyScanState*)planstate,
616+
pcxt);
617+
break;
618+
caseT_ForeignScanState:
619+
ExecForeignScanReInitializeDSM((ForeignScanState*)planstate,
620+
pcxt);
621+
break;
622+
caseT_CustomScanState:
623+
ExecCustomScanReInitializeDSM((CustomScanState*)planstate,
624+
pcxt);
625+
break;
626+
caseT_BitmapHeapScanState:
627+
ExecBitmapHeapReInitializeDSM((BitmapHeapScanState*)planstate,
628+
pcxt);
629+
break;
630+
631+
default:
632+
break;
633+
}
634+
}
635+
636+
returnplanstate_tree_walker(planstate,ExecParallelReInitializeDSM,pcxt);
637+
}
638+
580639
/*
581640
* Copy instrumentation information about this node and its descendants from
582641
* dynamic shared memory.

‎src/backend/executor/nodeBitmapHeapscan.c

Lines changed: 25 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -705,23 +705,6 @@ ExecReScanBitmapHeapScan(BitmapHeapScanState *node)
705705
node->shared_tbmiterator=NULL;
706706
node->shared_prefetch_iterator=NULL;
707707

708-
/* Reset parallel bitmap state, if present */
709-
if (node->pstate)
710-
{
711-
dsa_area*dsa=node->ss.ps.state->es_query_dsa;
712-
713-
node->pstate->state=BM_INITIAL;
714-
715-
if (DsaPointerIsValid(node->pstate->tbmiterator))
716-
tbm_free_shared_area(dsa,node->pstate->tbmiterator);
717-
718-
if (DsaPointerIsValid(node->pstate->prefetch_iterator))
719-
tbm_free_shared_area(dsa,node->pstate->prefetch_iterator);
720-
721-
node->pstate->tbmiterator=InvalidDsaPointer;
722-
node->pstate->prefetch_iterator=InvalidDsaPointer;
723-
}
724-
725708
ExecScanReScan(&node->ss);
726709

727710
/*
@@ -999,6 +982,31 @@ ExecBitmapHeapInitializeDSM(BitmapHeapScanState *node,
999982
node->pstate=pstate;
1000983
}
1001984

985+
/* ----------------------------------------------------------------
986+
*ExecBitmapHeapReInitializeDSM
987+
*
988+
*Reset shared state before beginning a fresh scan.
989+
* ----------------------------------------------------------------
990+
*/
991+
void
992+
ExecBitmapHeapReInitializeDSM(BitmapHeapScanState*node,
993+
ParallelContext*pcxt)
994+
{
995+
ParallelBitmapHeapState*pstate=node->pstate;
996+
dsa_area*dsa=node->ss.ps.state->es_query_dsa;
997+
998+
pstate->state=BM_INITIAL;
999+
1000+
if (DsaPointerIsValid(pstate->tbmiterator))
1001+
tbm_free_shared_area(dsa,pstate->tbmiterator);
1002+
1003+
if (DsaPointerIsValid(pstate->prefetch_iterator))
1004+
tbm_free_shared_area(dsa,pstate->prefetch_iterator);
1005+
1006+
pstate->tbmiterator=InvalidDsaPointer;
1007+
pstate->prefetch_iterator=InvalidDsaPointer;
1008+
}
1009+
10021010
/* ----------------------------------------------------------------
10031011
*ExecBitmapHeapInitializeWorker
10041012
*

‎src/backend/executor/nodeCustom.c

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -194,6 +194,21 @@ ExecCustomScanInitializeDSM(CustomScanState *node, ParallelContext *pcxt)
194194
}
195195
}
196196

197+
void
198+
ExecCustomScanReInitializeDSM(CustomScanState*node,ParallelContext*pcxt)
199+
{
200+
constCustomExecMethods*methods=node->methods;
201+
202+
if (methods->ReInitializeDSMCustomScan)
203+
{
204+
intplan_node_id=node->ss.ps.plan->plan_node_id;
205+
void*coordinate;
206+
207+
coordinate=shm_toc_lookup(pcxt->toc,plan_node_id, false);
208+
methods->ReInitializeDSMCustomScan(node,pcxt,coordinate);
209+
}
210+
}
211+
197212
void
198213
ExecCustomScanInitializeWorker(CustomScanState*node,shm_toc*toc)
199214
{

‎src/backend/executor/nodeForeignscan.c

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -332,7 +332,28 @@ ExecForeignScanInitializeDSM(ForeignScanState *node, ParallelContext *pcxt)
332332
}
333333

334334
/* ----------------------------------------------------------------
335-
*ExecForeignScanInitializeDSM
335+
*ExecForeignScanReInitializeDSM
336+
*
337+
*Reset shared state before beginning a fresh scan.
338+
* ----------------------------------------------------------------
339+
*/
340+
void
341+
ExecForeignScanReInitializeDSM(ForeignScanState*node,ParallelContext*pcxt)
342+
{
343+
FdwRoutine*fdwroutine=node->fdwroutine;
344+
345+
if (fdwroutine->ReInitializeDSMForeignScan)
346+
{
347+
intplan_node_id=node->ss.ps.plan->plan_node_id;
348+
void*coordinate;
349+
350+
coordinate=shm_toc_lookup(pcxt->toc,plan_node_id, false);
351+
fdwroutine->ReInitializeDSMForeignScan(node,pcxt,coordinate);
352+
}
353+
}
354+
355+
/* ----------------------------------------------------------------
356+
*ExecForeignScanInitializeWorker
336357
*
337358
*Initialization according to the parallel coordination information
338359
* ----------------------------------------------------------------

0 commit comments

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