Skip to content

Commit 418d18f

Browse files
lfittllfittl
committed
Use Instrumentation struct for parallel worker communication
This simplifies the allocations a bit, since we don't need to separately allocate WAL and buffer usage, and allows the easier addition of a planned third struct in Instrumentation. Author: Lukas Fittl <lukas@fittl.com> Reviewed-by: Discussion: https://postgr.es/m/
1 parent 6c03de9 commit 418d18f

8 files changed

Lines changed: 99 additions & 171 deletions

File tree

‎src/backend/access/brin/brin.c‎

Lines changed: 14 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -49,8 +49,7 @@
4949
#define PARALLEL_KEY_BRIN_SHARED UINT64CONST(0xB000000000000001)
5050
#define PARALLEL_KEY_TUPLESORT UINT64CONST(0xB000000000000002)
5151
#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xB000000000000003)
52-
#define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xB000000000000004)
53-
#define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xB000000000000005)
52+
#define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xB000000000000004)
5453

5554
/*
5655
* Status for index builds performed in parallel. This is allocated in a
@@ -146,8 +145,7 @@ typedef struct BrinLeader
146145
BrinShared *brinshared;
147146
Sharedsort *sharedsort;
148147
Snapshot snapshot;
149-
WalUsage *walusage;
150-
BufferUsage *bufferusage;
148+
Instrumentation *instr;
151149
} BrinLeader;
152150

153151
/*
@@ -2385,8 +2383,7 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
23852383
BrinShared *brinshared;
23862384
Sharedsort *sharedsort;
23872385
BrinLeader *brinleader = palloc0_object(BrinLeader);
2388-
WalUsage *walusage;
2389-
BufferUsage *bufferusage;
2386+
Instrumentation *instr;
23902387
bool leaderparticipates = true;
23912388
int querylen;
23922389

@@ -2428,18 +2425,14 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
24282425
shm_toc_estimate_keys(&pcxt->estimator, 2);
24292426

24302427
/*
2431-
* Estimate space for WalUsage and BufferUsage -- PARALLEL_KEY_WAL_USAGE
2432-
* and PARALLEL_KEY_BUFFER_USAGE.
2428+
* Estimate space for Instrumentation -- PARALLEL_KEY_INSTRUMENTATION.
24332429
*
24342430
* If there are no extensions loaded that care, we could skip this. We
24352431
* have no way of knowing whether anyone's looking at instrumentation, so
24362432
* do it unconditionally.
24372433
*/
24382434
shm_toc_estimate_chunk(&pcxt->estimator,
2439-
mul_size(sizeof(WalUsage), pcxt->nworkers));
2440-
shm_toc_estimate_keys(&pcxt->estimator, 1);
2441-
shm_toc_estimate_chunk(&pcxt->estimator,
2442-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
2435+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
24432436
shm_toc_estimate_keys(&pcxt->estimator, 1);
24442437

24452438
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
@@ -2512,15 +2505,12 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
25122505
}
25132506

25142507
/*
2515-
* Allocate space for each worker's WalUsage and BufferUsage; no need to
2508+
* Allocate space for each worker's Instrumentation; no need to
25162509
* initialize.
25172510
*/
2518-
walusage = shm_toc_allocate(pcxt->toc,
2519-
mul_size(sizeof(WalUsage), pcxt->nworkers));
2520-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_WAL_USAGE, walusage);
2521-
bufferusage = shm_toc_allocate(pcxt->toc,
2522-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
2523-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, bufferusage);
2511+
instr = shm_toc_allocate(pcxt->toc,
2512+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
2513+
shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION, instr);
25242514

25252515
/* Launch workers, saving status for leader/caller */
25262516
LaunchParallelWorkers(pcxt);
@@ -2531,8 +2521,7 @@ _brin_begin_parallel(BrinBuildState *buildstate, Relation heap, Relation index,
25312521
brinleader->brinshared = brinshared;
25322522
brinleader->sharedsort = sharedsort;
25332523
brinleader->snapshot = snapshot;
2534-
brinleader->walusage = walusage;
2535-
brinleader->bufferusage = bufferusage;
2524+
brinleader->instr = instr;
25362525

25372526
/* If no workers were successfully launched, back out (do serial build) */
25382527
if (pcxt->nworkers_launched == 0)
@@ -2571,7 +2560,7 @@ _brin_end_parallel(BrinLeader *brinleader, BrinBuildState *state)
25712560
* or we might get incomplete data.)
25722561
*/
25732562
for (i = 0; i < brinleader->pcxt->nworkers_launched; i++)
2574-
InstrAccumParallelQuery(&brinleader->bufferusage[i], &brinleader->walusage[i]);
2563+
InstrAccumParallelQuery(&brinleader->instr[i]);
25752564

