Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

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

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit8b1d447

Browse files
committed
Fix parallel pg_dump/pg_restore for failure to create worker processes.
If we failed to fork a worker process, or create a communication pipefor one, WaitForTerminatingWorkers would suffer an assertion failureif assert-enabled, otherwise crash or go into an infinite loop. Thiswas a consequence of not accounting for the startup condition wherewe've not yet forked all the workers.The original bug was that ParallelBackupStart would set workerStatus toWRKR_IDLE before it had successfully forked a worker. I made thingsworse in commitb7b8cc0 by not understanding the undocumented factthat the WRKR_TERMINATED state was also meant to represent the casewhere a worker hadn't been started yet: I changed enum T_WorkerStatusso that *all* the worker slots were initially in WRKR_IDLE state. Butthis wasn't any more broken in practice, since even one slot in thewrong state would keep WaitForTerminatingWorkers from terminating.In v10 and later, introduce an explicit T_WorkerStatus value forworker-not-started, in hopes of preventing future oversights of thesame ilk. Before that, just document that WRKR_TERMINATED is supposedto cover that case (partly because it wasn't actively broken, andpartly because the enum is exposed outside parallel.c in those branches,so there's microscopically more risk involved in changing it).In all branches, introduce a WORKER_IS_RUNNING status test macroto hide which T_WorkerStatus values mean that, and be more carefulnot to access ParallelSlot fields till we're sure they're valid.Per report from Vignesh C, though this is my patch not his.Back-patch to all supported branches.Discussion:https://postgr.es/m/CALDaNm1Luv-E3sarR+-unz-BjchquHHyfP+YC+2FS2pt_J+wxg@mail.gmail.com
1 parent4b96c03 commit8b1d447

File tree

1 file changed

+17
-12
lines changed

1 file changed

+17
-12
lines changed

‎src/bin/pg_dump/parallel.c

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@
4242
*
4343
* In the master process, the workerStatus field for each worker has one of
4444
* the following values:
45+
*WRKR_NOT_STARTED: we've not yet forked this worker
4546
*WRKR_IDLE: it's waiting for a command
4647
*WRKR_WORKING: it's working on a command
4748
*WRKR_TERMINATED: process ended
@@ -74,11 +75,15 @@
7475
/* Worker process statuses */
7576
typedefenum
7677
{
78+
WRKR_NOT_STARTED=0,
7779
WRKR_IDLE,
7880
WRKR_WORKING,
7981
WRKR_TERMINATED
8082
}T_WorkerStatus;
8183

84+
#defineWORKER_IS_RUNNING(workerStatus) \
85+
((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING)
86+
8287
/*
8388
* Private per-parallel-worker state (typedef for this is in parallel.h).
8489
*
@@ -415,7 +420,9 @@ ShutdownWorkersHard(ParallelState *pstate)
415420

416421
/*
417422
* Close our write end of the sockets so that any workers waiting for
418-
* commands know they can exit.
423+
* commands know they can exit. (Note: some of the pipeWrite fields might
424+
* still be zero, if we failed to initialize all the workers. Hence, just
425+
* ignore errors here.)
419426
*/
420427
for (i=0;i<pstate->numWorkers;i++)
421428
closesocket(pstate->parallelSlot[i].pipeWrite);
@@ -489,7 +496,7 @@ WaitForTerminatingWorkers(ParallelState *pstate)
489496

490497
for (j=0;j<pstate->numWorkers;j++)
491498
{
492-
if (pstate->parallelSlot[j].workerStatus!=WRKR_TERMINATED)
499+
if (WORKER_IS_RUNNING(pstate->parallelSlot[j].workerStatus))
493500
{
494501
lpHandles[nrun]= (HANDLE)pstate->parallelSlot[j].hThread;
495502
nrun++;
@@ -925,6 +932,7 @@ ParallelBackupStart(ArchiveHandle *AH)
925932
if (AH->public.numWorkers==1)
926933
returnpstate;
927934

935+
/* Create status arrays, being sure to initialize all fields to 0 */
928936
pstate->te= (TocEntry**)
929937
pg_malloc0(pstate->numWorkers*sizeof(TocEntry*));
930938
pstate->parallelSlot= (ParallelSlot*)
@@ -974,13 +982,6 @@ ParallelBackupStart(ArchiveHandle *AH)
974982
"could not create communication channels: %s\n",
975983
strerror(errno));
976984

977-
pstate->te[i]=NULL;/* just for safety */
978-
979-
slot->workerStatus=WRKR_IDLE;
980-
slot->AH=NULL;
981-
slot->callback=NULL;
982-
slot->callback_data=NULL;
983-
984985
/* master's ends of the pipes */
985986
slot->pipeRead=pipeWM[PIPE_READ];
986987
slot->pipeWrite=pipeMW[PIPE_WRITE];
@@ -998,6 +999,7 @@ ParallelBackupStart(ArchiveHandle *AH)
998999
handle=_beginthreadex(NULL,0, (void*)&init_spawned_worker_win32,
9991000
wi,0,&(slot->threadId));
10001001
slot->hThread=handle;
1002+
slot->workerStatus=WRKR_IDLE;
10011003
#else/* !WIN32 */
10021004
pid=fork();
10031005
if (pid==0)
@@ -1042,6 +1044,7 @@ ParallelBackupStart(ArchiveHandle *AH)
10421044

10431045
/* In Master after successful fork */
10441046
slot->pid=pid;
1047+
slot->workerStatus=WRKR_IDLE;
10451048

10461049
/* close read end of Master -> Worker */
10471050
closesocket(pipeMW[PIPE_READ]);
@@ -1271,7 +1274,7 @@ GetIdleWorker(ParallelState *pstate)
12711274
}
12721275

12731276
/*
1274-
* Return true iffevery worker isin the WRKR_TERMINATED state.
1277+
* Return true iffno worker isrunning.
12751278
*/
12761279
staticbool
12771280
HasEveryWorkerTerminated(ParallelState*pstate)
@@ -1280,7 +1283,7 @@ HasEveryWorkerTerminated(ParallelState *pstate)
12801283

12811284
for (i=0;i<pstate->numWorkers;i++)
12821285
{
1283-
if (pstate->parallelSlot[i].workerStatus!=WRKR_TERMINATED)
1286+
if (WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
12841287
return false;
12851288
}
12861289
return true;
@@ -1616,7 +1619,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
16161619
FD_ZERO(&workerset);
16171620
for (i=0;i<pstate->numWorkers;i++)
16181621
{
1619-
if (pstate->parallelSlot[i].workerStatus==WRKR_TERMINATED)
1622+
if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
16201623
continue;
16211624
FD_SET(pstate->parallelSlot[i].pipeRead,&workerset);
16221625
if (pstate->parallelSlot[i].pipeRead>maxFd)
@@ -1641,6 +1644,8 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
16411644
{
16421645
char*msg;
16431646

1647+
if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
1648+
continue;
16441649
if (!FD_ISSET(pstate->parallelSlot[i].pipeRead,&workerset))
16451650
continue;
16461651

0 commit comments

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