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

Commit383e426

Browse files
committed
Report an ERROR if a parallel worker fails to start properly.
Commit28724fd fixed things so thatif a background worker fails to start due to fork() failure or becauseit is terminated before startup succeeds, BGWH_STOPPED will bereported. However, that only helps if the code that uses thebackground worker machinery notices the change in status, and the codein parallel.c did not.To fix that, do two things. First, make sure that when a workerexits, it triggers the leader to read from error queues. That way, ifa worker which has attached to an error queue exits uncleanly, theleader is sure to throw some error, either the contents of theErrorResponse sent by the worker, or "lost connection to parallelworker" if it exited without sending one. To cover the case wherethe worker never starts up in the first place or exits beforeattaching to the error queue, the ParallelContext now keeps trackof which workers have sent at least one message via the errorqueue. A worker which sends no messages by the time the paralleloperation finishes will be checked to see whether it exited beforeattaching to the error queue; if so, a new error message, "parallelworker failed to initialize", will be reported. If not, we'llcontinue to wait until it either starts up and exits cleanly, startsup and exits uncleanly, or fails to start, and then take theappropriate action.Patch by me, reviewed by Amit Kapila.Discussion:http://postgr.es/m/CA+TgmoYnBgXgdTu6wk5YPdWhmgabYc9nY_pFLq=tB=FSLYkD8Q@mail.gmail.com
1 parent5aaa866 commit383e426

File tree

2 files changed

+110
-9
lines changed

2 files changed

+110
-9
lines changed

‎src/backend/access/transam/parallel.c

Lines changed: 109 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,9 @@ static FixedParallelState *MyFixedParallelState;
107107
/* List of active parallel contexts. */
108108
staticdlist_headpcxt_list=DLIST_STATIC_INIT(pcxt_list);
109109

110+
/* Backend-local copy of data from FixedParallelState. */
111+
staticpid_tParallelMasterPid;
112+
110113
/*
111114
* List of internal parallel worker entry points. We need this for
112115
* reasons explained in LookupParallelWorkerFunction(), below.
@@ -127,6 +130,7 @@ static const struct
127130
staticvoidHandleParallelMessage(ParallelContext*pcxt,inti,StringInfomsg);
128131
staticvoidWaitForParallelWorkersToExit(ParallelContext*pcxt);
129132
staticparallel_worker_main_typeLookupParallelWorkerFunction(constchar*libraryname,constchar*funcname);
133+
staticvoidParallelWorkerShutdown(intcode,Datumarg);
130134

131135

132136
/*
@@ -393,6 +397,11 @@ ReinitializeParallelDSM(ParallelContext *pcxt)
393397
WaitForParallelWorkersToFinish(pcxt);
394398
WaitForParallelWorkersToExit(pcxt);
395399
pcxt->nworkers_launched=0;
400+
if (pcxt->any_message_received)
401+
{
402+
pfree(pcxt->any_message_received);
403+
pcxt->any_message_received=NULL;
404+
}
396405
}
397406

398407
/* Reset a few bits of fixed parallel state to a clean state. */
@@ -490,6 +499,14 @@ LaunchParallelWorkers(ParallelContext *pcxt)
490499
}
491500
}
492501

502+
/*
503+
* Now that nworkers_launched has taken its final value, we can initialize
504+
* any_message_received.
505+
*/
506+
if (pcxt->nworkers_launched>0)
507+
pcxt->any_message_received=
508+
palloc0(sizeof(bool)*pcxt->nworkers_launched);
509+
493510
/* Restore previous memory context. */
494511
MemoryContextSwitchTo(oldcontext);
495512
}
@@ -511,6 +528,7 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt)
511528
for (;;)
512529
{
513530
boolanyone_alive= false;
531+
intnfinished=0;
514532
inti;
515533

516534
/*
@@ -522,15 +540,78 @@ WaitForParallelWorkersToFinish(ParallelContext *pcxt)
522540

523541
for (i=0;i<pcxt->nworkers_launched;++i)
524542
{
525-
if (pcxt->worker[i].error_mqh!=NULL)
543+
/*
544+
* If error_mqh is NULL, then the worker has already exited
545+
* cleanly. If we have received a message through error_mqh from
546+
* the worker, we know it started up cleanly, and therefore we're
547+
* certain to be notified when it exits.
548+
*/
549+
if (pcxt->worker[i].error_mqh==NULL)
550+
++nfinished;
551+
elseif (pcxt->any_message_received[i])
526552
{
527553
anyone_alive= true;
528554
break;
529555
}
530556
}
531557