25762565
/* Free last reference to MVCC snapshot, if one was used */
25772566
if (IsMVCCSnapshot(brinleader->snapshot))
@@ -2885,8 +2874,7 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
28852874
LOCKMODE heapLockmode;
28862875
LOCKMODE indexLockmode;
28872876
QueryInstrumentation *instr;
2888-
WalUsage *walusage;
2889-
BufferUsage *bufferusage;
2877+
Instrumentation *worker_instr;
28902878
int sortmem;
28912879

28922880
/*
@@ -2947,11 +2935,8 @@ _brin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
29472935
heapRel, indexRel, sortmem, false);
29482936

29492937
/* Report WAL/buffer usage during parallel execution */
2950-
bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
2951-
walusage = shm_toc_lookup(toc, PARALLEL_KEY_WAL_USAGE, false);
2952-
InstrEndParallelQuery(instr,
2953-
&bufferusage[ParallelWorkerNumber],
2954-
&walusage[ParallelWorkerNumber]);
2938+
worker_instr = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, false);
2939+
InstrEndParallelQuery(instr, &worker_instr[ParallelWorkerNumber]);
29552940

29562941
index_close(indexRel, indexLockmode);
29572942
table_close(heapRel, heapLockmode);

‎src/backend/access/gin/gininsert.c‎

Lines changed: 14 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,7 @@
4242
#define PARALLEL_KEY_GIN_SHARED UINT64CONST(0xB000000000000001)
4343
#define PARALLEL_KEY_TUPLESORT UINT64CONST(0xB000000000000002)
4444
#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xB000000000000003)
45-
#define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xB000000000000004)
46-
#define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xB000000000000005)
45+
#define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xB000000000000004)
4746

4847
/*
4948
* Status for index builds performed in parallel. This is allocated in a
@@ -135,8 +134,7 @@ typedef struct GinLeader
135134
GinBuildShared *ginshared;
136135
Sharedsort *sharedsort;
137136
Snapshot snapshot;
138-
WalUsage *walusage;
139-
BufferUsage *bufferusage;
137+
Instrumentation *instr;
140138
} GinLeader;
141139

142140
typedef struct
@@ -942,8 +940,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
942940
GinBuildShared *ginshared;
943941
Sharedsort *sharedsort;
944942
GinLeader *ginleader = palloc0_object(GinLeader);
945-
WalUsage *walusage;
946-
BufferUsage *bufferusage;
943+
Instrumentation *instr;
947944
bool leaderparticipates = true;
948945
int querylen;
949946

@@ -984,18 +981,14 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
984981
shm_toc_estimate_keys(&pcxt->estimator, 2);
985982

986983
/*
987-
* Estimate space for WalUsage and BufferUsage -- PARALLEL_KEY_WAL_USAGE
988-
* and PARALLEL_KEY_BUFFER_USAGE.
984+
* Estimate space for Instrumentation -- PARALLEL_KEY_INSTRUMENTATION.
989985
*
990986
* If there are no extensions loaded that care, we could skip this. We
991987
* have no way of knowing whether anyone's looking at instrumentation, so
992988
* do it unconditionally.
993989
*/
994990
shm_toc_estimate_chunk(&pcxt->estimator,
995-
mul_size(sizeof(WalUsage), pcxt->nworkers));
996-
shm_toc_estimate_keys(&pcxt->estimator, 1);
997-
shm_toc_estimate_chunk(&pcxt->estimator,
998-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
991+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
999992
shm_toc_estimate_keys(&pcxt->estimator, 1);
1000993

1001994
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
@@ -1063,15 +1056,12 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
10631056
}
10641057

10651058
/*
1066-
* Allocate space for each worker's WalUsage and BufferUsage; no need to
1059+
* Allocate space for each worker's Instrumentation; no need to
10671060
* initialize.
10681061
*/
1069-
walusage = shm_toc_allocate(pcxt->toc,
1070-
mul_size(sizeof(WalUsage), pcxt->nworkers));
1071-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_WAL_USAGE, walusage);
1072-
bufferusage = shm_toc_allocate(pcxt->toc,
1073-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
1074-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, bufferusage);
1062+
instr = shm_toc_allocate(pcxt->toc,
1063+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
1064+
shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION, instr);
10751065

