-
Notifications
You must be signed in to change notification settings - Fork 4.4k
/
parallel.c
1640 lines (1443 loc) · 51.9 KB
/
parallel.c
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*-------------------------------------------------------------------------
*
* parallel.c
* Infrastructure for launching parallel workers
*
* Portions Copyright (c) 1996-2023, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* src/backend/access/transam/parallel.c
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/nbtree.h"
#include "access/parallel.h"
#include "access/session.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/index.h"
#include "catalog/namespace.h"
#include "catalog/pg_enum.h"
#include "catalog/storage.h"
#include "commands/async.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "executor/execParallel.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "pgstat.h"
#include "storage/ipc.h"
#include "storage/predicate.h"
#include "storage/sinval.h"
#include "storage/spin.h"
#include "tcop/tcopprot.h"
#include "utils/combocid.h"
#include "utils/guc.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/relmapper.h"
#include "utils/snapmgr.h"
#include "utils/typcache.h"
/*
* We don't want to waste a lot of memory on an error queue which, most of
* the time, will process only a handful of small messages. However, it is
* desirable to make it large enough that a typical ErrorResponse can be sent
* without blocking. That way, a worker that errors out can write the whole
* message into the queue and terminate without waiting for the user backend.
*/
#define PARALLEL_ERROR_QUEUE_SIZE 16384
/* Magic number for parallel context TOC. */
#define PARALLEL_MAGIC 0x50477c7c
/*
* Magic numbers for per-context parallel state sharing. Higher-level code
* should use smaller values, leaving these very large ones for use by this
* module.
*/
#define PARALLEL_KEY_FIXED UINT64CONST(0xFFFFFFFFFFFF0001)
#define PARALLEL_KEY_ERROR_QUEUE UINT64CONST(0xFFFFFFFFFFFF0002)
#define PARALLEL_KEY_LIBRARY UINT64CONST(0xFFFFFFFFFFFF0003)
#define PARALLEL_KEY_GUC UINT64CONST(0xFFFFFFFFFFFF0004)
#define PARALLEL_KEY_COMBO_CID UINT64CONST(0xFFFFFFFFFFFF0005)
#define PARALLEL_KEY_TRANSACTION_SNAPSHOT UINT64CONST(0xFFFFFFFFFFFF0006)
#define PARALLEL_KEY_ACTIVE_SNAPSHOT UINT64CONST(0xFFFFFFFFFFFF0007)
#define PARALLEL_KEY_TRANSACTION_STATE UINT64CONST(0xFFFFFFFFFFFF0008)
#define PARALLEL_KEY_ENTRYPOINT UINT64CONST(0xFFFFFFFFFFFF0009)
#define PARALLEL_KEY_SESSION_DSM UINT64CONST(0xFFFFFFFFFFFF000A)
#define PARALLEL_KEY_PENDING_SYNCS UINT64CONST(0xFFFFFFFFFFFF000B)
#define PARALLEL_KEY_REINDEX_STATE UINT64CONST(0xFFFFFFFFFFFF000C)
#define PARALLEL_KEY_RELMAPPER_STATE UINT64CONST(0xFFFFFFFFFFFF000D)
#define PARALLEL_KEY_UNCOMMITTEDENUMS UINT64CONST(0xFFFFFFFFFFFF000E)
#define PARALLEL_KEY_CLIENTCONNINFO UINT64CONST(0xFFFFFFFFFFFF000F)
/* Fixed-size parallel state. */
typedef struct FixedParallelState
{
/* Fixed-size state that workers must restore. */
Oid database_id;
Oid authenticated_user_id;
Oid current_user_id;
Oid outer_user_id;
Oid temp_namespace_id;
Oid temp_toast_namespace_id;
int sec_context;
bool is_superuser;
PGPROC *parallel_leader_pgproc;
pid_t parallel_leader_pid;
BackendId parallel_leader_backend_id;
TimestampTz xact_ts;
TimestampTz stmt_ts;
SerializableXactHandle serializable_xact_handle;
/* Mutex protects remaining fields. */
slock_t mutex;
/* Maximum XactLastRecEnd of any worker. */
XLogRecPtr last_xlog_end;
} FixedParallelState;
/*
* Our parallel worker number. We initialize this to -1, meaning that we are
* not a parallel worker. In parallel workers, it will be set to a value >= 0
* and < the number of workers before any user code is invoked; each parallel
* worker will get a different parallel worker number.
*/
int ParallelWorkerNumber = -1;
/* Is there a parallel message pending which we need to receive? */
volatile sig_atomic_t ParallelMessagePending = false;
/* Are we initializing a parallel worker? */
bool InitializingParallelWorker = false;
/* Pointer to our fixed parallel state. */
static FixedParallelState *MyFixedParallelState;
/* List of active parallel contexts. */
static dlist_head pcxt_list = DLIST_STATIC_INIT(pcxt_list);
/* Backend-local copy of data from FixedParallelState. */
static pid_t ParallelLeaderPid;
/*
* List of internal parallel worker entry points. We need this for
* reasons explained in LookupParallelWorkerFunction(), below.
*/
static const struct
{
const char *fn_name;
parallel_worker_main_type fn_addr;
} InternalParallelWorkers[] =
{
{
"ParallelQueryMain", ParallelQueryMain
},
{
"_bt_parallel_build_main", _bt_parallel_build_main
},
{
"parallel_vacuum_main", parallel_vacuum_main
}
};
/* Private functions. */
static void HandleParallelMessage(ParallelContext *pcxt, int i, StringInfo msg);
static void WaitForParallelWorkersToExit(ParallelContext *pcxt);
static parallel_worker_main_type LookupParallelWorkerFunction(const char *libraryname, const char *funcname);
static void ParallelWorkerShutdown(int code, Datum arg);
/*
* Establish a new parallel context. This should be done after entering
* parallel mode, and (unless there is an error) the context should be
* destroyed before exiting the current subtransaction.
*/
ParallelContext *
CreateParallelContext(const char *library_name, const char *function_name,
int nworkers)
{
MemoryContext oldcontext;
ParallelContext *pcxt;
/* It is unsafe to create a parallel context if not in parallel mode. */
Assert(IsInParallelMode());
/* Number of workers should be non-negative. */
Assert(nworkers >= 0);
/* We might be running in a short-lived memory context. */
oldcontext = MemoryContextSwitchTo(TopTransactionContext);
/* Initialize a new ParallelContext. */
pcxt = palloc0(sizeof(ParallelContext));
pcxt->subid = GetCurrentSubTransactionId();
pcxt->nworkers = nworkers;
pcxt->nworkers_to_launch = nworkers;
pcxt->library_name = pstrdup(library_name);
pcxt->function_name = pstrdup(function_name);
pcxt->error_context_stack = error_context_stack;
shm_toc_initialize_estimator(&pcxt->estimator);
dlist_push_head(&pcxt_list, &pcxt->node);
/* Restore previous memory context. */
MemoryContextSwitchTo(oldcontext);
return pcxt;
}
/*
* Establish the dynamic shared memory segment for a parallel context and
* copy state and other bookkeeping information that will be needed by
* parallel workers into it.
*/
void
InitializeParallelDSM(ParallelContext *pcxt)
{
MemoryContext oldcontext;
Size library_len = 0;
Size guc_len = 0;
Size combocidlen = 0;
Size tsnaplen = 0;
Size asnaplen = 0;
Size tstatelen = 0;
Size pendingsyncslen = 0;
Size reindexlen = 0;
Size relmapperlen = 0;
Size uncommittedenumslen = 0;
Size clientconninfolen = 0;
Size segsize = 0;
int i;
FixedParallelState *fps;
dsm_handle session_dsm_handle = DSM_HANDLE_INVALID;
Snapshot transaction_snapshot = GetTransactionSnapshot();
Snapshot active_snapshot = GetActiveSnapshot();
/* We might be running in a very short-lived memory context. */
oldcontext = MemoryContextSwitchTo(TopTransactionContext);
/* Allow space to store the fixed-size parallel state. */
shm_toc_estimate_chunk(&pcxt->estimator, sizeof(FixedParallelState));
shm_toc_estimate_keys(&pcxt->estimator, 1);
/*
* Normally, the user will have requested at least one worker process, but
* if by chance they have not, we can skip a bunch of things here.
*/
if (pcxt->nworkers > 0)
{
/* Get (or create) the per-session DSM segment's handle. */
session_dsm_handle = GetSessionDsmHandle();
/*
* If we weren't able to create a per-session DSM segment, then we can
* continue but we can't safely launch any workers because their
* record typmods would be incompatible so they couldn't exchange
* tuples.
*/
if (session_dsm_handle == DSM_HANDLE_INVALID)
pcxt->nworkers = 0;
}
if (pcxt->nworkers > 0)
{
/* Estimate space for various kinds of state sharing. */
library_len = EstimateLibraryStateSpace();
shm_toc_estimate_chunk(&pcxt->estimator, library_len);
guc_len = EstimateGUCStateSpace();
shm_toc_estimate_chunk(&pcxt->estimator, guc_len);
combocidlen = EstimateComboCIDStateSpace();
shm_toc_estimate_chunk(&pcxt->estimator, combocidlen);
if (IsolationUsesXactSnapshot())
{
tsnaplen = EstimateSnapshotSpace(transaction_snapshot);
shm_toc_estimate_chunk(&pcxt->estimator, tsnaplen);
}
asnaplen = EstimateSnapshotSpace(active_snapshot);
shm_toc_estimate_chunk(&pcxt->estimator, asnaplen);
tstatelen = EstimateTransactionStateSpace();
shm_toc_estimate_chunk(&pcxt->estimator, tstatelen);
shm_toc_estimate_chunk(&pcxt->estimator, sizeof(dsm_handle));
pendingsyncslen = EstimatePendingSyncsSpace();
shm_toc_estimate_chunk(&pcxt->estimator, pendingsyncslen);
reindexlen = EstimateReindexStateSpace();
shm_toc_estimate_chunk(&pcxt->estimator, reindexlen);
relmapperlen = EstimateRelationMapSpace();
shm_toc_estimate_chunk(&pcxt->estimator, relmapperlen);
uncommittedenumslen = EstimateUncommittedEnumsSpace();
shm_toc_estimate_chunk(&pcxt->estimator, uncommittedenumslen);
clientconninfolen = EstimateClientConnectionInfoSpace();
shm_toc_estimate_chunk(&pcxt->estimator, clientconninfolen);
/* If you add more chunks here, you probably need to add keys. */
shm_toc_estimate_keys(&pcxt->estimator, 12);
/* Estimate space need for error queues. */
StaticAssertStmt(BUFFERALIGN(PARALLEL_ERROR_QUEUE_SIZE) ==
PARALLEL_ERROR_QUEUE_SIZE,
"parallel error queue size not buffer-aligned");
shm_toc_estimate_chunk(&pcxt->estimator,
mul_size(PARALLEL_ERROR_QUEUE_SIZE,
pcxt->nworkers));
shm_toc_estimate_keys(&pcxt->estimator, 1);
/* Estimate how much we'll need for the entrypoint info. */
shm_toc_estimate_chunk(&pcxt->estimator, strlen(pcxt->library_name) +
strlen(pcxt->function_name) + 2);
shm_toc_estimate_keys(&pcxt->estimator, 1);
}
/*
* Create DSM and initialize with new table of contents. But if the user
* didn't request any workers, then don't bother creating a dynamic shared
* memory segment; instead, just use backend-private memory.
*
* Also, if we can't create a dynamic shared memory segment because the
* maximum number of segments have already been created, then fall back to
* backend-private memory, and plan not to use any workers. We hope this
* won't happen very often, but it's better to abandon the use of
* parallelism than to fail outright.
*/
segsize = shm_toc_estimate(&pcxt->estimator);
if (pcxt->nworkers > 0)
pcxt->seg = dsm_create(segsize, DSM_CREATE_NULL_IF_MAXSEGMENTS);
if (pcxt->seg != NULL)
pcxt->toc = shm_toc_create(PARALLEL_MAGIC,
dsm_segment_address(pcxt->seg),
segsize);
else
{
pcxt->nworkers = 0;
pcxt->private_memory = MemoryContextAlloc(TopMemoryContext, segsize);
pcxt->toc = shm_toc_create(PARALLEL_MAGIC, pcxt->private_memory,
segsize);
}
/* Initialize fixed-size state in shared memory. */
fps = (FixedParallelState *)
shm_toc_allocate(pcxt->toc, sizeof(FixedParallelState));
fps->database_id = MyDatabaseId;
fps->authenticated_user_id = GetAuthenticatedUserId();
fps->outer_user_id = GetCurrentRoleId();
fps->is_superuser = current_role_is_superuser;
GetUserIdAndSecContext(&fps->current_user_id, &fps->sec_context);
GetTempNamespaceState(&fps->temp_namespace_id,
&fps->temp_toast_namespace_id);
fps->parallel_leader_pgproc = MyProc;
fps->parallel_leader_pid = MyProcPid;
fps->parallel_leader_backend_id = MyBackendId;
fps->xact_ts = GetCurrentTransactionStartTimestamp();
fps->stmt_ts = GetCurrentStatementStartTimestamp();
fps->serializable_xact_handle = ShareSerializableXact();
SpinLockInit(&fps->mutex);
fps->last_xlog_end = 0;
shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps);
/* We can skip the rest of this if we're not budgeting for any workers. */
if (pcxt->nworkers > 0)
{
char *libraryspace;
char *gucspace;
char *combocidspace;
char *tsnapspace;
char *asnapspace;
char *tstatespace;
char *pendingsyncsspace;
char *reindexspace;
char *relmapperspace;
char *error_queue_space;
char *session_dsm_handle_space;
char *entrypointstate;
char *uncommittedenumsspace;
char *clientconninfospace;
Size lnamelen;
/* Serialize shared libraries we have loaded. */
libraryspace = shm_toc_allocate(pcxt->toc, library_len);
SerializeLibraryState(library_len, libraryspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_LIBRARY, libraryspace);
/* Serialize GUC settings. */
gucspace = shm_toc_allocate(pcxt->toc, guc_len);
SerializeGUCState(guc_len, gucspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_GUC, gucspace);
/* Serialize combo CID state. */
combocidspace = shm_toc_allocate(pcxt->toc, combocidlen);
SerializeComboCIDState(combocidlen, combocidspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_COMBO_CID, combocidspace);
/*
* Serialize the transaction snapshot if the transaction isolation
* level uses a transaction snapshot.
*/
if (IsolationUsesXactSnapshot())
{
tsnapspace = shm_toc_allocate(pcxt->toc, tsnaplen);
SerializeSnapshot(transaction_snapshot, tsnapspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_SNAPSHOT,
tsnapspace);
}
/* Serialize the active snapshot. */
asnapspace = shm_toc_allocate(pcxt->toc, asnaplen);
SerializeSnapshot(active_snapshot, asnapspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_ACTIVE_SNAPSHOT, asnapspace);
/* Provide the handle for per-session segment. */
session_dsm_handle_space = shm_toc_allocate(pcxt->toc,
sizeof(dsm_handle));
*(dsm_handle *) session_dsm_handle_space = session_dsm_handle;
shm_toc_insert(pcxt->toc, PARALLEL_KEY_SESSION_DSM,
session_dsm_handle_space);
/* Serialize transaction state. */
tstatespace = shm_toc_allocate(pcxt->toc, tstatelen);
SerializeTransactionState(tstatelen, tstatespace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_TRANSACTION_STATE, tstatespace);
/* Serialize pending syncs. */
pendingsyncsspace = shm_toc_allocate(pcxt->toc, pendingsyncslen);
SerializePendingSyncs(pendingsyncslen, pendingsyncsspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_PENDING_SYNCS,
pendingsyncsspace);
/* Serialize reindex state. */
reindexspace = shm_toc_allocate(pcxt->toc, reindexlen);
SerializeReindexState(reindexlen, reindexspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_REINDEX_STATE, reindexspace);
/* Serialize relmapper state. */
relmapperspace = shm_toc_allocate(pcxt->toc, relmapperlen);
SerializeRelationMap(relmapperlen, relmapperspace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_RELMAPPER_STATE,
relmapperspace);
/* Serialize uncommitted enum state. */
uncommittedenumsspace = shm_toc_allocate(pcxt->toc,
uncommittedenumslen);
SerializeUncommittedEnums(uncommittedenumsspace, uncommittedenumslen);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_UNCOMMITTEDENUMS,
uncommittedenumsspace);
/* Serialize our ClientConnectionInfo. */
clientconninfospace = shm_toc_allocate(pcxt->toc, clientconninfolen);
SerializeClientConnectionInfo(clientconninfolen, clientconninfospace);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_CLIENTCONNINFO,
clientconninfospace);
/* Allocate space for worker information. */
pcxt->worker = palloc0(sizeof(ParallelWorkerInfo) * pcxt->nworkers);
/*
* Establish error queues in dynamic shared memory.
*
* These queues should be used only for transmitting ErrorResponse,
* NoticeResponse, and NotifyResponse protocol messages. Tuple data
* should be transmitted via separate (possibly larger?) queues.
*/
error_queue_space =
shm_toc_allocate(pcxt->toc,
mul_size(PARALLEL_ERROR_QUEUE_SIZE,
pcxt->nworkers));
for (i = 0; i < pcxt->nworkers; ++i)
{
char *start;
shm_mq *mq;
start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
shm_mq_set_receiver(mq, MyProc);
pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
}
shm_toc_insert(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, error_queue_space);
/*
* Serialize entrypoint information. It's unsafe to pass function
* pointers across processes, as the function pointer may be different
* in each process in EXEC_BACKEND builds, so we always pass library
* and function name. (We use library name "postgres" for functions
* in the core backend.)
*/
lnamelen = strlen(pcxt->library_name);
entrypointstate = shm_toc_allocate(pcxt->toc, lnamelen +
strlen(pcxt->function_name) + 2);
strcpy(entrypointstate, pcxt->library_name);
strcpy(entrypointstate + lnamelen + 1, pcxt->function_name);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_ENTRYPOINT, entrypointstate);
}
/* Restore previous memory context. */
MemoryContextSwitchTo(oldcontext);
}
/*
* Reinitialize the dynamic shared memory segment for a parallel context such
* that we could launch workers for it again.
*/
void
ReinitializeParallelDSM(ParallelContext *pcxt)
{
FixedParallelState *fps;
/* Wait for any old workers to exit. */
if (pcxt->nworkers_launched > 0)
{
WaitForParallelWorkersToFinish(pcxt);
WaitForParallelWorkersToExit(pcxt);
pcxt->nworkers_launched = 0;
if (pcxt->known_attached_workers)
{
pfree(pcxt->known_attached_workers);
pcxt->known_attached_workers = NULL;
pcxt->nknown_attached_workers = 0;
}
}
/* Reset a few bits of fixed parallel state to a clean state. */
fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
fps->last_xlog_end = 0;
/* Recreate error queues (if they exist). */
if (pcxt->nworkers > 0)
{
char *error_queue_space;
int i;
error_queue_space =
shm_toc_lookup(pcxt->toc, PARALLEL_KEY_ERROR_QUEUE, false);
for (i = 0; i < pcxt->nworkers; ++i)
{
char *start;
shm_mq *mq;
start = error_queue_space + i * PARALLEL_ERROR_QUEUE_SIZE;
mq = shm_mq_create(start, PARALLEL_ERROR_QUEUE_SIZE);
shm_mq_set_receiver(mq, MyProc);
pcxt->worker[i].error_mqh = shm_mq_attach(mq, pcxt->seg, NULL);
}
}
}
/*
* Reinitialize parallel workers for a parallel context such that we could
* launch a different number of workers. This is required for cases where
* we need to reuse the same DSM segment, but the number of workers can
* vary from run-to-run.
*/
void
ReinitializeParallelWorkers(ParallelContext *pcxt, int nworkers_to_launch)
{
/*
* The number of workers that need to be launched must be less than the
* number of workers with which the parallel context is initialized.
*/
Assert(pcxt->nworkers >= nworkers_to_launch);
pcxt->nworkers_to_launch = nworkers_to_launch;
}
/*
* Launch parallel workers.
*/
void
LaunchParallelWorkers(ParallelContext *pcxt)
{
MemoryContext oldcontext;
BackgroundWorker worker;
int i;
bool any_registrations_failed = false;
/* Skip this if we have no workers. */
if (pcxt->nworkers == 0 || pcxt->nworkers_to_launch == 0)
return;
/* We need to be a lock group leader. */
BecomeLockGroupLeader();
/* If we do have workers, we'd better have a DSM segment. */
Assert(pcxt->seg != NULL);
/* We might be running in a short-lived memory context. */
oldcontext = MemoryContextSwitchTo(TopTransactionContext);
/* Configure a worker. */
memset(&worker, 0, sizeof(worker));
snprintf(worker.bgw_name, BGW_MAXLEN, "parallel worker for PID %d",
MyProcPid);
snprintf(worker.bgw_type, BGW_MAXLEN, "parallel worker");
worker.bgw_flags =
BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION
| BGWORKER_CLASS_PARALLEL;
worker.bgw_start_time = BgWorkerStart_ConsistentState;
worker.bgw_restart_time = BGW_NEVER_RESTART;
sprintf(worker.bgw_library_name, "postgres");
sprintf(worker.bgw_function_name, "ParallelWorkerMain");
worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(pcxt->seg));
worker.bgw_notify_pid = MyProcPid;
/*
* Start workers.
*
* The caller must be able to tolerate ending up with fewer workers than
* expected, so there is no need to throw an error here if registration
* fails. It wouldn't help much anyway, because registering the worker in
* no way guarantees that it will start up and initialize successfully.
*/
for (i = 0; i < pcxt->nworkers_to_launch; ++i)
{
memcpy(worker.bgw_extra, &i, sizeof(int));
if (!any_registrations_failed &&
RegisterDynamicBackgroundWorker(&worker,
&pcxt->worker[i].bgwhandle))
{
shm_mq_set_handle(pcxt->worker[i].error_mqh,
pcxt->worker[i].bgwhandle);
pcxt->nworkers_launched++;
}
else
{
/*
* If we weren't able to register the worker, then we've bumped up
* against the max_worker_processes limit, and future
* registrations will probably fail too, so arrange to skip them.
* But we still have to execute this code for the remaining slots
* to make sure that we forget about the error queues we budgeted
* for those workers. Otherwise, we'll wait for them to start,
* but they never will.
*/
any_registrations_failed = true;
pcxt->worker[i].bgwhandle = NULL;
shm_mq_detach(pcxt->worker[i].error_mqh);
pcxt->worker[i].error_mqh = NULL;
}
}
/*
* Now that nworkers_launched has taken its final value, we can initialize
* known_attached_workers.
*/
if (pcxt->nworkers_launched > 0)
{
pcxt->known_attached_workers =
palloc0(sizeof(bool) * pcxt->nworkers_launched);
pcxt->nknown_attached_workers = 0;
}
/* Restore previous memory context. */
MemoryContextSwitchTo(oldcontext);
}
/*
* Wait for all workers to attach to their error queues, and throw an error if
* any worker fails to do this.
*
* Callers can assume that if this function returns successfully, then the
* number of workers given by pcxt->nworkers_launched have initialized and
* attached to their error queues. Whether or not these workers are guaranteed
* to still be running depends on what code the caller asked them to run;
* this function does not guarantee that they have not exited. However, it
* does guarantee that any workers which exited must have done so cleanly and
* after successfully performing the work with which they were tasked.
*
* If this function is not called, then some of the workers that were launched
* may not have been started due to a fork() failure, or may have exited during
* early startup prior to attaching to the error queue, so nworkers_launched
* cannot be viewed as completely reliable. It will never be less than the
* number of workers which actually started, but it might be more. Any workers
* that failed to start will still be discovered by
* WaitForParallelWorkersToFinish and an error will be thrown at that time,
* provided that function is eventually reached.
*
* In general, the leader process should do as much work as possible before
* calling this function. fork() failures and other early-startup failures
* are very uncommon, and having the leader sit idle when it could be doing
* useful work is undesirable. However, if the leader needs to wait for
* all of its workers or for a specific worker, it may want to call this
* function before doing so. If not, it must make some other provision for
* the failure-to-start case, lest it wait forever. On the other hand, a
* leader which never waits for a worker that might not be started yet, or
* at least never does so prior to WaitForParallelWorkersToFinish(), need not
* call this function at all.
*/
void
WaitForParallelWorkersToAttach(ParallelContext *pcxt)
{
int i;
/* Skip this if we have no launched workers. */
if (pcxt->nworkers_launched == 0)
return;
for (;;)
{
/*
* This will process any parallel messages that are pending and it may
* also throw an error propagated from a worker.
*/
CHECK_FOR_INTERRUPTS();
for (i = 0; i < pcxt->nworkers_launched; ++i)
{
BgwHandleStatus status;
shm_mq *mq;
int rc;
pid_t pid;
if (pcxt->known_attached_workers[i])
continue;
/*
* If error_mqh is NULL, then the worker has already exited
* cleanly.
*/
if (pcxt->worker[i].error_mqh == NULL)
{
pcxt->known_attached_workers[i] = true;
++pcxt->nknown_attached_workers;
continue;
}
status = GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle, &pid);
if (status == BGWH_STARTED)
{
/* Has the worker attached to the error queue? */
mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
if (shm_mq_get_sender(mq) != NULL)
{
/* Yes, so it is known to be attached. */
pcxt->known_attached_workers[i] = true;
++pcxt->nknown_attached_workers;
}
}
else if (status == BGWH_STOPPED)
{
/*
* If the worker stopped without attaching to the error queue,
* throw an error.
*/
mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
if (shm_mq_get_sender(mq) == NULL)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("parallel worker failed to initialize"),
errhint("More details may be available in the server log.")));
pcxt->known_attached_workers[i] = true;
++pcxt->nknown_attached_workers;
}
else
{
/*
* Worker not yet started, so we must wait. The postmaster
* will notify us if the worker's state changes. Our latch
* might also get set for some other reason, but if so we'll
* just end up waiting for the same worker again.
*/
rc = WaitLatch(MyLatch,
WL_LATCH_SET | WL_EXIT_ON_PM_DEATH,
-1, WAIT_EVENT_BGWORKER_STARTUP);
if (rc & WL_LATCH_SET)
ResetLatch(MyLatch);
}
}
/* If all workers are known to have started, we're done. */
if (pcxt->nknown_attached_workers >= pcxt->nworkers_launched)
{
Assert(pcxt->nknown_attached_workers == pcxt->nworkers_launched);
break;
}
}
}
/*
* Wait for all workers to finish computing.
*
* Even if the parallel operation seems to have completed successfully, it's
* important to call this function afterwards. We must not miss any errors
* the workers may have thrown during the parallel operation, or any that they
* may yet throw while shutting down.
*
* Also, we want to update our notion of XactLastRecEnd based on worker
* feedback.
*/
void
WaitForParallelWorkersToFinish(ParallelContext *pcxt)
{
for (;;)
{
bool anyone_alive = false;
int nfinished = 0;
int i;
/*
* This will process any parallel messages that are pending, which may
* change the outcome of the loop that follows. It may also throw an
* error propagated from a worker.
*/
CHECK_FOR_INTERRUPTS();
for (i = 0; i < pcxt->nworkers_launched; ++i)
{
/*
* If error_mqh is NULL, then the worker has already exited
* cleanly. If we have received a message through error_mqh from
* the worker, we know it started up cleanly, and therefore we're
* certain to be notified when it exits.
*/
if (pcxt->worker[i].error_mqh == NULL)
++nfinished;
else if (pcxt->known_attached_workers[i])
{
anyone_alive = true;
break;
}
}
if (!anyone_alive)
{
/* If all workers are known to have finished, we're done. */
if (nfinished >= pcxt->nworkers_launched)
{
Assert(nfinished == pcxt->nworkers_launched);
break;
}
/*
* We didn't detect any living workers, but not all workers are
* known to have exited cleanly. Either not all workers have
* launched yet, or maybe some of them failed to start or
* terminated abnormally.
*/
for (i = 0; i < pcxt->nworkers_launched; ++i)
{
pid_t pid;
shm_mq *mq;
/*
* If the worker is BGWH_NOT_YET_STARTED or BGWH_STARTED, we
* should just keep waiting. If it is BGWH_STOPPED, then
* further investigation is needed.
*/
if (pcxt->worker[i].error_mqh == NULL ||
pcxt->worker[i].bgwhandle == NULL ||
GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle,
&pid) != BGWH_STOPPED)
continue;
/*
* Check whether the worker ended up stopped without ever
* attaching to the error queue. If so, the postmaster was
* unable to fork the worker or it exited without initializing
* properly. We must throw an error, since the caller may
* have been expecting the worker to do some work before
* exiting.
*/
mq = shm_mq_get_queue(pcxt->worker[i].error_mqh);
if (shm_mq_get_sender(mq) == NULL)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("parallel worker failed to initialize"),
errhint("More details may be available in the server log.")));
/*
* The worker is stopped, but is attached to the error queue.
* Unless there's a bug somewhere, this will only happen when
* the worker writes messages and terminates after the
* CHECK_FOR_INTERRUPTS() near the top of this function and
* before the call to GetBackgroundWorkerPid(). In that case,
* or latch should have been set as well and the right things
* will happen on the next pass through the loop.
*/
}
}
(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1,
WAIT_EVENT_PARALLEL_FINISH);
ResetLatch(MyLatch);
}
if (pcxt->toc != NULL)
{
FixedParallelState *fps;
fps = shm_toc_lookup(pcxt->toc, PARALLEL_KEY_FIXED, false);
if (fps->last_xlog_end > XactLastRecEnd)
XactLastRecEnd = fps->last_xlog_end;
}
}
/*
* Wait for all workers to exit.
*
* This function ensures that workers have been completely shutdown. The
* difference between WaitForParallelWorkersToFinish and this function is
* that the former just ensures that last message sent by a worker backend is
* received by the leader backend whereas this ensures the complete shutdown.
*/
static void
WaitForParallelWorkersToExit(ParallelContext *pcxt)
{
int i;
/* Wait until the workers actually die. */
for (i = 0; i < pcxt->nworkers_launched; ++i)
{
BgwHandleStatus status;
if (pcxt->worker == NULL || pcxt->worker[i].bgwhandle == NULL)
continue;
status = WaitForBackgroundWorkerShutdown(pcxt->worker[i].bgwhandle);
/*
* If the postmaster kicked the bucket, we have no chance of cleaning
* up safely -- we won't be able to tell when our workers are actually
* dead. This doesn't necessitate a PANIC since they will all abort
* eventually, but we can't safely continue this session.
*/
if (status == BGWH_POSTMASTER_DIED)
ereport(FATAL,
(errcode(ERRCODE_ADMIN_SHUTDOWN),
errmsg("postmaster exited during a parallel transaction")));
/* Release memory. */
pfree(pcxt->worker[i].bgwhandle);
pcxt->worker[i].bgwhandle = NULL;
}
}
/*
* Destroy a parallel context.
*
* If expecting a clean exit, you should use WaitForParallelWorkersToFinish()
* first, before calling this function. When this function is invoked, any
* remaining workers are forcibly killed; the dynamic shared memory segment
* is unmapped; and we then wait (uninterruptibly) for the workers to exit.
*/
void
DestroyParallelContext(ParallelContext *pcxt)
{
int i;
/*
* Be careful about order of operations here! We remove the parallel
* context from the list before we do anything else; otherwise, if an
* error occurs during a subsequent step, we might try to nuke it again
* from AtEOXact_Parallel or AtEOSubXact_Parallel.
*/
dlist_delete(&pcxt->node);
/* Kill each worker in turn, and forget their error queues. */
if (pcxt->worker != NULL)
{
for (i = 0; i < pcxt->nworkers_launched; ++i)
{
if (pcxt->worker[i].error_mqh != NULL)
{
TerminateBackgroundWorker(pcxt->worker[i].bgwhandle);
shm_mq_detach(pcxt->worker[i].error_mqh);
pcxt->worker[i].error_mqh = NULL;
}
}
}
/*
* If we have allocated a shared memory segment, detach it. This will
* implicitly detach the error queues, and any other shared memory queues,
* stored there.
*/
if (pcxt->seg != NULL)
{
dsm_detach(pcxt->seg);
pcxt->seg = NULL;
}
/*
* If this parallel context is actually in backend-private memory rather
* than shared memory, free that memory instead.
*/
if (pcxt->private_memory != NULL)
{
pfree(pcxt->private_memory);
pcxt->private_memory = NULL;
}
/*
* We can't finish transaction commit or abort until all of the workers
* have exited. This means, in particular, that we can't respond to
* interrupts at this stage.
*/
HOLD_INTERRUPTS();
WaitForParallelWorkersToExit(pcxt);
RESUME_INTERRUPTS();
/* Free the worker array itself. */
if (pcxt->worker != NULL)
{
pfree(pcxt->worker);
pcxt->worker = NULL;
}
/* Free memory. */
pfree(pcxt->library_name);
pfree(pcxt->function_name);
pfree(pcxt);
}
/*
* Are there any parallel contexts currently active?
*/