532558
if (!anyone_alive)
533-
break;
559+
{
560+
/* If all workers are known to have finished, we're done. */
561+
if (nfinished >=pcxt->nworkers_launched)
562+
{
563+
Assert(nfinished==pcxt->nworkers_launched);
564+
break;
565+
}
566+
567+
/*
568+
* We didn't detect any living workers, but not all workers are
569+
* known to have exited cleanly. Either not all workers have
570+
* launched yet, or maybe some of them failed to start or
571+
* terminated abnormally.
572+
*/
573+
for (i=0;i<pcxt->nworkers_launched;++i)
574+
{
575+
pid_tpid;
576+
shm_mq*mq;
577+
578+
/*
579+
* If the worker is BGWH_NOT_YET_STARTED or BGWH_STARTED, we
580+
* should just keep waiting. If it is BGWH_STOPPED, then
581+
* further investigation is needed.
582+
*/
583+
if (pcxt->worker[i].error_mqh==NULL||
584+
pcxt->worker[i].bgwhandle==NULL||
585+
GetBackgroundWorkerPid(pcxt->worker[i].bgwhandle,
586+
&pid)!=BGWH_STOPPED)
587+
continue;
588+
589+
/*
590+
* Check whether the worker ended up stopped without ever
591+
* attaching to the error queue. If so, the postmaster was
592+
* unable to fork the worker or it exited without initializing
593+
* properly. We must throw an error, since the caller may
594+
* have been expecting the worker to do some work before
595+
* exiting.
596+
*/
597+
mq=shm_mq_get_queue(pcxt->worker[i].error_mqh);
598+
if (shm_mq_get_sender(mq)==NULL)
599+
ereport(ERROR,
600+
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
601+
errmsg("parallel worker failed to initialize"),
602+
errhint("More details may be available in the server log.")));
603+
604+
/*
605+
* The worker is stopped, but is attached to the error queue.
606+
* Unless there's a bug somewhere, this will only happen when
607+
* the worker writes messages and terminates after the
608+
* CHECK_FOR_INTERRUPTS() near the top of this function and
609+
* before the call to GetBackgroundWorkerPid(). In that case,
610+
* or latch should have been set as well and the right things
611+
* will happen on the next pass through the loop.
612+
*/
613+
}
614+
}
534615

535616
WaitLatch(MyLatch,WL_LATCH_SET,-1,
536617
WAIT_EVENT_PARALLEL_FINISH);
@@ -787,6 +868,9 @@ HandleParallelMessage(ParallelContext *pcxt, int i, StringInfo msg)
787868
{
788869
charmsgtype;
789870

871+
if (pcxt->any_message_received!=NULL)
872+
pcxt->any_message_received[i]= true;
873+
790874
msgtype=pq_getmsgbyte(msg);
791875

792876
switch (msgtype)
@@ -981,11 +1065,16 @@ ParallelWorkerMain(Datum main_arg)
9811065
fps=shm_toc_lookup(toc,PARALLEL_KEY_FIXED, false);
9821066
MyFixedParallelState=fps;
9831067

1068+
/* Arrange to signal the leader if we exit. */
1069+
ParallelMasterPid=fps->parallel_master_pid;
1070+
ParallelMasterBackendId=fps->parallel_master_backend_id;
1071+
on_shmem_exit(ParallelWorkerShutdown, (Datum)0);
1072+
9841073
/*
985-
* Nowthatwehave a worker number, wecan find and attach to the error
986-
*queue provided for us. That'sgood, because until we do that, any
987-
*errors that happen here will not bereported back to the process that
988-
*requested that this worker belaunched.
1074+
* Now we can find and attach to the error queue provided for us. That's
1075+
* good, because until we do that, any errors that happen here will not be
1076+
* reported back to the process that requested that this worker be
1077+
* launched.
9891078
*/
9901079
error_queue_space=shm_toc_lookup(toc,PARALLEL_KEY_ERROR_QUEUE, false);
9911080
mq= (shm_mq*) (error_queue_space+
@@ -1098,9 +1187,6 @@ ParallelWorkerMain(Datum main_arg)
10981187
SetTempNamespaceState(fps->temp_namespace_id,
10991188
fps->temp_toast_namespace_id);
11001189

1101-
/* Set ParallelMasterBackendId so we know how to address temp relations. */
1102-
ParallelMasterBackendId=fps->parallel_master_backend_id;
1103-
11041190
/*
11051191
* We've initialized all of our state now; nothing should change
11061192
* hereafter.
@@ -1142,6 +1228,20 @@ ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end)
11421228
SpinLockRelease(&fps->mutex);
11431229
}
11441230

1231+
/*
1232+
* Make sure the leader tries to read from our error queue one more time.
1233+
* This guards against the case where we exit uncleanly without sending an
1234+
* ErrorResponse to the leader, for example because some code calls proc_exit
1235+
* directly.
1236+
*/
1237+
staticvoid
1238+
ParallelWorkerShutdown(intcode,Datumarg)
1239+
{
1240+
SendProcSignal(ParallelMasterPid,
1241+
PROCSIG_PARALLEL_MESSAGE,
1242+
ParallelMasterBackendId);
1243+
}
1244+
11451245
/*
11461246
* Look up (and possibly load) a parallel worker entry point function.
11471247
*

‎src/include/access/parallel.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@ typedef struct ParallelContext
4343
void*private_memory;
4444
shm_toc*toc;
4545
ParallelWorkerInfo*worker;
46+
bool*any_message_received;
4647
}ParallelContext;
4748

4849
externvolatileboolParallelMessagePending;

0 commit comments

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