10761066
/* Launch workers, saving status for leader/caller */
10771067
LaunchParallelWorkers(pcxt);
@@ -1082,8 +1072,7 @@ _gin_begin_parallel(GinBuildState *buildstate, Relation heap, Relation index,
10821072
ginleader->ginshared = ginshared;
10831073
ginleader->sharedsort = sharedsort;
10841074
ginleader->snapshot = snapshot;
1085-
ginleader->walusage = walusage;
1086-
ginleader->bufferusage = bufferusage;
1075+
ginleader->instr = instr;
10871076

10881077
/* If no workers were successfully launched, back out (do serial build) */
10891078
if (pcxt->nworkers_launched == 0)
@@ -1122,7 +1111,7 @@ _gin_end_parallel(GinLeader *ginleader, GinBuildState *state)
11221111
* or we might get incomplete data.)
11231112
*/
11241113
for (i = 0; i < ginleader->pcxt->nworkers_launched; i++)
1125-
InstrAccumParallelQuery(&ginleader->bufferusage[i], &ginleader->walusage[i]);
1114+
InstrAccumParallelQuery(&ginleader->instr[i]);
11261115

11271116
/* Free last reference to MVCC snapshot, if one was used */
11281117
if (IsMVCCSnapshot(ginleader->snapshot))
@@ -2115,8 +2104,7 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
21152104
LOCKMODE heapLockmode;
21162105
LOCKMODE indexLockmode;
21172106
QueryInstrumentation *instr;
2118-
WalUsage *walusage;
2119-
BufferUsage *bufferusage;
2107+
Instrumentation *worker_instr;
21202108
int sortmem;
21212109

21222110
/*
@@ -2196,11 +2184,8 @@ _gin_parallel_build_main(dsm_segment *seg, shm_toc *toc)
21962184
heapRel, indexRel, sortmem, false);
21972185

21982186
/* Report WAL/buffer usage during parallel execution */
2199-
bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
2200-
walusage = shm_toc_lookup(toc, PARALLEL_KEY_WAL_USAGE, false);
2201-
InstrEndParallelQuery(instr,
2202-
&bufferusage[ParallelWorkerNumber],
2203-
&walusage[ParallelWorkerNumber]);
2187+
worker_instr = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, false);
2188+
InstrEndParallelQuery(instr, &worker_instr[ParallelWorkerNumber]);
22042189

22052190
index_close(indexRel, indexLockmode);
22062191
table_close(heapRel, heapLockmode);

‎src/backend/access/nbtree/nbtsort.c‎

Lines changed: 14 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -65,8 +65,7 @@
6565
#define PARALLEL_KEY_TUPLESORT UINT64CONST(0xA000000000000002)
6666
#define PARALLEL_KEY_TUPLESORT_SPOOL2 UINT64CONST(0xA000000000000003)
6767
#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xA000000000000004)
68-
#define PARALLEL_KEY_WAL_USAGE UINT64CONST(0xA000000000000005)
69-
#define PARALLEL_KEY_BUFFER_USAGE UINT64CONST(0xA000000000000006)
68+
#define PARALLEL_KEY_INSTRUMENTATION UINT64CONST(0xA000000000000005)
7069

7170
/*
7271
* DISABLE_LEADER_PARTICIPATION disables the leader's participation in
@@ -194,8 +193,7 @@ typedef struct BTLeader
194193
Sharedsort *sharedsort;
195194
Sharedsort *sharedsort2;
196195
Snapshot snapshot;
197-
WalUsage *walusage;
198-
BufferUsage *bufferusage;
196+
Instrumentation *instr;
199197
} BTLeader;
200198

201199
/*
@@ -1407,8 +1405,7 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request)
14071405
Sharedsort *sharedsort2;
14081406
BTSpool *btspool = buildstate->spool;
14091407
BTLeader *btleader = palloc0_object(BTLeader);
1410-
WalUsage *walusage;
1411-
BufferUsage *bufferusage;
1408+
Instrumentation *instr;
14121409
bool leaderparticipates = true;
14131410
int querylen;
14141411

@@ -1461,18 +1458,14 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request)
14611458
}
14621459

14631460
/*
1464-
* Estimate space for WalUsage and BufferUsage -- PARALLEL_KEY_WAL_USAGE
1465-
* and PARALLEL_KEY_BUFFER_USAGE.
1461+
* Estimate space for Instrumentation -- PARALLEL_KEY_INSTRUMENTATION.
14661462
*
14671463
* If there are no extensions loaded that care, we could skip this. We
14681464
* have no way of knowing whether anyone's looking at instrumentation, so
14691465
* do it unconditionally.
14701466
*/
14711467
shm_toc_estimate_chunk(&pcxt->estimator,
1472-
mul_size(sizeof(WalUsage), pcxt->nworkers));
1473-
shm_toc_estimate_keys(&pcxt->estimator, 1);
1474-
shm_toc_estimate_chunk(&pcxt->estimator,
1475-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
1468+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
14761469
shm_toc_estimate_keys(&pcxt->estimator, 1);
14771470

14781471
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
@@ -1559,15 +1552,12 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request)
15591552
}
15601553

