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

Commit6363488

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 parent8b29c75 commit6363488

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
@@ -76,11 +77,15 @@
7677
/* Worker process statuses */
7778
typedefenum
7879
{
80+
WRKR_NOT_STARTED=0,
7981
WRKR_IDLE,
8082
WRKR_WORKING,
8183
WRKR_TERMINATED
8284
}T_WorkerStatus;
8385

86+
#defineWORKER_IS_RUNNING(workerStatus) \
87+
((workerStatus) == WRKR_IDLE || (workerStatus) == WRKR_WORKING)
88+
8489
/*
8590
* Private per-parallel-worker state (typedef for this is in parallel.h).
8691
*
@@ -417,7 +422,9 @@ ShutdownWorkersHard(ParallelState *pstate)
417422

418423
/*
419424
* Close our write end of the sockets so that any workers waiting for
420-
* commands know they can exit.
425+
* commands know they can exit. (Note: some of the pipeWrite fields might
426+
* still be zero, if we failed to initialize all the workers. Hence, just
427+
* ignore errors here.)
421428
*/
422429
for (i=0;i<pstate->numWorkers;i++)
423430
closesocket(pstate->parallelSlot[i].pipeWrite);
@@ -491,7 +498,7 @@ WaitForTerminatingWorkers(ParallelState *pstate)
491498

492499
for (j=0;j<pstate->numWorkers;j++)
493500
{
494-
if (pstate->parallelSlot[j].workerStatus!=WRKR_TERMINATED)
501+
if (WORKER_IS_RUNNING(pstate->parallelSlot[j].workerStatus))
495502
{
496503
lpHandles[nrun]= (HANDLE)pstate->parallelSlot[j].hThread;
497504
nrun++;
@@ -927,6 +934,7 @@ ParallelBackupStart(ArchiveHandle *AH)
927934
if (AH->public.numWorkers==1)
928935
returnpstate;
929936

937+
/* Create status arrays, being sure to initialize all fields to 0 */
930938
pstate->te= (TocEntry**)
931939
pg_malloc0(pstate->numWorkers*sizeof(TocEntry*));
932940
pstate->parallelSlot= (ParallelSlot*)
@@ -976,13 +984,6 @@ ParallelBackupStart(ArchiveHandle *AH)
976984
"could not create communication channels: %s\n",
977985
strerror(errno));
978986

979-
pstate->te[i]=NULL;/* just for safety */
980-
981-
slot->workerStatus=WRKR_IDLE;
982-
slot->AH=NULL;
983-
slot->callback=NULL;
984-
slot->callback_data=NULL;
985-
986987
/* master's ends of the pipes */
987988
slot->pipeRead=pipeWM[PIPE_READ];
988989
slot->pipeWrite=pipeMW[PIPE_WRITE];
@@ -1000,6 +1001,7 @@ ParallelBackupStart(ArchiveHandle *AH)
10001001
handle=_beginthreadex(NULL,0, (void*)&init_spawned_worker_win32,
10011002
wi,0,&(slot->threadId));
10021003
slot->hThread=handle;
1004+
slot->workerStatus=WRKR_IDLE;
10031005
#else/* !WIN32 */
10041006
pid=fork();
10051007
if (pid==0)
@@ -1044,6 +1046,7 @@ ParallelBackupStart(ArchiveHandle *AH)
10441046

10451047
/* In Master after successful fork */
10461048
slot->pid=pid;
1049+
slot->workerStatus=WRKR_IDLE;
10471050

10481051
/* close read end of Master -> Worker */
10491052
closesocket(pipeMW[PIPE_READ]);
@@ -1273,7 +1276,7 @@ GetIdleWorker(ParallelState *pstate)
12731276
}
12741277

12751278
/*
1276-
* Return true iffevery worker isin the WRKR_TERMINATED state.
1279+
* Return true iffno worker isrunning.
12771280
*/
12781281
staticbool
12791282
HasEveryWorkerTerminated(ParallelState*pstate)
@@ -1282,7 +1285,7 @@ HasEveryWorkerTerminated(ParallelState *pstate)
12821285

12831286
for (i=0;i<pstate->numWorkers;i++)
12841287
{
1285-
if (pstate->parallelSlot[i].workerStatus!=WRKR_TERMINATED)
1288+
if (WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
12861289
return false;
12871290
}
12881291
return true;
@@ -1618,7 +1621,7 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
16181621
FD_ZERO(&workerset);
16191622
for (i=0;i<pstate->numWorkers;i++)
16201623
{
1621-
if (pstate->parallelSlot[i].workerStatus==WRKR_TERMINATED)
1624+
if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
16221625
continue;
16231626
FD_SET(pstate->parallelSlot[i].pipeRead,&workerset);
16241627
if (pstate->parallelSlot[i].pipeRead>maxFd)
@@ -1643,6 +1646,8 @@ getMessageFromWorker(ParallelState *pstate, bool do_wait, int *worker)
16431646
{
16441647
char*msg;
16451648

1649+
if (!WORKER_IS_RUNNING(pstate->parallelSlot[i].workerStatus))
1650+
continue;
16461651
if (!FD_ISSET(pstate->parallelSlot[i].pipeRead,&workerset))
16471652
continue;
16481653

0 commit comments

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