diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index bdb30752e098c..f9c6ed44787d8 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -2845,7 +2845,7 @@ _brin_parallel_scan_and_build(BrinBuildState *state, scan = table_beginscan_parallel(heap, ParallelTableScanFromBrinShared(brinshared), - SO_NONE); + NULL, SO_NONE); reltuples = table_index_build_scan(heap, index, indexInfo, true, true, brinbuildCallbackParallel, state, scan); diff --git a/src/backend/access/gin/gininsert.c b/src/backend/access/gin/gininsert.c index cb9ed3b563c6f..ee3e166b71965 100644 --- a/src/backend/access/gin/gininsert.c +++ b/src/backend/access/gin/gininsert.c @@ -2069,7 +2069,7 @@ _gin_parallel_scan_and_build(GinBuildState *state, scan = table_beginscan_parallel(heap, ParallelTableScanFromGinBuildShared(ginshared), - SO_NONE); + NULL, SO_NONE); reltuples = table_index_build_scan(heap, index, indexInfo, true, progress, ginBuildCallbackParallel, state, scan); diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index abfd8e8970a60..a22a51e85e6f6 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -1167,6 +1167,7 @@ TableScanDesc heap_beginscan(Relation relation, Snapshot snapshot, int nkeys, ScanKey key, ParallelTableScanDesc parallel_scan, + struct TableScanInstrumentation *instrument, uint32 flags) { HeapScanDesc scan; @@ -1317,9 +1318,15 @@ heap_beginscan(Relation relation, Snapshot snapshot, /* enable read stream instrumentation */ if ((flags & SO_SCAN_INSTRUMENT) && (scan->rs_read_stream != NULL)) { - scan->rs_base.rs_instrument = palloc0_object(TableScanInstrumentation); - read_stream_enable_stats(scan->rs_read_stream, - &scan->rs_base.rs_instrument->io); + /* + * The caller (the executor) always supplies the location where the + * statistics should be accumulated -- the node's own storage, or a + * parallel worker's slot in shared memory -- so the scan never owns + * the instrumentation and never frees it. + */ + Assert(instrument != NULL); + scan->rs_base.rs_instrument = instrument; + read_stream_enable_stats(scan->rs_read_stream, &instrument->io); } scan->rs_vmbuffer = InvalidBuffer; @@ -1425,9 +1432,6 @@ heap_endscan(TableScanDesc sscan) if (scan->rs_base.rs_flags & SO_TEMP_SNAPSHOT) UnregisterSnapshot(scan->rs_base.rs_snapshot); - if (scan->rs_base.rs_instrument) - pfree(scan->rs_base.rs_instrument); - pfree(scan); } diff --git a/src/backend/access/nbtree/nbtsort.c b/src/backend/access/nbtree/nbtsort.c index 756dfa3dcf47e..170fa5eaa111a 100644 --- a/src/backend/access/nbtree/nbtsort.c +++ b/src/backend/access/nbtree/nbtsort.c @@ -1929,7 +1929,7 @@ _bt_parallel_scan_and_sort(BTSpool *btspool, BTSpool *btspool2, indexInfo->ii_Concurrent = btshared->isconcurrent; scan = table_beginscan_parallel(btspool->heap, ParallelTableScanFromBTShared(btshared), - SO_NONE); + NULL, SO_NONE); reltuples = table_index_build_scan(btspool->heap, btspool->index, indexInfo, true, progress, _bt_build_callback, &buildstate, scan); diff --git a/src/backend/access/table/tableam.c b/src/backend/access/table/tableam.c index 68ff0966f1c57..935aca187fad0 100644 --- a/src/backend/access/table/tableam.c +++ b/src/backend/access/table/tableam.c @@ -118,7 +118,7 @@ table_beginscan_catalog(Relation relation, int nkeys, ScanKeyData *key) Snapshot snapshot = RegisterSnapshot(GetCatalogSnapshot(relid)); return table_beginscan_common(relation, snapshot, nkeys, key, - NULL, flags, SO_NONE); + NULL, NULL, flags, SO_NONE); } @@ -164,6 +164,7 @@ table_parallelscan_initialize(Relation rel, ParallelTableScanDesc pscan, TableScanDesc table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags) { Snapshot snapshot; @@ -186,12 +187,13 @@ table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan, } return table_beginscan_common(relation, snapshot, 0, NULL, - pscan, internal_flags, flags); + pscan, instrument, internal_flags, flags); } TableScanDesc table_beginscan_parallel_tidrange(Relation relation, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags) { Snapshot snapshot; @@ -217,7 +219,7 @@ table_beginscan_parallel_tidrange(Relation relation, } sscan = table_beginscan_common(relation, snapshot, 0, NULL, - pscan, internal_flags, flags); + pscan, instrument, internal_flags, flags); return sscan; } diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c index a40d03d35f3ba..3b48eb3d62aa4 100644 --- a/src/backend/commands/explain.c +++ b/src/backend/commands/explain.c @@ -3150,7 +3150,8 @@ show_sort_info(SortState *sortstate, ExplainState *es) const char *spaceType; int64 spaceUsed; - sinstrument = &sortstate->shared_info->sinstrument[n]; + sinstrument = GetWorkerInstr(sortstate->shared_info, + TuplesortInstrumentation, n); if (sinstrument->sortMethod == SORT_TYPE_STILL_IN_PROGRESS) continue; /* ignore any unfilled slots */ sortMethod = tuplesort_method_name(sinstrument->sortMethod); @@ -3345,7 +3346,7 @@ show_incremental_sort_info(IncrementalSortState *incrsortstate, for (n = 0; n < incrsortstate->shared_info->num_workers; n++) { IncrementalSortInfo *incsort_info = - &incrsortstate->shared_info->sinfo[n]; + GetWorkerInstr(incrsortstate->shared_info, IncrementalSortInfo, n); /* * If a worker hasn't processed any sort groups at all, then @@ -3416,12 +3417,13 @@ show_hash_info(HashState *hashstate, ExplainState *es) */ if (hashstate->shared_info) { - SharedHashInfo *shared_info = hashstate->shared_info; + SharedWorkerInstrumentation *shared_info = hashstate->shared_info; int i; for (i = 0; i < shared_info->num_workers; ++i) { - HashInstrumentation *worker_hi = &shared_info->hinstrument[i]; + HashInstrumentation *worker_hi = GetWorkerInstr(shared_info, + HashInstrumentation, i); hinstrument.nbuckets = Max(hinstrument.nbuckets, worker_hi->nbuckets); @@ -3697,7 +3699,7 @@ show_memoize_info(MemoizeState *mstate, List *ancestors, ExplainState *es) { MemoizeInstrumentation *si; - si = &mstate->shared_info->sinstrument[n]; + si = GetWorkerInstr(mstate->shared_info, MemoizeInstrumentation, n); /* * Skip workers that didn't do any work. We needn't bother checking @@ -3752,7 +3754,7 @@ static void show_hashagg_info(AggState *aggstate, ExplainState *es) { Agg *agg = (Agg *) aggstate->ss.ps.plan; - int64 memPeakKb = BYTES_TO_KILOBYTES(aggstate->hash_mem_peak); + int64 memPeakKb = BYTES_TO_KILOBYTES(aggstate->stats.hash_mem_peak); if (agg->aggstrategy != AGG_HASHED && agg->aggstrategy != AGG_MIXED) @@ -3769,13 +3771,13 @@ show_hashagg_info(AggState *aggstate, ExplainState *es) * detect this by checking how much memory it used. If we find it * didn't do any work then we don't show its properties. */ - if (es->analyze && aggstate->hash_mem_peak > 0) + if (es->analyze && aggstate->stats.hash_mem_peak > 0) { ExplainPropertyInteger("HashAgg Batches", NULL, - aggstate->hash_batches_used, es); + aggstate->stats.hash_batches_used, es); ExplainPropertyInteger("Peak Memory Usage", "kB", memPeakKb, es); ExplainPropertyInteger("Disk Usage", "kB", - aggstate->hash_disk_used, es); + aggstate->stats.hash_disk_used, es); } } else @@ -3795,7 +3797,7 @@ show_hashagg_info(AggState *aggstate, ExplainState *es) * detect this by checking how much memory it used. If we find it * didn't do any work then we don't show its properties. */ - if (es->analyze && aggstate->hash_mem_peak > 0) + if (es->analyze && aggstate->stats.hash_mem_peak > 0) { if (!gotone) ExplainIndentText(es); @@ -3803,14 +3805,14 @@ show_hashagg_info(AggState *aggstate, ExplainState *es) appendStringInfoSpaces(es->str, 2); appendStringInfo(es->str, "Batches: %d Memory Usage: " INT64_FORMAT "kB", - aggstate->hash_batches_used, memPeakKb); + aggstate->stats.hash_batches_used, memPeakKb); gotone = true; /* Only display disk usage if we spilled to disk */ - if (aggstate->hash_batches_used > 1) + if (aggstate->stats.hash_batches_used > 1) { appendStringInfo(es->str, " Disk Usage: " UINT64_FORMAT "kB", - aggstate->hash_disk_used); + aggstate->stats.hash_disk_used); } } @@ -3827,7 +3829,8 @@ show_hashagg_info(AggState *aggstate, ExplainState *es) uint64 hash_disk_used; int hash_batches_used; - sinstrument = &aggstate->shared_info->sinstrument[n]; + sinstrument = GetWorkerInstr(aggstate->shared_info, + AggregateInstrumentation, n); /* Skip workers that didn't do anything */ if (sinstrument->hash_mem_peak == 0) continue; @@ -3874,7 +3877,7 @@ static void show_indexsearches_info(PlanState *planstate, ExplainState *es) { Plan *plan = planstate->plan; - SharedIndexScanInstrumentation *SharedInfo = NULL; + SharedWorkerInstrumentation *SharedInfo = NULL; uint64 nsearches = 0; if (!es->analyze) @@ -3916,7 +3919,8 @@ show_indexsearches_info(PlanState *planstate, ExplainState *es) { for (int i = 0; i < SharedInfo->num_workers; ++i) { - IndexScanInstrumentation *winstrument = &SharedInfo->winstrument[i]; + IndexScanInstrumentation *winstrument = + GetWorkerInstr(SharedInfo, IndexScanInstrumentation, i); nsearches += winstrument->nsearches; } @@ -3960,7 +3964,9 @@ show_tidbitmap_info(BitmapHeapScanState *planstate, ExplainState *es) { for (int n = 0; n < planstate->sinstrument->num_workers; n++) { - BitmapHeapScanInstrumentation *si = &planstate->sinstrument->sinstrument[n]; + BitmapHeapScanInstrumentation *si = + GetWorkerInstr(planstate->sinstrument, + BitmapHeapScanInstrumentation, n); if (si->exact_pages == 0 && si->lossy_pages == 0) continue; @@ -4084,14 +4090,16 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es) { case T_BitmapHeapScan: { - SharedBitmapHeapInstrumentation *sinstrument + SharedWorkerInstrumentation *sinstrument = ((BitmapHeapScanState *) planstate)->sinstrument; if (sinstrument) { for (int i = 0; i < sinstrument->num_workers; ++i) { - BitmapHeapScanInstrumentation *winstrument = &sinstrument->sinstrument[i]; + BitmapHeapScanInstrumentation *winstrument = + GetWorkerInstr(sinstrument, + BitmapHeapScanInstrumentation, i); AccumulateIOStats(&stats, &winstrument->stats.io); @@ -4108,14 +4116,15 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es) } case T_SeqScan: { - SharedSeqScanInstrumentation *sinstrument + SharedWorkerInstrumentation *sinstrument = ((SeqScanState *) planstate)->sinstrument; if (sinstrument) { for (int i = 0; i < sinstrument->num_workers; ++i) { - SeqScanInstrumentation *winstrument = &sinstrument->sinstrument[i]; + SeqScanInstrumentation *winstrument = + GetWorkerInstr(sinstrument, SeqScanInstrumentation, i); AccumulateIOStats(&stats, &winstrument->stats.io); @@ -4132,14 +4141,16 @@ show_scan_io_usage(ScanState *planstate, ExplainState *es) } case T_TidRangeScan: { - SharedTidRangeScanInstrumentation *sinstrument + SharedWorkerInstrumentation *sinstrument = ((TidRangeScanState *) planstate)->trss_sinstrument; if (sinstrument) { for (int i = 0; i < sinstrument->num_workers; ++i) { - TidRangeScanInstrumentation *winstrument = &sinstrument->sinstrument[i]; + TidRangeScanInstrumentation *winstrument = + GetWorkerInstr(sinstrument, + TidRangeScanInstrumentation, i); AccumulateIOStats(&stats, &winstrument->stats.io); diff --git a/src/backend/executor/execParallel.c b/src/backend/executor/execParallel.c index 81b87d82fab47..bebf316aeb0cf 100644 --- a/src/backend/executor/execParallel.c +++ b/src/backend/executor/execParallel.c @@ -110,9 +110,20 @@ struct SharedExecutorInstrumentation * follows. */ }; -#define GetInstrumentationArray(sei) \ + +/* + * While the query runs, each parallel worker writes its per-node statistics + * directly into its own slot in this array (indexed by ParallelWorkerNumber), + * so there is exactly one writer per slot and no write-side locking is needed. + * To keep that hot path cheap we pad every slot out to a cache line so that two + * workers updating neighbouring slots don't ping-pong the same line. + */ +#define SizeOfInstrumentationSlot \ + CACHELINEALIGN(sizeof(NodeInstrumentation)) +#define GetInstrumentationSlot(sei, node, worker) \ (StaticAssertVariableIsOfTypeMacro(sei, SharedExecutorInstrumentation *), \ - (NodeInstrumentation *) (((char *) sei) + sei->instrument_offset)) + (NodeInstrumentation *) (((char *) (sei)) + (sei)->instrument_offset + \ + (((Size) (node)) * (sei)->num_workers + (worker)) * SizeOfInstrumentationSlot)) /* Context object for ExecParallelEstimate. */ typedef struct ExecParallelEstimateContext @@ -242,6 +253,67 @@ ExecSerializePlan(Plan *plan, EState *estate) * While we're at it, count the number of PlanState nodes in the tree, so * we know how many Instrumentation structures we need. */ + +/* + * Generic helpers for per-worker node instrumentation containers. + * + * Many node types collect a fixed-size instrumentation struct per parallel + * worker in a SharedWorkerInstrumentation container (see instrument_node.h). + * These helpers factor out the identical estimate/allocate/lookup/retrieve + * plumbing; only the element size, the TOC key, and the gating conditions vary + * between nodes, and those stay in the per-node callers. + */ + +/* Reserve DSM space (one chunk + one key) for the container. */ +void +ExecInstrEstimate(ParallelContext *pcxt, Size elemsz) +{ + shm_toc_estimate_chunk(&pcxt->estimator, + SharedWorkerInstrSize(pcxt->nworkers, elemsz)); + shm_toc_estimate_keys(&pcxt->estimator, 1); +} + +/* Allocate, zero, and publish the container under 'key'; returns it. */ +SharedWorkerInstrumentation * +ExecInstrInitDSM(ParallelContext *pcxt, uint64 key, Size elemsz) +{ + Size size = SharedWorkerInstrSize(pcxt->nworkers, elemsz); + SharedWorkerInstrumentation *si = shm_toc_allocate(pcxt->toc, size); + + /* ensure any unfilled slots will contain zeroes */ + memset(si, 0, size); + si->num_workers = pcxt->nworkers; + shm_toc_insert(pcxt->toc, key, si); + return si; +} + +/* Worker side: look up the container previously published under 'key'. */ +SharedWorkerInstrumentation * +ExecInstrInitWorker(shm_toc *toc, uint64 key, bool missing_ok) +{ + return (SharedWorkerInstrumentation *) shm_toc_lookup(toc, key, missing_ok); +} + +/* + * Leader side: copy the container out of (soon to be freed) DSM into a + * backend-local copy, so EXPLAIN can read it after the parallel context is + * torn down. Returns NULL if there was no instrumentation. + */ +SharedWorkerInstrumentation * +ExecInstrRetrieve(SharedWorkerInstrumentation *shared, Size elemsz) +{ + Size size; + SharedWorkerInstrumentation *si; + + if (shared == NULL) + return NULL; + + size = SharedWorkerInstrSize(shared->num_workers, elemsz); + si = palloc(size); + memcpy(si, shared, size); + return si; +} + static bool ExecParallelEstimate(PlanState *planstate, ExecParallelEstimateContext *e) { @@ -763,10 +835,11 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, instrumentation_len = offsetof(SharedExecutorInstrumentation, plan_node_id) + sizeof(int) * e.nnodes; - instrumentation_len = MAXALIGN(instrumentation_len); + /* Cache-line align so the first padded slot starts on a line boundary */ + instrumentation_len = CACHELINEALIGN(instrumentation_len); instrument_offset = instrumentation_len; instrumentation_len += - mul_size(sizeof(NodeInstrumentation), + mul_size(SizeOfInstrumentationSlot, mul_size(e.nnodes, nworkers)); shm_toc_estimate_chunk(&pcxt->estimator, instrumentation_len); shm_toc_estimate_keys(&pcxt->estimator, 1); @@ -852,17 +925,26 @@ ExecInitParallelPlan(PlanState *planstate, EState *estate, */ if (estate->es_instrument) { - NodeInstrumentation *instrument; - int i; + int node; + int worker; instrumentation = shm_toc_allocate(pcxt->toc, instrumentation_len); instrumentation->instrument_options = estate->es_instrument; instrumentation->instrument_offset = instrument_offset; instrumentation->num_workers = nworkers; instrumentation->num_plan_nodes = e.nnodes; - instrument = GetInstrumentationArray(instrumentation); - for (i = 0; i < nworkers * e.nnodes; ++i) - InstrInitNode(&instrument[i], estate->es_instrument, false); + + /* + * Initialize each worker's slot. Workers write into these directly + * as the query runs and never reset them, so this must happen exactly + * once (not on relaunch, where the slots accumulate across rounds). + * The async_mode flag is fixed up by each worker once it knows its + * node. + */ + for (node = 0; node < e.nnodes; ++node) + for (worker = 0; worker < nworkers; ++worker) + InstrInitNode(GetInstrumentationSlot(instrumentation, node, worker), + estate->es_instrument, false); shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION, instrumentation); pei->instrumentation = instrumentation; @@ -1094,7 +1176,6 @@ static bool ExecParallelRetrieveInstrumentation(PlanState *planstate, SharedExecutorInstrumentation *instrumentation) { - NodeInstrumentation *instrument; int i; int n; int ibytes; @@ -1108,11 +1189,14 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate, if (i >= instrumentation->num_plan_nodes) elog(ERROR, "plan node %d not found", plan_node_id); - /* Accumulate the statistics from all workers. */ - instrument = GetInstrumentationArray(instrumentation); - instrument += i * instrumentation->num_workers; + /* + * Accumulate the statistics from all workers. The workers already + * flushed their final cycle with InstrEndLoop before exiting, so no + * InstrEndLoop is needed here. + */ for (n = 0; n < instrumentation->num_workers; ++n) - InstrAggNode(planstate->instrument, &instrument[n]); + InstrAggNode(planstate->instrument, + GetInstrumentationSlot(instrumentation, i, n)); /* * Also store the per-worker detail. @@ -1128,7 +1212,15 @@ ExecParallelRetrieveInstrumentation(PlanState *planstate, MemoryContextSwitchTo(oldcontext); planstate->worker_instrument->num_workers = instrumentation->num_workers; - memcpy(&planstate->worker_instrument->instrument, instrument, ibytes); + + /* + * The shared slots are cache-line padded, so copy them into the tightly + * packed per-worker array one at a time rather than with a single memcpy. + */ + for (n = 0; n < instrumentation->num_workers; ++n) + memcpy(&planstate->worker_instrument->instrument[n], + GetInstrumentationSlot(instrumentation, i, n), + sizeof(NodeInstrumentation)); /* Perform any node-type-specific work that needs to be done. */ switch (nodeTag(planstate)) @@ -1351,24 +1443,27 @@ ExecParallelGetQueryDesc(shm_toc *toc, DestReceiver *receiver, } /* - * Copy instrumentation information from this node and its descendants into - * dynamic shared memory, so that the parallel leader can retrieve it. + * Point each node's instrumentation at this worker's slot in shared memory, so + * that the executor writes its statistics there directly as the query runs + * (rather than into worker-local memory that is copied out at shutdown). This + * runs once per worker, after ExecutorStart() has allocated the PlanState tree + * but before execution begins. */ static bool -ExecParallelReportInstrumentation(PlanState *planstate, - SharedExecutorInstrumentation *instrumentation) +ExecParallelRepointInstrumentation(PlanState *planstate, + SharedExecutorInstrumentation *instrumentation) { int i; int plan_node_id = planstate->plan->plan_node_id; - NodeInstrumentation *instrument; + NodeInstrumentation *slot; - InstrEndLoop(planstate->instrument); + Assert(IsParallelWorker()); + Assert(ParallelWorkerNumber < instrumentation->num_workers); /* - * If we shuffled the plan_node_id values in ps_instrument into sorted - * order, we could use binary search here. This might matter someday if - * we're pushing down sufficiently large plan trees. For now, do it the - * slow, dumb way. + * If we shuffled the plan_node_id values into sorted order, we could use + * binary search here. This might matter someday if we're pushing down + * sufficiently large plan trees. For now, do it the slow, dumb way. */ for (i = 0; i < instrumentation->num_plan_nodes; ++i) if (instrumentation->plan_node_id[i] == plan_node_id) @@ -1376,20 +1471,39 @@ ExecParallelReportInstrumentation(PlanState *planstate, if (i >= instrumentation->num_plan_nodes) elog(ERROR, "plan node %d not found", plan_node_id); + slot = GetInstrumentationSlot(instrumentation, i, ParallelWorkerNumber); + /* - * Add our statistics to the per-node, per-worker totals. It's possible - * that this could happen more than once if we relaunched workers. + * The leader initialized the slot's options but cannot know each node's + * async mode, so carry that over from the local instrumentation. We do + * not otherwise reset the slot: when workers are relaunched (e.g. a + * rescanned Gather) the slots accumulate across rounds. */ - instrument = GetInstrumentationArray(instrumentation); - instrument += i * instrumentation->num_workers; - Assert(IsParallelWorker()); - Assert(ParallelWorkerNumber < instrumentation->num_workers); - InstrAggNode(&instrument[ParallelWorkerNumber], planstate->instrument); + slot->async_mode = planstate->instrument->async_mode; - return planstate_tree_walker(planstate, ExecParallelReportInstrumentation, + /* Discard the worker-local instrumentation and write into shmem instead. */ + pfree(planstate->instrument); + planstate->instrument = slot; + + return planstate_tree_walker(planstate, ExecParallelRepointInstrumentation, instrumentation); } +/* + * Finish each node's instrumentation before this worker exits. Since the + * instrumentation already lives in shared memory, all that remains is to flush + * the final run cycle into the accumulated totals with InstrEndLoop(); the + * leader (and any relaunched worker) then sees a consistent slot. + */ +static bool +ExecParallelFinishInstrumentation(PlanState *planstate, void *context) +{ + InstrEndLoop(planstate->instrument); + + return planstate_tree_walker(planstate, ExecParallelFinishInstrumentation, + context); +} + /* * Initialize the PlanState and its descendants with the information * retrieved from shared memory. This has to be done once the PlanState @@ -1404,10 +1518,15 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) switch (nodeTag(planstate)) { case T_SeqScanState: + + /* + * Set up instrumentation first (even when not parallel-aware, for + * EXPLAIN ANALYZE) so that a parallel-aware scan can be pointed + * at the worker's shared I/O stats slot when it begins. + */ + ExecSeqScanInstrumentInitWorker((SeqScanState *) planstate, pwcxt); if (planstate->plan->parallel_aware) ExecSeqScanInitializeWorker((SeqScanState *) planstate, pwcxt); - /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecSeqScanInstrumentInitWorker((SeqScanState *) planstate, pwcxt); break; case T_IndexScanState: if (planstate->plan->parallel_aware) @@ -1436,12 +1555,17 @@ ExecParallelInitializeWorker(PlanState *planstate, ParallelWorkerContext *pwcxt) pwcxt); break; case T_TidRangeScanState: + + /* + * Set up instrumentation first (even when not parallel-aware, for + * EXPLAIN ANALYZE) so that a parallel-aware scan can be pointed + * at the worker's shared I/O stats slot when it begins. + */ + ExecTidRangeScanInstrumentInitWorker((TidRangeScanState *) planstate, + pwcxt); if (planstate->plan->parallel_aware) ExecTidRangeScanInitializeWorker((TidRangeScanState *) planstate, pwcxt); - /* even when not parallel-aware, for EXPLAIN ANALYZE */ - ExecTidRangeScanInstrumentInitWorker((TidRangeScanState *) planstate, - pwcxt); break; case T_AppendState: if (planstate->plan->parallel_aware) @@ -1564,6 +1688,14 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) pwcxt.seg = seg; ExecParallelInitializeWorker(queryDesc->planstate, &pwcxt); + /* + * If instrumentation is enabled, point each node's instrumentation at + * this worker's slot in shared memory so the executor writes statistics + * there directly, with no copy needed at shutdown. + */ + if (instrumentation != NULL) + ExecParallelRepointInstrumentation(queryDesc->planstate, instrumentation); + /* Pass down any tuple bound */ ExecSetTupleBound(fpes->tuples_needed, queryDesc->planstate); @@ -1593,10 +1725,13 @@ ParallelQueryMain(dsm_segment *seg, shm_toc *toc) InstrEndParallelQuery(&buffer_usage[ParallelWorkerNumber], &wal_usage[ParallelWorkerNumber]); - /* Report instrumentation data if any instrumentation options are set. */ + /* + * Finalize instrumentation if any instrumentation options are set. The + * statistics were written straight into shared memory during execution, + * so this just flushes each node's final run cycle. + */ if (instrumentation != NULL) - ExecParallelReportInstrumentation(queryDesc->planstate, - instrumentation); + ExecParallelFinishInstrumentation(queryDesc->planstate, NULL); /* Report JIT instrumentation data if any */ if (queryDesc->estate->es_jit && jit_instrumentation != NULL) diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c index 925caadd2cea9..44476a8cb2d3a 100644 --- a/src/backend/executor/nodeAgg.c +++ b/src/backend/executor/nodeAgg.c @@ -255,6 +255,7 @@ #include "catalog/pg_proc.h" #include "catalog/pg_type.h" #include "common/hashfn.h" +#include "executor/execParallel.h" #include "executor/execExpr.h" #include "executor/executor.h" #include "executor/instrument.h" @@ -1971,16 +1972,16 @@ hash_agg_update_metrics(AggState *aggstate, bool from_tape, int npartitions) /* update peak mem */ total_mem = meta_mem + entry_mem + hashkey_mem + buffer_mem; - if (total_mem > aggstate->hash_mem_peak) - aggstate->hash_mem_peak = total_mem; + if (total_mem > aggstate->instr->hash_mem_peak) + aggstate->instr->hash_mem_peak = total_mem; /* update disk usage */ if (aggstate->hash_tapeset != NULL) { uint64 disk_used = LogicalTapeSetBlocks(aggstate->hash_tapeset) * (BLCKSZ / 1024); - if (aggstate->hash_disk_used < disk_used) - aggstate->hash_disk_used = disk_used; + if (aggstate->instr->hash_disk_used < disk_used) + aggstate->instr->hash_disk_used = disk_used; } /* update hashentrysize estimate based on contents */ @@ -3226,7 +3227,7 @@ hashagg_spill_finish(AggState *aggstate, HashAggSpill *spill, int setno) spill->ntuples[i], cardinality, used_bits); aggstate->hash_batches = lappend(aggstate->hash_batches, new_batch); - aggstate->hash_batches_used++; + aggstate->instr->hash_batches_used++; } pfree(spill->ntuples); @@ -3315,6 +3316,13 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) aggstate->ss.ps.state = estate; aggstate->ss.ps.ExecProcNode = ExecAgg; + /* + * Accumulate hash instrumentation into the node's own storage by default; + * a parallel worker redirects this to its slot in shared memory (see + * ExecAggInitializeWorker). + */ + aggstate->instr = &aggstate->stats; + aggstate->aggs = NIL; aggstate->numaggs = 0; aggstate->numtrans = 0; @@ -3721,7 +3729,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags) aggstate->table_filled = false; /* Initialize this to 1, meaning nothing spilled, yet */ - aggstate->hash_batches_used = 1; + aggstate->instr->hash_batches_used = 1; } /* @@ -4401,20 +4409,10 @@ ExecEndAgg(AggState *node) int setno; /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE. + * In a parallel worker the hash statistics were accumulated directly into + * the worker's slot in shared memory (node->instr), so nothing needs to + * be copied here. */ - if (node->shared_info && IsParallelWorker()) - { - AggregateInstrumentation *si; - - Assert(ParallelWorkerNumber < node->shared_info->num_workers); - si = &node->shared_info->sinstrument[ParallelWorkerNumber]; - si->hash_batches_used = node->hash_batches_used; - si->hash_disk_used = node->hash_disk_used; - si->hash_mem_peak = node->hash_mem_peak; - } /* Make sure we have closed any open tuplesorts */ @@ -4780,16 +4778,11 @@ AggRegisterCallback(FunctionCallInfo fcinfo, void ExecAggEstimate(AggState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = mul_size(pcxt->nworkers, sizeof(AggregateInstrumentation)); - size = add_size(size, offsetof(SharedAggInfo, sinstrument)); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(AggregateInstrumentation)); } /* ---------------------------------------------------------------- @@ -4801,20 +4794,13 @@ ExecAggEstimate(AggState *node, ParallelContext *pcxt) void ExecAggInitializeDSM(AggState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedAggInfo, sinstrument) - + pcxt->nworkers * sizeof(AggregateInstrumentation); - node->shared_info = shm_toc_allocate(pcxt->toc, size); - /* ensure any unfilled slots will contain zeroes */ - memset(node->shared_info, 0, size); - node->shared_info->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, - node->shared_info); + node->shared_info = + ExecInstrInitDSM(pcxt, node->ss.ps.plan->plan_node_id, + sizeof(AggregateInstrumentation)); } /* ---------------------------------------------------------------- @@ -4827,7 +4813,22 @@ void ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt) { node->shared_info = - shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + ExecInstrInitWorker(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + + /* + * Accumulate hash statistics straight into this worker's slot in shared + * memory as the aggregation runs, rather than copying them out at + * shutdown. Carry over any values already initialized in ExecInitAgg + * (notably hash_batches_used = 1); this also resets the slot if workers + * are relaunched, so each round reports its own statistics. + */ + if (node->shared_info != NULL) + { + Assert(ParallelWorkerNumber < node->shared_info->num_workers); + node->instr = GetWorkerInstr(node->shared_info, AggregateInstrumentation, + ParallelWorkerNumber); + *node->instr = node->stats; + } } /* ---------------------------------------------------------------- @@ -4839,15 +4840,7 @@ ExecAggInitializeWorker(AggState *node, ParallelWorkerContext *pwcxt) void ExecAggRetrieveInstrumentation(AggState *node) { - Size size; - SharedAggInfo *si; - - if (node->shared_info == NULL) - return; - - size = offsetof(SharedAggInfo, sinstrument) - + node->shared_info->num_workers * sizeof(AggregateInstrumentation); - si = palloc(size); - memcpy(si, node->shared_info, size); - node->shared_info = si; + node->shared_info = + ExecInstrRetrieve(node->shared_info, + sizeof(AggregateInstrumentation)); } diff --git a/src/backend/executor/nodeBitmapHeapscan.c b/src/backend/executor/nodeBitmapHeapscan.c index 83d6478bc2b63..3e9bc94207da9 100644 --- a/src/backend/executor/nodeBitmapHeapscan.c +++ b/src/backend/executor/nodeBitmapHeapscan.c @@ -38,6 +38,7 @@ #include "access/relscan.h" #include "access/tableam.h" #include "access/visibilitymap.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/instrument.h" #include "executor/nodeBitmapHeapscan.h" @@ -93,6 +94,25 @@ typedef struct ParallelBitmapHeapState } ParallelBitmapHeapState; +/* + * Where this scan should accumulate instrumentation (page counts and I/O): this + * worker's slot in shared memory when run in a parallel worker, so statistics + * are written there directly as the scan runs (no copy at shutdown, and the + * leader can observe progress mid-query); otherwise the node's own local + * storage. + */ +static BitmapHeapScanInstrumentation * +ExecBitmapHeapInstrumentation(BitmapHeapScanState *node) +{ + if (node->sinstrument != NULL && IsParallelWorker()) + { + Assert(ParallelWorkerNumber < node->sinstrument->num_workers); + return GetWorkerInstr(node->sinstrument, BitmapHeapScanInstrumentation, + ParallelWorkerNumber); + } + return &node->stats; +} + /* * Do the underlying index scan, build the bitmap, set up the parallel state * needed for parallel workers to iterate through the bitmap, and set up the @@ -157,6 +177,7 @@ BitmapTableScanSetup(BitmapHeapScanState *node) node->ss.ps.state->es_snapshot, 0, NULL, + &ExecBitmapHeapInstrumentation(node)->stats, flags); } @@ -175,6 +196,7 @@ BitmapHeapNext(BitmapHeapScanState *node) { ExprContext *econtext = node->ss.ps.ps_ExprContext; TupleTableSlot *slot = node->ss.ss_ScanTupleSlot; + BitmapHeapScanInstrumentation *instr = ExecBitmapHeapInstrumentation(node); /* * If we haven't yet performed the underlying index scan, do it, and begin @@ -185,8 +207,8 @@ BitmapHeapNext(BitmapHeapScanState *node) while (table_scan_bitmap_next_tuple(node->ss.ss_currentScanDesc, slot, &node->recheck, - &node->stats.lossy_pages, - &node->stats.exact_pages)) + &instr->lossy_pages, + &instr->exact_pages)) { /* * Continuing in previously obtained page. @@ -317,35 +339,10 @@ ExecEndBitmapHeapScan(BitmapHeapScanState *node) TableScanDesc scanDesc; /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE. + * In a parallel worker the scan accumulated its page counts and I/O + * statistics straight into the worker's shared-memory slot (see + * ExecBitmapHeapInstrumentation), so there is nothing to collect here. */ - if (node->sinstrument != NULL && IsParallelWorker()) - { - BitmapHeapScanInstrumentation *si; - - Assert(ParallelWorkerNumber < node->sinstrument->num_workers); - si = &node->sinstrument->sinstrument[ParallelWorkerNumber]; - - /* - * Here we accumulate the stats rather than performing memcpy on - * node->stats into si. When a Gather/GatherMerge node finishes it - * will perform planner shutdown on the workers. On rescan it will - * spin up new workers which will have a new BitmapHeapScanState and - * zeroed stats. - */ - si->exact_pages += node->stats.exact_pages; - si->lossy_pages += node->stats.lossy_pages; - - /* collect I/O instrumentation for this process */ - if (node->ss.ss_currentScanDesc && - node->ss.ss_currentScanDesc->rs_instrument) - { - AccumulateIOStats(&si->stats.io, - &node->ss.ss_currentScanDesc->rs_instrument->io); - } - } /* * extract information from the node @@ -597,15 +594,10 @@ void ExecBitmapHeapInstrumentEstimate(BitmapHeapScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedBitmapHeapInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(BitmapHeapScanInstrumentation))); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(BitmapHeapScanInstrumentation)); } /* @@ -615,23 +607,14 @@ void ExecBitmapHeapInstrumentInitDSM(BitmapHeapScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedBitmapHeapInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(BitmapHeapScanInstrumentation))); node->sinstrument = - (SharedBitmapHeapInstrumentation *) shm_toc_allocate(pcxt->toc, size); - - /* Each per-worker area must start out as zeroes */ - memset(node->sinstrument, 0, size); - node->sinstrument->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - node->sinstrument); + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(BitmapHeapScanInstrumentation)); } /* @@ -644,11 +627,11 @@ ExecBitmapHeapInstrumentInitWorker(BitmapHeapScanState *node, if (!node->ss.ps.instrument) return; - node->sinstrument = (SharedBitmapHeapInstrumentation *) - shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->sinstrument = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); } /* ---------------------------------------------------------------- @@ -660,15 +643,7 @@ ExecBitmapHeapInstrumentInitWorker(BitmapHeapScanState *node, void ExecBitmapHeapRetrieveInstrumentation(BitmapHeapScanState *node) { - SharedBitmapHeapInstrumentation *sinstrument = node->sinstrument; - Size size; - - if (sinstrument == NULL) - return; - - size = offsetof(SharedBitmapHeapInstrumentation, sinstrument) - + sinstrument->num_workers * sizeof(BitmapHeapScanInstrumentation); - - node->sinstrument = palloc(size); - memcpy(node->sinstrument, sinstrument, size); + node->sinstrument = + ExecInstrRetrieve(node->sinstrument, + sizeof(BitmapHeapScanInstrumentation)); } diff --git a/src/backend/executor/nodeBitmapIndexscan.c b/src/backend/executor/nodeBitmapIndexscan.c index 7978514e1bc94..29fbac6eec882 100644 --- a/src/backend/executor/nodeBitmapIndexscan.c +++ b/src/backend/executor/nodeBitmapIndexscan.c @@ -22,6 +22,8 @@ #include "postgres.h" #include "access/genam.h" +#include "access/relscan.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/instrument.h" #include "executor/nodeBitmapIndexscan.h" @@ -186,25 +188,11 @@ ExecEndBitmapIndexScan(BitmapIndexScanState *node) indexScanDesc = node->biss_ScanDesc; /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE + * In a parallel worker biss_Instrument points directly at the worker's + * slot in shared memory (see ExecBitmapIndexScanInitializeWorker), so the + * stats are already where the leader will read them; nothing to copy + * here. */ - if (node->biss_SharedInfo != NULL && IsParallelWorker()) - { - IndexScanInstrumentation *winstrument; - - Assert(ParallelWorkerNumber < node->biss_SharedInfo->num_workers); - winstrument = &node->biss_SharedInfo->winstrument[ParallelWorkerNumber]; - - /* - * We have to accumulate the stats rather than performing a memcpy. - * When a Gather/GatherMerge node finishes it will perform planner - * shutdown on the workers. On rescan it will spin up new workers - * which will have a new BitmapIndexScanState and zeroed stats. - */ - winstrument->nsearches += node->biss_Instrument->nsearches; - } /* * close the index relation (no-op if we didn't open it) @@ -358,8 +346,6 @@ ExecInitBitmapIndexScan(BitmapIndexScan *node, EState *estate, int eflags) void ExecBitmapIndexScanEstimate(BitmapIndexScanState *node, ParallelContext *pcxt) { - Size size; - /* * Parallel bitmap index scans are not supported, but we still need to * store the scan's instrumentation in DSM during parallel query @@ -367,10 +353,7 @@ ExecBitmapIndexScanEstimate(BitmapIndexScanState *node, ParallelContext *pcxt) if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedIndexScanInstrumentation, winstrument) + - pcxt->nworkers * sizeof(IndexScanInstrumentation); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(IndexScanInstrumentation)); } /* ---------------------------------------------------------------- @@ -383,25 +366,15 @@ void ExecBitmapIndexScanInitializeDSM(BitmapIndexScanState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedIndexScanInstrumentation, winstrument) + - pcxt->nworkers * sizeof(IndexScanInstrumentation); node->biss_SharedInfo = - (SharedIndexScanInstrumentation *) shm_toc_allocate(pcxt->toc, - size); - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - node->biss_SharedInfo); - - /* Each per-worker area must start out as zeroes */ - memset(node->biss_SharedInfo, 0, size); - node->biss_SharedInfo->num_workers = pcxt->nworkers; + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(IndexScanInstrumentation)); } /* ---------------------------------------------------------------- @@ -418,11 +391,34 @@ ExecBitmapIndexScanInitializeWorker(BitmapIndexScanState *node, if (!node->ss.ps.instrument) return; - node->biss_SharedInfo = (SharedIndexScanInstrumentation *) - shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->biss_SharedInfo = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); + + /* + * Write statistics straight into this worker's shared slot as the scan + * runs, rather than into worker-local memory copied out at shutdown. This + * has no write-side contention (one writer per slot) and lets the leader + * observe progress mid-query. The slot is not reset here: when workers + * are relaunched (e.g. a rescanned Gather) the counts accumulate across + * rounds. + */ + Assert(ParallelWorkerNumber < node->biss_SharedInfo->num_workers); + pfree(node->biss_Instrument); + node->biss_Instrument = GetWorkerInstr(node->biss_SharedInfo, + IndexScanInstrumentation, + ParallelWorkerNumber); + + /* + * Unlike a plain index scan, the bitmap index scan descriptor is created + * eagerly in ExecInitBitmapIndexScan, before this runs, and it captured a + * pointer to the now-freed local instrumentation. Repoint it at the + * shared slot so the AM writes there (and not into freed memory). + */ + if (node->biss_ScanDesc != NULL) + node->biss_ScanDesc->instrument = node->biss_Instrument; } /* ---------------------------------------------------------------- @@ -434,15 +430,8 @@ ExecBitmapIndexScanInitializeWorker(BitmapIndexScanState *node, void ExecBitmapIndexScanRetrieveInstrumentation(BitmapIndexScanState *node) { - SharedIndexScanInstrumentation *SharedInfo = node->biss_SharedInfo; - size_t size; - - if (SharedInfo == NULL) - return; - /* Create a copy of SharedInfo in backend-local memory */ - size = offsetof(SharedIndexScanInstrumentation, winstrument) + - SharedInfo->num_workers * sizeof(IndexScanInstrumentation); - node->biss_SharedInfo = palloc(size); - memcpy(node->biss_SharedInfo, SharedInfo, size); + node->biss_SharedInfo = + ExecInstrRetrieve(node->biss_SharedInfo, + sizeof(IndexScanInstrumentation)); } diff --git a/src/backend/executor/nodeHash.c b/src/backend/executor/nodeHash.c index 8825bb6fa23fe..99b4bdfe3b8f5 100644 --- a/src/backend/executor/nodeHash.c +++ b/src/backend/executor/nodeHash.c @@ -30,6 +30,7 @@ #include "access/parallel.h" #include "catalog/pg_statistic.h" #include "commands/tablespace.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/hashjoin.h" #include "executor/instrument.h" @@ -2818,16 +2819,11 @@ ExecHashBuildNullTupleStore(HashJoinTable hashtable) void ExecHashEstimate(HashState *node, ParallelContext *pcxt) { - size_t size; - /* don't need this if not instrumenting or no workers */ if (!node->ps.instrument || pcxt->nworkers == 0) return; - size = mul_size(pcxt->nworkers, sizeof(HashInstrumentation)); - size = add_size(size, offsetof(SharedHashInfo, hinstrument)); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(HashInstrumentation)); } /* @@ -2837,22 +2833,13 @@ ExecHashEstimate(HashState *node, ParallelContext *pcxt) void ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt) { - size_t size; - /* don't need this if not instrumenting or no workers */ if (!node->ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedHashInfo, hinstrument) + - pcxt->nworkers * sizeof(HashInstrumentation); - node->shared_info = (SharedHashInfo *) shm_toc_allocate(pcxt->toc, size); - - /* Each per-worker area must start out as zeroes. */ - memset(node->shared_info, 0, size); - - node->shared_info->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, node->ps.plan->plan_node_id, - node->shared_info); + node->shared_info = + ExecInstrInitDSM(pcxt, node->ps.plan->plan_node_id, + sizeof(HashInstrumentation)); } /* @@ -2862,8 +2849,6 @@ ExecHashInitializeDSM(HashState *node, ParallelContext *pcxt) void ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt) { - SharedHashInfo *shared_info; - /* don't need this if not instrumenting */ if (!node->ps.instrument) return; @@ -2873,9 +2858,10 @@ ExecHashInitializeWorker(HashState *node, ParallelWorkerContext *pwcxt) * we'll accumulate stats there when shutting down or rebuilding the hash * table. */ - shared_info = (SharedHashInfo *) - shm_toc_lookup(pwcxt->toc, node->ps.plan->plan_node_id, false); - node->hinstrument = &shared_info->hinstrument[ParallelWorkerNumber]; + node->shared_info = + ExecInstrInitWorker(pwcxt->toc, node->ps.plan->plan_node_id, false); + node->hinstrument = GetWorkerInstr(node->shared_info, HashInstrumentation, + ParallelWorkerNumber); } /* @@ -2903,17 +2889,10 @@ ExecShutdownHash(HashState *node) void ExecHashRetrieveInstrumentation(HashState *node) { - SharedHashInfo *shared_info = node->shared_info; - size_t size; - - if (shared_info == NULL) - return; - /* Replace node->shared_info with a copy in backend-local memory. */ - size = offsetof(SharedHashInfo, hinstrument) + - shared_info->num_workers * sizeof(HashInstrumentation); - node->shared_info = palloc(size); - memcpy(node->shared_info, shared_info, size); + node->shared_info = + ExecInstrRetrieve(node->shared_info, + sizeof(HashInstrumentation)); } /* diff --git a/src/backend/executor/nodeIncrementalSort.c b/src/backend/executor/nodeIncrementalSort.c index 1d831049b65aa..964fa85a53b5f 100644 --- a/src/backend/executor/nodeIncrementalSort.c +++ b/src/backend/executor/nodeIncrementalSort.c @@ -78,6 +78,7 @@ #include "postgres.h" +#include "executor/execParallel.h" #include "executor/execdebug.h" #include "executor/nodeIncrementalSort.h" #include "miscadmin.h" @@ -103,7 +104,7 @@ { \ Assert(IsParallelWorker()); \ Assert(ParallelWorkerNumber < (node)->shared_info->num_workers); \ - instrumentSortedGroup(&(node)->shared_info->sinfo[ParallelWorkerNumber].groupName##GroupInfo, \ + instrumentSortedGroup(&GetWorkerInstr((node)->shared_info, IncrementalSortInfo, ParallelWorkerNumber)->groupName##GroupInfo, \ (node)->groupName##_state); \ } \ else \ @@ -1172,16 +1173,11 @@ ExecReScanIncrementalSort(IncrementalSortState *node) void ExecIncrementalSortEstimate(IncrementalSortState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = mul_size(pcxt->nworkers, sizeof(IncrementalSortInfo)); - size = add_size(size, offsetof(SharedIncrementalSortInfo, sinfo)); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(IncrementalSortInfo)); } /* ---------------------------------------------------------------- @@ -1193,20 +1189,13 @@ ExecIncrementalSortEstimate(IncrementalSortState *node, ParallelContext *pcxt) void ExecIncrementalSortInitializeDSM(IncrementalSortState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedIncrementalSortInfo, sinfo) - + pcxt->nworkers * sizeof(IncrementalSortInfo); - node->shared_info = shm_toc_allocate(pcxt->toc, size); - /* ensure any unfilled slots will contain zeroes */ - memset(node->shared_info, 0, size); - node->shared_info->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, - node->shared_info); + node->shared_info = + ExecInstrInitDSM(pcxt, node->ss.ps.plan->plan_node_id, + sizeof(IncrementalSortInfo)); } /* ---------------------------------------------------------------- @@ -1219,7 +1208,7 @@ void ExecIncrementalSortInitializeWorker(IncrementalSortState *node, ParallelWorkerContext *pwcxt) { node->shared_info = - shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + ExecInstrInitWorker(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); node->am_worker = true; } @@ -1232,15 +1221,7 @@ ExecIncrementalSortInitializeWorker(IncrementalSortState *node, ParallelWorkerCo void ExecIncrementalSortRetrieveInstrumentation(IncrementalSortState *node) { - Size size; - SharedIncrementalSortInfo *si; - - if (node->shared_info == NULL) - return; - - size = offsetof(SharedIncrementalSortInfo, sinfo) - + node->shared_info->num_workers * sizeof(IncrementalSortInfo); - si = palloc(size); - memcpy(si, node->shared_info, size); - node->shared_info = si; + node->shared_info = + ExecInstrRetrieve(node->shared_info, + sizeof(IncrementalSortInfo)); } diff --git a/src/backend/executor/nodeIndexonlyscan.c b/src/backend/executor/nodeIndexonlyscan.c index d52012e8a6987..adb41e624dcc5 100644 --- a/src/backend/executor/nodeIndexonlyscan.c +++ b/src/backend/executor/nodeIndexonlyscan.c @@ -36,6 +36,7 @@ #include "access/tupdesc.h" #include "access/visibilitymap.h" #include "catalog/pg_type.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/instrument.h" #include "executor/nodeIndexonlyscan.h" @@ -418,25 +419,11 @@ ExecEndIndexOnlyScan(IndexOnlyScanState *node) } /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE + * In a parallel worker ioss_Instrument points directly at the worker's + * slot in shared memory (see ExecIndexOnlyScanInstrumentInitWorker), so + * the stats are already where the leader will read them; nothing to copy + * here. */ - if (node->ioss_SharedInfo != NULL && IsParallelWorker()) - { - IndexScanInstrumentation *winstrument; - - Assert(ParallelWorkerNumber < node->ioss_SharedInfo->num_workers); - winstrument = &node->ioss_SharedInfo->winstrument[ParallelWorkerNumber]; - - /* - * We have to accumulate the stats rather than performing a memcpy. - * When a Gather/GatherMerge node finishes it will perform planner - * shutdown on the workers. On rescan it will spin up new workers - * which will have a new IndexOnlyScanState and zeroed stats. - */ - winstrument->nsearches += node->ioss_Instrument->nsearches; - } /* * close the index relation (no-op if we didn't open it) @@ -844,20 +831,10 @@ void ExecIndexOnlyScanInstrumentEstimate(IndexOnlyScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - /* - * This size calculation is trivial enough that we don't bother saving it - * in the IndexOnlyScanState. We'll recalculate the needed size in - * ExecIndexOnlyScanInstrumentInitDSM(). - */ - size = add_size(offsetof(SharedIndexScanInstrumentation, winstrument), - mul_size(pcxt->nworkers, sizeof(IndexScanInstrumentation))); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(IndexScanInstrumentation)); } /* @@ -867,23 +844,14 @@ void ExecIndexOnlyScanInstrumentInitDSM(IndexOnlyScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedIndexScanInstrumentation, winstrument), - mul_size(pcxt->nworkers, sizeof(IndexScanInstrumentation))); node->ioss_SharedInfo = - (SharedIndexScanInstrumentation *) shm_toc_allocate(pcxt->toc, size); - - /* Each per-worker area must start out as zeroes */ - memset(node->ioss_SharedInfo, 0, size); - node->ioss_SharedInfo->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - node->ioss_SharedInfo); + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(IndexScanInstrumentation)); } /* @@ -896,11 +864,25 @@ ExecIndexOnlyScanInstrumentInitWorker(IndexOnlyScanState *node, if (!node->ss.ps.instrument) return; - node->ioss_SharedInfo = (SharedIndexScanInstrumentation *) - shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->ioss_SharedInfo = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); + + /* + * Write statistics straight into this worker's shared slot as the scan + * runs, rather than into worker-local memory copied out at shutdown. This + * has no write-side contention (one writer per slot) and lets the leader + * observe progress mid-query. The slot is not reset here: when workers + * are relaunched (e.g. a rescanned Gather) the counts accumulate across + * rounds. + */ + Assert(ParallelWorkerNumber < node->ioss_SharedInfo->num_workers); + pfree(node->ioss_Instrument); + node->ioss_Instrument = GetWorkerInstr(node->ioss_SharedInfo, + IndexScanInstrumentation, + ParallelWorkerNumber); } /* ---------------------------------------------------------------- @@ -912,15 +894,8 @@ ExecIndexOnlyScanInstrumentInitWorker(IndexOnlyScanState *node, void ExecIndexOnlyScanRetrieveInstrumentation(IndexOnlyScanState *node) { - SharedIndexScanInstrumentation *SharedInfo = node->ioss_SharedInfo; - size_t size; - - if (SharedInfo == NULL) - return; - /* Create a copy of SharedInfo in backend-local memory */ - size = offsetof(SharedIndexScanInstrumentation, winstrument) + - SharedInfo->num_workers * sizeof(IndexScanInstrumentation); - node->ioss_SharedInfo = palloc(size); - memcpy(node->ioss_SharedInfo, SharedInfo, size); + node->ioss_SharedInfo = + ExecInstrRetrieve(node->ioss_SharedInfo, + sizeof(IndexScanInstrumentation)); } diff --git a/src/backend/executor/nodeIndexscan.c b/src/backend/executor/nodeIndexscan.c index 39f6691ee35ed..5c2fcec9702a5 100644 --- a/src/backend/executor/nodeIndexscan.c +++ b/src/backend/executor/nodeIndexscan.c @@ -33,6 +33,7 @@ #include "access/relscan.h" #include "access/tableam.h" #include "catalog/pg_am.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/instrument.h" #include "executor/nodeIndexscan.h" @@ -800,25 +801,11 @@ ExecEndIndexScan(IndexScanState *node) indexScanDesc = node->iss_ScanDesc; /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE + * In a parallel worker iss_Instrument points directly at the worker's + * slot in shared memory (see ExecIndexScanInstrumentInitWorker), so the + * stats are already where the leader will read them; nothing to copy + * here. */ - if (node->iss_SharedInfo != NULL && IsParallelWorker()) - { - IndexScanInstrumentation *winstrument; - - Assert(ParallelWorkerNumber < node->iss_SharedInfo->num_workers); - winstrument = &node->iss_SharedInfo->winstrument[ParallelWorkerNumber]; - - /* - * We have to accumulate the stats rather than performing a memcpy. - * When a Gather/GatherMerge node finishes it will perform planner - * shutdown on the workers. On rescan it will spin up new workers - * which will have a new IndexOnlyScanState and zeroed stats. - */ - winstrument->nsearches += node->iss_Instrument->nsearches; - } /* * close the index relation (no-op if we didn't open it) @@ -1779,20 +1766,10 @@ void ExecIndexScanInstrumentEstimate(IndexScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - /* - * This size calculation is trivial enough that we don't bother saving it - * in the IndexScanState. We'll recalculate the needed size in - * ExecIndexScanInstrumentInitDSM(). - */ - size = add_size(offsetof(SharedIndexScanInstrumentation, winstrument), - mul_size(pcxt->nworkers, sizeof(IndexScanInstrumentation))); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(IndexScanInstrumentation)); } /* @@ -1802,23 +1779,14 @@ void ExecIndexScanInstrumentInitDSM(IndexScanState *node, ParallelContext *pcxt) { - Size size; - if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedIndexScanInstrumentation, winstrument), - mul_size(pcxt->nworkers, sizeof(IndexScanInstrumentation))); node->iss_SharedInfo = - (SharedIndexScanInstrumentation *) shm_toc_allocate(pcxt->toc, size); - - /* Each per-worker area must start out as zeroes */ - memset(node->iss_SharedInfo, 0, size); - node->iss_SharedInfo->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - node->iss_SharedInfo); + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(IndexScanInstrumentation)); } /* @@ -1831,11 +1799,25 @@ ExecIndexScanInstrumentInitWorker(IndexScanState *node, if (!node->ss.ps.instrument) return; - node->iss_SharedInfo = (SharedIndexScanInstrumentation *) - shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->iss_SharedInfo = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); + + /* + * Write statistics straight into this worker's shared slot as the scan + * runs, rather than into worker-local memory copied out at shutdown. This + * has no write-side contention (one writer per slot) and lets the leader + * observe progress mid-query. The slot is not reset here: when workers + * are relaunched (e.g. a rescanned Gather) the counts accumulate across + * rounds. + */ + Assert(ParallelWorkerNumber < node->iss_SharedInfo->num_workers); + pfree(node->iss_Instrument); + node->iss_Instrument = GetWorkerInstr(node->iss_SharedInfo, + IndexScanInstrumentation, + ParallelWorkerNumber); } /* ---------------------------------------------------------------- @@ -1847,15 +1829,8 @@ ExecIndexScanInstrumentInitWorker(IndexScanState *node, void ExecIndexScanRetrieveInstrumentation(IndexScanState *node) { - SharedIndexScanInstrumentation *SharedInfo = node->iss_SharedInfo; - size_t size; - - if (SharedInfo == NULL) - return; - /* Create a copy of SharedInfo in backend-local memory */ - size = offsetof(SharedIndexScanInstrumentation, winstrument) + - SharedInfo->num_workers * sizeof(IndexScanInstrumentation); - node->iss_SharedInfo = palloc(size); - memcpy(node->iss_SharedInfo, SharedInfo, size); + node->iss_SharedInfo = + ExecInstrRetrieve(node->iss_SharedInfo, + sizeof(IndexScanInstrumentation)); } diff --git a/src/backend/executor/nodeMemoize.c b/src/backend/executor/nodeMemoize.c index fdca97d7426f1..c5bc75d477d36 100644 --- a/src/backend/executor/nodeMemoize.c +++ b/src/backend/executor/nodeMemoize.c @@ -68,6 +68,7 @@ #include "access/htup_details.h" #include "common/hashfn.h" +#include "executor/execParallel.h" #include "executor/executor.h" #include "executor/nodeMemoize.h" #include "lib/ilist.h" @@ -425,7 +426,7 @@ cache_purge_all(MemoizeState *mstate) mstate->mem_used = 0; /* XXX should we add something new to track these purges? */ - mstate->stats.cache_evictions += evictions; /* Update Stats */ + mstate->instr->cache_evictions += evictions; /* Update Stats */ } /* @@ -445,8 +446,8 @@ cache_reduce_memory(MemoizeState *mstate, MemoizeKey *specialkey) uint64 evictions = 0; /* Update peak memory usage */ - if (mstate->mem_used > mstate->stats.mem_peak) - mstate->stats.mem_peak = mstate->mem_used; + if (mstate->mem_used > mstate->instr->mem_peak) + mstate->instr->mem_peak = mstate->mem_used; /* We expect only to be called when we've gone over budget on memory */ Assert(mstate->mem_used > mstate->mem_limit); @@ -508,7 +509,7 @@ cache_reduce_memory(MemoizeState *mstate, MemoizeKey *specialkey) break; } - mstate->stats.cache_evictions += evictions; /* Update Stats */ + mstate->instr->cache_evictions += evictions; /* Update Stats */ return specialkey_intact; } @@ -742,7 +743,7 @@ ExecMemoize(PlanState *pstate) if (found && entry->complete) { - node->stats.cache_hits += 1; /* stats update */ + node->instr->cache_hits += 1; /* stats update */ /* * Set last_tuple and entry so that the state @@ -770,7 +771,7 @@ ExecMemoize(PlanState *pstate) } /* Handle cache miss */ - node->stats.cache_misses += 1; /* stats update */ + node->instr->cache_misses += 1; /* stats update */ if (found) { @@ -813,7 +814,7 @@ ExecMemoize(PlanState *pstate) if (unlikely(entry == NULL || !cache_store_tuple(node, outerslot))) { - node->stats.cache_overflows += 1; /* stats update */ + node->instr->cache_overflows += 1; /* stats update */ node->mstatus = MEMO_CACHE_BYPASS_MODE; @@ -897,7 +898,7 @@ ExecMemoize(PlanState *pstate) if (unlikely(!cache_store_tuple(node, outerslot))) { /* Couldn't store it? Handle overflow */ - node->stats.cache_overflows += 1; /* stats update */ + node->instr->cache_overflows += 1; /* stats update */ node->mstatus = MEMO_CACHE_BYPASS_MODE; @@ -1065,8 +1066,13 @@ ExecInitMemoize(Memoize *node, EState *estate, int eflags) */ mstate->binary_mode = node->binary_mode; - /* Zero the statistics counters */ + /* + * Zero the statistics counters. Accumulate into the node's own storage + * by default; a parallel worker will redirect this to its slot in shared + * memory (see ExecMemoizeInitializeWorker). + */ memset(&mstate->stats, 0, sizeof(MemoizeInstrumentation)); + mstate->instr = &mstate->stats; /* * Because it may require a large allocation, we delay building of the @@ -1111,21 +1117,14 @@ ExecEndMemoize(MemoizeState *node) #endif /* - * When ending a parallel worker, copy the statistics gathered by the - * worker back into shared memory so that it can be picked up by the main - * process to report in EXPLAIN ANALYZE. + * In a parallel worker the statistics were accumulated directly into the + * worker's slot in shared memory (node->instr), so nothing needs to be + * copied here; just make sure mem_peak is populated for EXPLAIN. */ if (node->shared_info != NULL && IsParallelWorker()) { - MemoizeInstrumentation *si; - - /* Make mem_peak available for EXPLAIN */ - if (node->stats.mem_peak == 0) - node->stats.mem_peak = node->mem_used; - - Assert(ParallelWorkerNumber < node->shared_info->num_workers); - si = &node->shared_info->sinstrument[ParallelWorkerNumber]; - memcpy(si, &node->stats, sizeof(MemoizeInstrumentation)); + if (node->instr->mem_peak == 0) + node->instr->mem_peak = node->mem_used; } /* Remove the cache context */ @@ -1190,16 +1189,11 @@ ExecEstimateCacheEntryOverheadBytes(double ntuples) void ExecMemoizeEstimate(MemoizeState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = mul_size(pcxt->nworkers, sizeof(MemoizeInstrumentation)); - size = add_size(size, offsetof(SharedMemoizeInfo, sinstrument)); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(MemoizeInstrumentation)); } /* ---------------------------------------------------------------- @@ -1211,20 +1205,13 @@ ExecMemoizeEstimate(MemoizeState *node, ParallelContext *pcxt) void ExecMemoizeInitializeDSM(MemoizeState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedMemoizeInfo, sinstrument) - + pcxt->nworkers * sizeof(MemoizeInstrumentation); - node->shared_info = shm_toc_allocate(pcxt->toc, size); - /* ensure any unfilled slots will contain zeroes */ - memset(node->shared_info, 0, size); - node->shared_info->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, - node->shared_info); + node->shared_info = + ExecInstrInitDSM(pcxt, node->ss.ps.plan->plan_node_id, + sizeof(MemoizeInstrumentation)); } /* ---------------------------------------------------------------- @@ -1237,7 +1224,21 @@ void ExecMemoizeInitializeWorker(MemoizeState *node, ParallelWorkerContext *pwcxt) { node->shared_info = - shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + ExecInstrInitWorker(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + + /* + * Accumulate statistics straight into this worker's slot in shared memory + * as the scan runs, rather than copying them out at shutdown. Seed the + * slot from the node's initial counters; this also resets the slot if + * workers are relaunched, so each round reports its own statistics. + */ + if (node->shared_info != NULL) + { + Assert(ParallelWorkerNumber < node->shared_info->num_workers); + node->instr = GetWorkerInstr(node->shared_info, MemoizeInstrumentation, + ParallelWorkerNumber); + *node->instr = node->stats; + } } /* ---------------------------------------------------------------- @@ -1249,15 +1250,7 @@ ExecMemoizeInitializeWorker(MemoizeState *node, ParallelWorkerContext *pwcxt) void ExecMemoizeRetrieveInstrumentation(MemoizeState *node) { - Size size; - SharedMemoizeInfo *si; - - if (node->shared_info == NULL) - return; - - size = offsetof(SharedMemoizeInfo, sinstrument) - + node->shared_info->num_workers * sizeof(MemoizeInstrumentation); - si = palloc(size); - memcpy(si, node->shared_info, size); - node->shared_info = si; + node->shared_info = + ExecInstrRetrieve(node->shared_info, + sizeof(MemoizeInstrumentation)); } diff --git a/src/backend/executor/nodeSeqscan.c b/src/backend/executor/nodeSeqscan.c index 5bcb0a861d74e..197f047952cad 100644 --- a/src/backend/executor/nodeSeqscan.c +++ b/src/backend/executor/nodeSeqscan.c @@ -42,6 +42,28 @@ static TupleTableSlot *SeqNext(SeqScanState *node); * ---------------------------------------------------------------- */ +/* + * Where this scan should accumulate I/O instrumentation. + * + * In a parallel worker this is the worker's own slot in shared memory, so the + * scan writes statistics there directly as it runs (no copy at shutdown, and + * the leader can observe progress mid-query). Otherwise it's the node's own + * local storage, read back for EXPLAIN via the scan descriptor. Either way the + * executor owns the storage; the table AM never allocates instrumentation + * itself (mirroring how index_beginscan() is handed its instrumentation). + */ +static TableScanInstrumentation * +ExecSeqScanIOStatsLocation(SeqScanState *node) +{ + if (node->sinstrument != NULL && IsParallelWorker()) + { + Assert(ParallelWorkerNumber < node->sinstrument->num_workers); + return &GetWorkerInstr(node->sinstrument, SeqScanInstrumentation, + ParallelWorkerNumber)->stats; + } + return &node->stats; +} + /* ---------------------------------------------------------------- * SeqNext * @@ -78,9 +100,11 @@ SeqNext(SeqScanState *node) * We reach here if the scan is not parallel, or if we're serially * executing a scan that was planned to be parallel. */ - scandesc = table_beginscan(node->ss.ss_currentRelation, - estate->es_snapshot, - 0, NULL, flags); + scandesc = table_beginscan_instrument(node->ss.ss_currentRelation, + estate->es_snapshot, + 0, NULL, + ExecSeqScanIOStatsLocation(node), + flags); node->ss.ss_currentScanDesc = scandesc; } @@ -310,20 +334,10 @@ ExecEndSeqScan(SeqScanState *node) scanDesc = node->ss.ss_currentScanDesc; /* - * Collect I/O stats for this process into shared instrumentation. + * In a parallel worker the scan wrote its I/O statistics straight into + * the worker's shared-memory slot (see ExecSeqScanIOStatsLocation), so + * there is nothing to collect here. */ - if (node->sinstrument != NULL && IsParallelWorker()) - { - SeqScanInstrumentation *si; - - Assert(ParallelWorkerNumber < node->sinstrument->num_workers); - si = &node->sinstrument->sinstrument[ParallelWorkerNumber]; - - if (scanDesc && scanDesc->rs_instrument) - { - AccumulateIOStats(&si->stats.io, &scanDesc->rs_instrument->io); - } - } /* * close heap scan @@ -408,7 +422,8 @@ ExecSeqScanInitializeDSM(SeqScanState *node, shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, pscan); node->ss.ss_currentScanDesc = - table_beginscan_parallel(node->ss.ss_currentRelation, pscan, flags); + table_beginscan_parallel(node->ss.ss_currentRelation, pscan, + ExecSeqScanIOStatsLocation(node), flags); } /* ---------------------------------------------------------------- @@ -448,7 +463,8 @@ ExecSeqScanInitializeWorker(SeqScanState *node, pscan = shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, false); node->ss.ss_currentScanDesc = - table_beginscan_parallel(node->ss.ss_currentRelation, pscan, flags); + table_beginscan_parallel(node->ss.ss_currentRelation, pscan, + ExecSeqScanIOStatsLocation(node), flags); } /* @@ -459,16 +475,11 @@ void ExecSeqScanInstrumentEstimate(SeqScanState *node, ParallelContext *pcxt) { EState *estate = node->ss.ps.state; - Size size; if ((estate->es_instrument & INSTRUMENT_IO) == 0 || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedSeqScanInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(SeqScanInstrumentation))); - - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(SeqScanInstrumentation)); } /* @@ -478,22 +489,15 @@ void ExecSeqScanInstrumentInitDSM(SeqScanState *node, ParallelContext *pcxt) { EState *estate = node->ss.ps.state; - SharedSeqScanInstrumentation *sinstrument; - Size size; if ((estate->es_instrument & INSTRUMENT_IO) == 0 || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedSeqScanInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(SeqScanInstrumentation))); - sinstrument = shm_toc_allocate(pcxt->toc, size); - memset(sinstrument, 0, size); - sinstrument->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - sinstrument); - node->sinstrument = sinstrument; + node->sinstrument = + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(SeqScanInstrumentation)); } /* @@ -508,10 +512,11 @@ ExecSeqScanInstrumentInitWorker(SeqScanState *node, if ((estate->es_instrument & INSTRUMENT_IO) == 0) return; - node->sinstrument = shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->sinstrument = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); } /* @@ -520,15 +525,7 @@ ExecSeqScanInstrumentInitWorker(SeqScanState *node, void ExecSeqScanRetrieveInstrumentation(SeqScanState *node) { - SharedSeqScanInstrumentation *sinstrument = node->sinstrument; - Size size; - - if (sinstrument == NULL) - return; - - size = offsetof(SharedSeqScanInstrumentation, sinstrument) - + sinstrument->num_workers * sizeof(SeqScanInstrumentation); - - node->sinstrument = palloc(size); - memcpy(node->sinstrument, sinstrument, size); + node->sinstrument = + ExecInstrRetrieve(node->sinstrument, + sizeof(SeqScanInstrumentation)); } diff --git a/src/backend/executor/nodeSort.c b/src/backend/executor/nodeSort.c index e02313f7813e1..946a8149197cf 100644 --- a/src/backend/executor/nodeSort.c +++ b/src/backend/executor/nodeSort.c @@ -16,6 +16,7 @@ #include "postgres.h" #include "access/parallel.h" +#include "executor/execParallel.h" #include "executor/execdebug.h" #include "executor/nodeSort.h" #include "miscadmin.h" @@ -176,7 +177,8 @@ ExecSort(PlanState *pstate) Assert(IsParallelWorker()); Assert(ParallelWorkerNumber < node->shared_info->num_workers); - si = &node->shared_info->sinstrument[ParallelWorkerNumber]; + si = GetWorkerInstr(node->shared_info, TuplesortInstrumentation, + ParallelWorkerNumber); tuplesort_get_stats(tuplesortstate, si); } SO1_printf("ExecSort: %s\n", "sorting done"); @@ -415,16 +417,11 @@ ExecReScanSort(SortState *node) void ExecSortEstimate(SortState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = mul_size(pcxt->nworkers, sizeof(TuplesortInstrumentation)); - size = add_size(size, offsetof(SharedSortInfo, sinstrument)); - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(TuplesortInstrumentation)); } /* ---------------------------------------------------------------- @@ -436,20 +433,13 @@ ExecSortEstimate(SortState *node, ParallelContext *pcxt) void ExecSortInitializeDSM(SortState *node, ParallelContext *pcxt) { - Size size; - /* don't need this if not instrumenting or no workers */ if (!node->ss.ps.instrument || pcxt->nworkers == 0) return; - size = offsetof(SharedSortInfo, sinstrument) - + pcxt->nworkers * sizeof(TuplesortInstrumentation); - node->shared_info = shm_toc_allocate(pcxt->toc, size); - /* ensure any unfilled slots will contain zeroes */ - memset(node->shared_info, 0, size); - node->shared_info->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, - node->shared_info); + node->shared_info = + ExecInstrInitDSM(pcxt, node->ss.ps.plan->plan_node_id, + sizeof(TuplesortInstrumentation)); } /* ---------------------------------------------------------------- @@ -462,7 +452,7 @@ void ExecSortInitializeWorker(SortState *node, ParallelWorkerContext *pwcxt) { node->shared_info = - shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); + ExecInstrInitWorker(pwcxt->toc, node->ss.ps.plan->plan_node_id, true); node->am_worker = true; } @@ -475,15 +465,7 @@ ExecSortInitializeWorker(SortState *node, ParallelWorkerContext *pwcxt) void ExecSortRetrieveInstrumentation(SortState *node) { - Size size; - SharedSortInfo *si; - - if (node->shared_info == NULL) - return; - - size = offsetof(SharedSortInfo, sinstrument) - + node->shared_info->num_workers * sizeof(TuplesortInstrumentation); - si = palloc(size); - memcpy(si, node->shared_info, size); - node->shared_info = si; + node->shared_info = + ExecInstrRetrieve(node->shared_info, + sizeof(TuplesortInstrumentation)); } diff --git a/src/backend/executor/nodeTidrangescan.c b/src/backend/executor/nodeTidrangescan.c index b387ed6c30836..84a68943c1d3a 100644 --- a/src/backend/executor/nodeTidrangescan.c +++ b/src/backend/executor/nodeTidrangescan.c @@ -212,6 +212,25 @@ TidRangeEval(TidRangeScanState *node) return true; } +/* + * Where this scan should accumulate I/O instrumentation: this worker's slot in + * shared memory when run in a parallel worker (so statistics are written there + * directly), otherwise the node's own local storage. The executor always owns + * the storage; the table AM never allocates it. See the equivalent in + * nodeSeqscan.c for details. + */ +static TableScanInstrumentation * +ExecTidRangeScanIOStatsLocation(TidRangeScanState *node) +{ + if (node->trss_sinstrument != NULL && IsParallelWorker()) + { + Assert(ParallelWorkerNumber < node->trss_sinstrument->num_workers); + return &GetWorkerInstr(node->trss_sinstrument, TidRangeScanInstrumentation, + ParallelWorkerNumber)->stats; + } + return &node->stats; +} + /* ---------------------------------------------------------------- * TidRangeNext * @@ -256,6 +275,7 @@ TidRangeNext(TidRangeScanState *node) estate->es_snapshot, &node->trss_mintid, &node->trss_maxtid, + ExecTidRangeScanIOStatsLocation(node), flags); node->ss.ss_currentScanDesc = scandesc; } @@ -351,19 +371,11 @@ ExecEndTidRangeScan(TidRangeScanState *node) { TableScanDesc scan = node->ss.ss_currentScanDesc; - /* Collect IO stats for this process into shared instrumentation */ - if (node->trss_sinstrument != NULL && IsParallelWorker()) - { - TidRangeScanInstrumentation *si; - - Assert(ParallelWorkerNumber < node->trss_sinstrument->num_workers); - si = &node->trss_sinstrument->sinstrument[ParallelWorkerNumber]; - - if (scan && scan->rs_instrument) - { - AccumulateIOStats(&si->stats.io, &scan->rs_instrument->io); - } - } + /* + * In a parallel worker the scan wrote its I/O statistics straight into + * the worker's shared-memory slot (see ExecTidRangeScanIOStatsLocation), + * so there is nothing to collect here. + */ if (scan != NULL) table_endscan(scan); @@ -492,7 +504,9 @@ ExecTidRangeScanInitializeDSM(TidRangeScanState *node, ParallelContext *pcxt) shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, pscan); node->ss.ss_currentScanDesc = table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, - pscan, flags); + pscan, + ExecTidRangeScanIOStatsLocation(node), + flags); } /* ---------------------------------------------------------------- @@ -533,7 +547,9 @@ ExecTidRangeScanInitializeWorker(TidRangeScanState *node, pscan = shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, false); node->ss.ss_currentScanDesc = table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, - pscan, flags); + pscan, + ExecTidRangeScanIOStatsLocation(node), + flags); } /* @@ -545,16 +561,11 @@ ExecTidRangeScanInstrumentEstimate(TidRangeScanState *node, ParallelContext *pcxt) { EState *estate = node->ss.ps.state; - Size size; if ((estate->es_instrument & INSTRUMENT_IO) == 0 || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedTidRangeScanInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(TidRangeScanInstrumentation))); - - shm_toc_estimate_chunk(&pcxt->estimator, size); - shm_toc_estimate_keys(&pcxt->estimator, 1); + ExecInstrEstimate(pcxt, sizeof(TidRangeScanInstrumentation)); } /* @@ -565,22 +576,15 @@ ExecTidRangeScanInstrumentInitDSM(TidRangeScanState *node, ParallelContext *pcxt) { EState *estate = node->ss.ps.state; - SharedTidRangeScanInstrumentation *sinstrument; - Size size; if ((estate->es_instrument & INSTRUMENT_IO) == 0 || pcxt->nworkers == 0) return; - size = add_size(offsetof(SharedTidRangeScanInstrumentation, sinstrument), - mul_size(pcxt->nworkers, sizeof(TidRangeScanInstrumentation))); - sinstrument = shm_toc_allocate(pcxt->toc, size); - memset(sinstrument, 0, size); - sinstrument->num_workers = pcxt->nworkers; - shm_toc_insert(pcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - sinstrument); - node->trss_sinstrument = sinstrument; + node->trss_sinstrument = + ExecInstrInitDSM(pcxt, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + sizeof(TidRangeScanInstrumentation)); } /* @@ -595,10 +599,11 @@ ExecTidRangeScanInstrumentInitWorker(TidRangeScanState *node, if ((estate->es_instrument & INSTRUMENT_IO) == 0) return; - node->trss_sinstrument = shm_toc_lookup(pwcxt->toc, - node->ss.ps.plan->plan_node_id + - PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, - false); + node->trss_sinstrument = + ExecInstrInitWorker(pwcxt->toc, + node->ss.ps.plan->plan_node_id + + PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET, + false); } /* @@ -607,15 +612,7 @@ ExecTidRangeScanInstrumentInitWorker(TidRangeScanState *node, void ExecTidRangeScanRetrieveInstrumentation(TidRangeScanState *node) { - SharedTidRangeScanInstrumentation *sinstrument = node->trss_sinstrument; - Size size; - - if (sinstrument == NULL) - return; - - size = offsetof(SharedTidRangeScanInstrumentation, sinstrument) - + sinstrument->num_workers * sizeof(TidRangeScanInstrumentation); - - node->trss_sinstrument = palloc(size); - memcpy(node->trss_sinstrument, sinstrument, size); + node->trss_sinstrument = + ExecInstrRetrieve(node->trss_sinstrument, + sizeof(TidRangeScanInstrumentation)); } diff --git a/src/include/access/heapam.h b/src/include/access/heapam.h index 5176478c29583..1fa74fa934268 100644 --- a/src/include/access/heapam.h +++ b/src/include/access/heapam.h @@ -350,6 +350,7 @@ typedef struct PruneFreezeResult extern TableScanDesc heap_beginscan(Relation relation, Snapshot snapshot, int nkeys, ScanKey key, ParallelTableScanDesc parallel_scan, + struct TableScanInstrumentation *instrument, uint32 flags); extern void heap_setscanlimits(TableScanDesc sscan, BlockNumber startBlk, BlockNumber numBlks); diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index 2ea06a67a6346..8973f247f9628 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -67,7 +67,9 @@ typedef struct TableScanDescData * information */ /* - * Instrumentation counters maintained by all table AMs. + * Instrumentation counters maintained by all table AMs. This points at + * caller-provided storage (the scan node's own storage, or a parallel + * worker's slot in shared memory); the scan does not own it. */ struct TableScanInstrumentation *rs_instrument; } TableScanDescData; diff --git a/src/include/access/tableam.h b/src/include/access/tableam.h index f2c36696bcad0..9de3e2c1cb912 100644 --- a/src/include/access/tableam.h +++ b/src/include/access/tableam.h @@ -361,6 +361,7 @@ typedef struct TableAmRoutine Snapshot snapshot, int nkeys, ScanKeyData *key, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags); /* @@ -916,6 +917,7 @@ extern TupleTableSlot *table_slot_create(Relation relation, List **reglist); static TableScanDesc table_beginscan_common(Relation rel, Snapshot snapshot, int nkeys, ScanKeyData *key, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags, uint32 user_flags) { Assert((user_flags & SO_INTERNAL_FLAGS) == 0); @@ -930,7 +932,8 @@ table_beginscan_common(Relation rel, Snapshot snapshot, int nkeys, if (unlikely(TransactionIdIsValid(CheckXidAlive) && !bsysscan)) elog(ERROR, "scan started during logical decoding"); - return rel->rd_tableam->scan_begin(rel, snapshot, nkeys, key, pscan, flags); + return rel->rd_tableam->scan_begin(rel, snapshot, nkeys, key, pscan, + instrument, flags); } /* @@ -947,7 +950,29 @@ table_beginscan(Relation rel, Snapshot snapshot, SO_ALLOW_STRAT | SO_ALLOW_SYNC | SO_ALLOW_PAGEMODE; return table_beginscan_common(rel, snapshot, nkeys, key, NULL, - internal_flags, flags); + NULL, internal_flags, flags); +} + +/* + * Like table_beginscan(), but lets the caller provide the location where I/O + * instrumentation should be accumulated. When SO_SCAN_INSTRUMENT is set the + * caller must supply this (the scan node's own storage, or a parallel worker's + * slot in shared memory); the scan AM never allocates instrumentation itself. + * This is a separate entry point so that the many uninstrumented callers of + * table_beginscan() are unaffected; it mirrors the dedicated instrument + * argument that index_beginscan() already takes. + */ +static inline TableScanDesc +table_beginscan_instrument(Relation rel, Snapshot snapshot, + int nkeys, ScanKeyData *key, + struct TableScanInstrumentation *instrument, + uint32 flags) +{ + uint32 internal_flags = SO_TYPE_SEQSCAN | + SO_ALLOW_STRAT | SO_ALLOW_SYNC | SO_ALLOW_PAGEMODE; + + return table_beginscan_common(rel, snapshot, nkeys, key, NULL, + instrument, internal_flags, flags); } /* @@ -977,7 +1002,7 @@ table_beginscan_strat(Relation rel, Snapshot snapshot, flags |= SO_ALLOW_SYNC; return table_beginscan_common(rel, snapshot, nkeys, key, NULL, - flags, SO_NONE); + NULL, flags, SO_NONE); } /* @@ -990,12 +1015,13 @@ table_beginscan_strat(Relation rel, Snapshot snapshot, */ static inline TableScanDesc table_beginscan_bm(Relation rel, Snapshot snapshot, - int nkeys, ScanKeyData *key, uint32 flags) + int nkeys, ScanKeyData *key, + struct TableScanInstrumentation *instrument, uint32 flags) { uint32 internal_flags = SO_TYPE_BITMAPSCAN | SO_ALLOW_PAGEMODE; return table_beginscan_common(rel, snapshot, nkeys, key, NULL, - internal_flags, flags); + instrument, internal_flags, flags); } /* @@ -1023,7 +1049,7 @@ table_beginscan_sampling(Relation rel, Snapshot snapshot, internal_flags |= SO_ALLOW_PAGEMODE; return table_beginscan_common(rel, snapshot, nkeys, key, NULL, - internal_flags, flags); + NULL, internal_flags, flags); } /* @@ -1037,7 +1063,7 @@ table_beginscan_tid(Relation rel, Snapshot snapshot) uint32 flags = SO_TYPE_TIDSCAN; return table_beginscan_common(rel, snapshot, 0, NULL, NULL, - flags, SO_NONE); + NULL, flags, SO_NONE); } /* @@ -1051,7 +1077,7 @@ table_beginscan_analyze(Relation rel) uint32 flags = SO_TYPE_ANALYZE; return table_beginscan_common(rel, NULL, 0, NULL, NULL, - flags, SO_NONE); + NULL, flags, SO_NONE); } /* @@ -1118,13 +1144,15 @@ table_scan_getnextslot(TableScanDesc sscan, ScanDirection direction, TupleTableS static inline TableScanDesc table_beginscan_tidrange(Relation rel, Snapshot snapshot, ItemPointer mintid, - ItemPointer maxtid, uint32 flags) + ItemPointer maxtid, + struct TableScanInstrumentation *instrument, + uint32 flags) { TableScanDesc sscan; uint32 internal_flags = SO_TYPE_TIDRANGESCAN | SO_ALLOW_PAGEMODE; sscan = table_beginscan_common(rel, snapshot, 0, NULL, NULL, - internal_flags, flags); + instrument, internal_flags, flags); /* Set the range of TIDs to scan */ sscan->rs_rd->rd_tableam->scan_set_tidrange(sscan, mintid, maxtid); @@ -1202,6 +1230,7 @@ extern void table_parallelscan_initialize(Relation rel, */ extern TableScanDesc table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags); /* @@ -1215,6 +1244,7 @@ extern TableScanDesc table_beginscan_parallel(Relation relation, */ extern TableScanDesc table_beginscan_parallel_tidrange(Relation relation, ParallelTableScanDesc pscan, + struct TableScanInstrumentation *instrument, uint32 flags); /* diff --git a/src/include/executor/execParallel.h b/src/include/executor/execParallel.h index 5a2034811d563..66526f0ce5516 100644 --- a/src/include/executor/execParallel.h +++ b/src/include/executor/execParallel.h @@ -48,4 +48,13 @@ extern void ExecParallelReinitialize(PlanState *planstate, extern void ParallelQueryMain(dsm_segment *seg, shm_toc *toc); +/* Generic per-worker node instrumentation container helpers */ +extern void ExecInstrEstimate(ParallelContext *pcxt, Size elemsz); +extern SharedWorkerInstrumentation *ExecInstrInitDSM(ParallelContext *pcxt, + uint64 key, Size elemsz); +extern SharedWorkerInstrumentation *ExecInstrInitWorker(shm_toc *toc, + uint64 key, bool missing_ok); +extern SharedWorkerInstrumentation *ExecInstrRetrieve(SharedWorkerInstrumentation *shared, + Size elemsz); + #endif /* EXECPARALLEL_H */ diff --git a/src/include/executor/instrument_node.h b/src/include/executor/instrument_node.h index 4076990408ef7..ffda04f97adba 100644 --- a/src/include/executor/instrument_node.h +++ b/src/include/executor/instrument_node.h @@ -27,6 +27,48 @@ */ #define PARALLEL_KEY_SCAN_INSTRUMENT_OFFSET UINT64CONST(0xD000000000000000) +/* --------------------- + * Generic shared container for per-worker node instrumentation + * + * Many node types collect a fixed-size instrumentation struct per parallel + * worker. They all share the same shared-memory layout: a num_workers header + * followed by one slot per worker, which the worker writes directly during + * execution and the leader reads back for EXPLAIN. This container, together + * with the ExecInstr* helpers in execParallel.c, factors out that common + * plumbing. Each slot is padded to a cache line so concurrent writes from + * different workers don't share a cache line (false sharing); slots are + * therefore accessed via GetWorkerInstr(), never by plain array indexing. + * + * The per-node SharedXxx types below are thin typed views over this container. + * --------------------- + */ +typedef struct SharedWorkerInstrumentation +{ + int num_workers; + /* num_workers cache-line-padded slots follow; see GetWorkerInstr() */ +} SharedWorkerInstrumentation; + +/* DSM bytes needed for a container holding nworkers slots of elemsz bytes */ +static inline Size +SharedWorkerInstrSize(int nworkers, Size elemsz) +{ + return CACHELINEALIGN(sizeof(SharedWorkerInstrumentation)) + + (Size) nworkers * CACHELINEALIGN(elemsz); +} + +/* Address of a given worker's slot (untyped) */ +static inline void * +GetWorkerInstrSlot(SharedWorkerInstrumentation *si, Size elemsz, int worker) +{ + return (char *) si + CACHELINEALIGN(sizeof(SharedWorkerInstrumentation)) + + (Size) worker * CACHELINEALIGN(elemsz); +} + +/* Typed accessor for a worker's slot */ +#define GetWorkerInstr(si, typ, worker) \ + ((typ *) GetWorkerInstrSlot((SharedWorkerInstrumentation *) (si), \ + sizeof(typ), (worker))) + /* --------------------- * Instrumentation information for aggregate function execution * --------------------- @@ -38,15 +80,6 @@ typedef struct AggregateInstrumentation int hash_batches_used; /* batches used during entire execution */ } AggregateInstrumentation; -/* - * Shared memory container for per-worker aggregate information - */ -typedef struct SharedAggInfo -{ - int num_workers; - AggregateInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedAggInfo; - /* --------------------- * Instrumentation information about read streams and I/O @@ -107,15 +140,6 @@ typedef struct IndexScanInstrumentation uint64 nsearches; } IndexScanInstrumentation; -/* - * Shared memory container for per-worker information - */ -typedef struct SharedIndexScanInstrumentation -{ - int num_workers; - IndexScanInstrumentation winstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedIndexScanInstrumentation; - /* --------------------- * Instrumentation information for bitmap heap scans @@ -131,15 +155,6 @@ typedef struct BitmapHeapScanInstrumentation TableScanInstrumentation stats; } BitmapHeapScanInstrumentation; -/* - * Shared memory container for per-worker information - */ -typedef struct SharedBitmapHeapInstrumentation -{ - int num_workers; - BitmapHeapScanInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedBitmapHeapInstrumentation; - /* --------------------- * Instrumentation information for Memoize @@ -160,15 +175,6 @@ typedef struct MemoizeInstrumentation uint64 mem_peak; /* peak memory usage in bytes */ } MemoizeInstrumentation; -/* - * Shared memory container for per-worker memoize information - */ -typedef struct SharedMemoizeInfo -{ - int num_workers; - MemoizeInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedMemoizeInfo; - /* --------------------- * Instrumentation information for Sorts. @@ -206,15 +212,6 @@ typedef struct TuplesortInstrumentation int64 spaceUsed; /* space consumption, in kB */ } TuplesortInstrumentation; -/* - * Shared memory container for per-worker sort information - */ -typedef struct SharedSortInfo -{ - int num_workers; - TuplesortInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedSortInfo; - /* --------------------- * Instrumentation information for nodeHash.c @@ -229,15 +226,6 @@ typedef struct HashInstrumentation Size space_peak; /* peak memory usage in bytes */ } HashInstrumentation; -/* - * Shared memory container for per-worker information - */ -typedef struct SharedHashInfo -{ - int num_workers; - HashInstrumentation hinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedHashInfo; - /* --------------------- * Instrumentation information for IncrementalSort @@ -259,13 +247,6 @@ typedef struct IncrementalSortInfo IncrementalSortGroupInfo prefixsortGroupInfo; } IncrementalSortInfo; -/* Shared memory container for per-worker incremental sort information */ -typedef struct SharedIncrementalSortInfo -{ - int num_workers; - IncrementalSortInfo sinfo[FLEXIBLE_ARRAY_MEMBER]; -} SharedIncrementalSortInfo; - /* --------------------- * Instrumentation information for sequential scans @@ -276,15 +257,6 @@ typedef struct SeqScanInstrumentation TableScanInstrumentation stats; } SeqScanInstrumentation; -/* - * Shared memory container for per-worker information - */ -typedef struct SharedSeqScanInstrumentation -{ - int num_workers; - SeqScanInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedSeqScanInstrumentation; - /* * Instrumentation information for TID range scans @@ -294,13 +266,4 @@ typedef struct TidRangeScanInstrumentation TableScanInstrumentation stats; } TidRangeScanInstrumentation; -/* - * Shared memory container for per-worker information - */ -typedef struct SharedTidRangeScanInstrumentation -{ - int num_workers; - TidRangeScanInstrumentation sinstrument[FLEXIBLE_ARRAY_MEMBER]; -} SharedTidRangeScanInstrumentation; - #endif /* INSTRUMENT_NODE_H */ diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index e64fd8c7ea300..f0dd48d00b93a 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -1673,7 +1673,10 @@ typedef struct SeqScanState { ScanState ss; /* its first field is NodeTag */ Size pscan_len; /* size of parallel heap scan descriptor */ - struct SharedSeqScanInstrumentation *sinstrument; + /* Local I/O instrumentation, used by the leader or a non-parallel scan */ + TableScanInstrumentation stats; + /* SeqScanInstrumentation slots, one per worker */ + struct SharedWorkerInstrumentation *sinstrument; } SeqScanState; /* ---------------- @@ -1763,7 +1766,8 @@ typedef struct IndexScanState Relation iss_RelationDesc; struct IndexScanDescData *iss_ScanDesc; IndexScanInstrumentation *iss_Instrument; - SharedIndexScanInstrumentation *iss_SharedInfo; + /* IndexScanInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *iss_SharedInfo; /* These are needed for re-checking ORDER BY expr ordering */ pairingheap *iss_ReorderQueue; @@ -1814,7 +1818,8 @@ typedef struct IndexOnlyScanState Relation ioss_RelationDesc; struct IndexScanDescData *ioss_ScanDesc; IndexScanInstrumentation *ioss_Instrument; - SharedIndexScanInstrumentation *ioss_SharedInfo; + /* IndexScanInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *ioss_SharedInfo; TupleTableSlot *ioss_TableSlot; Buffer ioss_VMBuffer; Size ioss_PscanLen; @@ -1855,7 +1860,8 @@ typedef struct BitmapIndexScanState Relation biss_RelationDesc; struct IndexScanDescData *biss_ScanDesc; IndexScanInstrumentation *biss_Instrument; - SharedIndexScanInstrumentation *biss_SharedInfo; + /* IndexScanInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *biss_SharedInfo; } BitmapIndexScanState; @@ -1883,7 +1889,8 @@ typedef struct BitmapHeapScanState BitmapHeapScanInstrumentation stats; bool initialized; ParallelBitmapHeapState *pstate; - SharedBitmapHeapInstrumentation *sinstrument; + /* BitmapHeapScanInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *sinstrument; bool recheck; } BitmapHeapScanState; @@ -1925,7 +1932,10 @@ typedef struct TidRangeScanState ItemPointerData trss_maxtid; bool trss_inScan; Size trss_pscanlen; - struct SharedTidRangeScanInstrumentation *trss_sinstrument; + /* Local I/O instrumentation, used by the leader or a non-parallel scan */ + TableScanInstrumentation stats; + /* TidRangeScanInstrumentation slots, one per worker */ + struct SharedWorkerInstrumentation *trss_sinstrument; } TidRangeScanState; /* ---------------- @@ -2330,8 +2340,11 @@ typedef struct MemoizeState * complete after caching the first tuple. */ bool binary_mode; /* true when cache key should be compared bit * by bit, false when using hash equality ops */ - MemoizeInstrumentation stats; /* execution statistics */ - SharedMemoizeInfo *shared_info; /* statistics for parallel workers */ + MemoizeInstrumentation stats; /* local execution statistics (leader) */ + MemoizeInstrumentation *instr; /* where to accumulate stats: &stats, or a + * parallel worker's slot in shared memory */ + /* MemoizeInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *shared_info; Bitmapset *keyparamids; /* Param->paramids of expressions belonging to * param_exprs */ } MemoizeState; @@ -2366,7 +2379,8 @@ typedef struct SortState void *tuplesortstate; /* private state of tuplesort.c */ bool am_worker; /* are we a worker? */ bool datumSort; /* Datum sort instead of tuple sort? */ - SharedSortInfo *shared_info; /* one entry per worker */ + /* TuplesortInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *shared_info; } SortState; typedef enum @@ -2397,7 +2411,8 @@ typedef struct IncrementalSortState TupleTableSlot *group_pivot; TupleTableSlot *transfer_tuple; bool am_worker; /* are we a worker? */ - SharedIncrementalSortInfo *shared_info; /* one entry per worker */ + /* IncrementalSortInfo slots, one per worker */ + SharedWorkerInstrumentation *shared_info; } IncrementalSortState; /* --------------------- @@ -2490,11 +2505,12 @@ typedef struct AggState int hash_planned_partitions; /* number of partitions planned * for first pass */ double hashentrysize; /* estimate revised during execution */ - Size hash_mem_peak; /* peak hash table memory usage */ uint64 hash_ngroups_current; /* number of groups currently in * memory in all hash tables */ - uint64 hash_disk_used; /* kB of disk space used */ - int hash_batches_used; /* batches used during entire execution */ + /* Local execution statistics (leader), reported by EXPLAIN ANALYZE */ + AggregateInstrumentation stats; + /* where to accumulate stats: &stats, or a worker's slot in shared memory */ + AggregateInstrumentation *instr; AggStatePerHash perhash; /* array of per-hashtable data */ AggStatePerGroup *hash_pergroup; /* grouping set indexed array of @@ -2504,7 +2520,8 @@ typedef struct AggState #define FIELDNO_AGGSTATE_ALL_PERGROUPS 54 AggStatePerGroup *all_pergroups; /* array of first ->pergroups, than * ->hash_pergroup */ - SharedAggInfo *shared_info; /* one entry per worker */ + /* AggregateInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *shared_info; } AggState; /* ---------------- @@ -2709,7 +2726,8 @@ typedef struct HashState * shared-memory info back to local storage before DSM shutdown. The * shared_info field remains NULL in workers, or in non-parallel joins. */ - SharedHashInfo *shared_info; + /* HashInstrumentation slots, one per worker */ + SharedWorkerInstrumentation *shared_info; /* * If we are collecting hash stats, this points to an initially-zeroed diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index c5db6ca67051c..8aa326ccb766e 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2860,16 +2860,11 @@ SetToDefault SetVarReturningType_context SetupWorkerPtrType ShDependObjectInfo -SharedAggInfo -SharedBitmapHeapInstrumentation SharedBitmapState SharedDependencyObjectType SharedDependencyType SharedExecutorInstrumentation SharedFileSet -SharedHashInfo -SharedIncrementalSortInfo -SharedIndexScanInstrumentation SharedInvalCatalogMsg SharedInvalCatcacheMsg SharedInvalRelSyncMsg @@ -2879,18 +2874,15 @@ SharedInvalSmgrMsg SharedInvalSnapshotMsg SharedInvalidationMessage SharedJitInstrumentation -SharedMemoizeInfo SharedRecordTableEntry SharedRecordTableKey SharedRecordTypmodRegistry -SharedSeqScanInstrumentation -SharedSortInfo -SharedTidRangeScanInstrumentation SharedTuplestore SharedTuplestoreAccessor SharedTuplestoreChunk SharedTuplestoreParticipant SharedTypmodTableEntry +SharedWorkerInstrumentation Sharedsort ShellTypeInfo ShippableCacheEntry