15611554
/*
1562-
* Allocate space for each worker's WalUsage and BufferUsage; no need to
1555+
* Allocate space for each worker's Instrumentation; no need to
15631556
* initialize.
15641557
*/
1565-
walusage = shm_toc_allocate(pcxt->toc,
1566-
mul_size(sizeof(WalUsage), pcxt->nworkers));
1567-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_WAL_USAGE, walusage);
1568-
bufferusage = shm_toc_allocate(pcxt->toc,
1569-
mul_size(sizeof(BufferUsage), pcxt->nworkers));
1570-
shm_toc_insert(pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, bufferusage);
1558+
instr = shm_toc_allocate(pcxt->toc,
1559+
mul_size(sizeof(Instrumentation), pcxt->nworkers));
1560+
shm_toc_insert(pcxt->toc, PARALLEL_KEY_INSTRUMENTATION, instr);
15711561

15721562
/* Launch workers, saving status for leader/caller */
15731563
LaunchParallelWorkers(pcxt);
@@ -1579,8 +1569,7 @@ _bt_begin_parallel(BTBuildState *buildstate, bool isconcurrent, int request)
15791569
btleader->sharedsort = sharedsort;
15801570
btleader->sharedsort2 = sharedsort2;
15811571
btleader->snapshot = snapshot;
1582-
btleader->walusage = walusage;
1583-
btleader->bufferusage = bufferusage;
1572+
btleader->instr = instr;
15841573

15851574
/* If no workers were successfully launched, back out (do serial build) */
15861575
if (pcxt->nworkers_launched == 0)
@@ -1619,7 +1608,7 @@ _bt_end_parallel(BTLeader *btleader)
16191608
* or we might get incomplete data.)
16201609
*/
16211610
for (i = 0; i < btleader->pcxt->nworkers_launched; i++)
1622-
InstrAccumParallelQuery(&btleader->bufferusage[i], &btleader->walusage[i]);
1611+
InstrAccumParallelQuery(&btleader->instr[i]);
16231612

16241613
/* Free last reference to MVCC snapshot, if one was used */
16251614
if (IsMVCCSnapshot(btleader->snapshot))
@@ -1753,8 +1742,7 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
17531742
LOCKMODE heapLockmode;
17541743
LOCKMODE indexLockmode;
17551744
QueryInstrumentation *instr;
1756-
WalUsage *walusage;
1757-
BufferUsage *bufferusage;
1745+
Instrumentation *worker_instr;
17581746
int sortmem;
17591747

17601748
#ifdef BTREE_BUILD_STATS
@@ -1836,11 +1824,8 @@ _bt_parallel_build_main(dsm_segment *seg, shm_toc *toc)
18361824
sharedsort2, sortmem, false);
18371825

18381826
/* Report WAL/buffer usage during parallel execution */
1839-
bufferusage = shm_toc_lookup(toc, PARALLEL_KEY_BUFFER_USAGE, false);
1840-
walusage = shm_toc_lookup(toc, PARALLEL_KEY_WAL_USAGE, false);
1841-
InstrEndParallelQuery(instr,
1842-
&bufferusage[ParallelWorkerNumber],
1843-
&walusage[ParallelWorkerNumber]);
1827+
worker_instr = shm_toc_lookup(toc, PARALLEL_KEY_INSTRUMENTATION, false);
1828+
InstrEndParallelQuery(instr, &worker_instr[ParallelWorkerNumber]);
18441829

18451830
#ifdef BTREE_BUILD_STATS
18461831
if (log_btree_build_stats)

0 commit comments

Comments
 (0)