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

Commitaf6555b

Browse files
committed
Fix broken error handling in parallel pg_dump/pg_restore.
In the original design for parallel dump, worker processes reported errorsby sending them up to the master process, which would print the messages.This is unworkably fragile for a couple of reasons: it risks deadlock if aworker sends an error at an unexpected time, and if the master has alreadydied for some reason, the user will never get to see the error at all.Revert that idea and go back to just always printing messages to stderr.This approach means that if all the workers fail for similar reasons (eg,bad password or server shutdown), the user will see N copies of thatmessage, not only one as before. While that's slightly annoying, it'scertainly better than not seeing any message; not to mention that weshouldn't assume that only the first failure is interesting.An additional problem in the same area was that the master failed todisable SIGPIPE (at least until much too late), which meant that sending acommand to an already-dead worker would cause the master to crash silently.That was bad enough in itself but was made worse by the total reliance onthe master to print errors: even if the worker had reported an error, youwould probably not see it, depending on timing. Instead disable SIGPIPEright after we've forked the workers, before attempting to send themanything.Additionally, the master relies on seeing socket EOF to realize that aworker has exited prematurely --- but on Windows, there would be no EOFsince the socket is attached to the process that includes both the masterand worker threads, so it remains open. Make archive_close_connection()close the worker end of the sockets so that this acts more like the Unixcase. It's not perfect, because if a worker thread exits without goingthrough exit_nicely() the closures won't happen; but that's not reallysupposed to happen.This has been wrong all along, so back-patch to 9.3 where parallel dumpwas introduced.Report: <2458.1450894615@sss.pgh.pa.us>
1 parentbbbe2c9 commitaf6555b

File tree

3 files changed

+98
-116
lines changed

3 files changed

+98
-116
lines changed

‎src/bin/pg_dump/parallel.c

Lines changed: 65 additions & 113 deletions
Original file line numberDiff line numberDiff line change
@@ -76,8 +76,6 @@ static ShutdownInformation shutdown_info;
7676
staticconstchar*modulename=gettext_noop("parallel archiver");
7777

7878
staticParallelSlot*GetMyPSlot(ParallelState*pstate);
79-
staticvoidparallel_msg_master(ParallelSlot*slot,constchar*modulename,
80-
constchar*fmt,va_listap)pg_attribute_printf(3,0);
8179
staticvoidarchive_close_connection(intcode,void*arg);
8280
staticvoidShutdownWorkersHard(ParallelState*pstate);
8381
staticvoidWaitForTerminatingWorkers(ParallelState*pstate);
@@ -161,65 +159,6 @@ GetMyPSlot(ParallelState *pstate)
161159
returnNULL;
162160
}
163161

164-
/*
165-
* Fail and die, with a message to stderr. Parameters as for write_msg.
166-
*
167-
* This is defined in parallel.c, because in parallel mode, things are more
168-
* complicated. If the worker process does exit_horribly(), we forward its
169-
* last words to the master process. The master process then does
170-
* exit_horribly() with this error message itself and prints it normally.
171-
* After printing the message, exit_horribly() on the master will shut down
172-
* the remaining worker processes.
173-
*/
174-
void
175-
exit_horribly(constchar*modulename,constchar*fmt,...)
176-
{
177-
va_listap;
178-
ParallelState*pstate=shutdown_info.pstate;
179-
ParallelSlot*slot;
180-
181-
va_start(ap,fmt);
182-
183-
if (pstate==NULL)
184-
{
185-
/* Not in parallel mode, just write to stderr */
186-
vwrite_msg(modulename,fmt,ap);
187-
}
188-
else
189-
{
190-
slot=GetMyPSlot(pstate);
191-
192-
if (!slot)
193-
/* We're the parent, just write the message out */
194-
vwrite_msg(modulename,fmt,ap);
195-
else
196-
/* If we're a worker process, send the msg to the master process */
197-
parallel_msg_master(slot,modulename,fmt,ap);
198-
}
199-
200-
va_end(ap);
201-
202-
exit_nicely(1);
203-
}
204-
205-
/* Sends the error message from the worker to the master process */
206-
staticvoid
207-
parallel_msg_master(ParallelSlot*slot,constchar*modulename,
208-
constchar*fmt,va_listap)
209-
{
210-
charbuf[512];
211-
intpipefd[2];
212-
213-
pipefd[PIPE_READ]=slot->pipeRevRead;
214-
pipefd[PIPE_WRITE]=slot->pipeRevWrite;
215-
216-
strcpy(buf,"ERROR ");
217-
vsnprintf(buf+strlen("ERROR "),
218-
sizeof(buf)-strlen("ERROR "),fmt,ap);
219-
220-
sendMessageToMaster(pipefd,buf);
221-
}
222-
223162
/*
224163
* A thread-local version of getLocalPQExpBuffer().
225164
*
@@ -270,7 +209,7 @@ getThreadLocalPQExpBuffer(void)
270209

271210
/*
272211
* pg_dump and pg_restore register the Archive pointer for the exit handler
273-
* (called fromexit_horribly). This function mainly exists so that we can
212+
* (called fromexit_nicely). This function mainly exists so that we can
274213
* keep shutdown_info in file scope only.
275214
*/
276215
void
@@ -281,8 +220,8 @@ on_exit_close_archive(Archive *AHX)
281220
}
282221

283222
/*
284-
*This function can close archives in both the paralleland non-parallel
285-
*case.
223+
*on_exit_nicely handler for shutting down database connectionsand
224+
*worker processes cleanly.
286225
*/
287226
staticvoid
288227
archive_close_connection(intcode,void*arg)
@@ -291,42 +230,62 @@ archive_close_connection(int code, void *arg)
291230

292231
if (si->pstate)
293232
{
233+
/* In parallel mode, must figure out who we are */
294234
ParallelSlot*slot=GetMyPSlot(si->pstate);
295235

296236
if (!slot)
297237
{
298238
/*
299-
* We're the master: We have already printed out the message
300-
* passed to exit_horribly() either from the master itself or from
301-
* a worker process. Now we need to close our own database
302-
* connection (only open during parallel dump but not restore) and
303-
* shut down the remaining workers.
239+
* We're the master. Close our own database connection, if any,
240+
* and then forcibly shut down workers.
304241
*/
305-
DisconnectDatabase(si->AHX);
242+
if (si->AHX)
243+
DisconnectDatabase(si->AHX);
244+
306245
#ifndefWIN32
307246

308247
/*
309-
* Setting aborting to true switches to best-effort-mode
310-
* (send/receive but ignore errors) in communicating with our
311-
* workers.
248+
* Setting aborting to true shuts off error/warning messages that
249+
* are no longer useful once we start killing workers.
312250
*/
313251
aborting= true;
314252
#endif
315253
ShutdownWorkersHard(si->pstate);
316254
}
317-
elseif (slot->args->AH)
318-
DisconnectDatabase(&(slot->args->AH->public));
255+
else
256+
{
257+
/*
258+
* We're a worker. Shut down our own DB connection if any. On
259+
* Windows, we also have to close our communication sockets, to
260+
* emulate what will happen on Unix when the worker process exits.
261+
* (Without this, if this is a premature exit, the master would
262+
* fail to detect it because there would be no EOF condition on
263+
* the other end of the pipe.)
264+
*/
265+
if (slot->args->AH)
266+
DisconnectDatabase(&(slot->args->AH->public));
267+
268+
#ifdefWIN32
269+
closesocket(slot->pipeRevRead);
270+
closesocket(slot->pipeRevWrite);
271+
#endif
272+
}
273+
}
274+
else
275+
{
276+
/* Non-parallel operation: just kill the master DB connection */
277+
if (si->AHX)
278+
DisconnectDatabase(si->AHX);
319279
}
320-
elseif (si->AHX)
321-
DisconnectDatabase(si->AHX);
322280
}
323281

324282
/*
325283
* If we have one worker that terminates for some reason, we'd like the other
326284
* threads to terminate as well (and not finish with their 70 GB table dump
327285
* first...). Now in UNIX we can just kill these processes, and let the signal
328286
* handler set wantAbort to 1. In Windows we set a termEvent and this serves
329-
* as the signal for everyone to terminate.
287+
* as the signal for everyone to terminate. We don't print any error message,
288+
* that would just clutter the screen.
330289
*/
331290
void
332291
checkAborting(ArchiveHandle*AH)
@@ -336,7 +295,7 @@ checkAborting(ArchiveHandle *AH)
336295
#else
337296
if (wantAbort)
338297
#endif
339-
exit_horribly(modulename,"worker is terminating\n");
298+
exit_nicely(1);
340299
}
341300

342301
/*
@@ -351,8 +310,6 @@ ShutdownWorkersHard(ParallelState *pstate)
351310
#ifndefWIN32
352311
inti;
353312

354-
signal(SIGPIPE,SIG_IGN);
355-
356313
/*
357314
* Close our write end of the sockets so that the workers know they can
358315
* exit.
@@ -427,27 +384,21 @@ sigTermHandler(int signum)
427384
#endif
428385

429386
/*
430-
* This function is called by both UNIX and Windows variants to set up a
431-
* worker process.
387+
* This function is called by both UNIX and Windows variants to set up
388+
* and run a worker process. Caller should exit the process (or thread)
389+
* upon return.
432390
*/
433391
staticvoid
434392
SetupWorker(ArchiveHandle*AH,intpipefd[2],intworker)
435393
{
436394
/*
437395
* Call the setup worker function that's defined in the ArchiveHandle.
438-
*
439-
* We get the raw connection only for the reason that we can close it
440-
* properly when we shut down. This happens only that way when it is
441-
* brought down because of an error.
442396
*/
443397
(AH->SetupWorkerPtr) ((Archive*)AH);
444398

445399
Assert(AH->connection!=NULL);
446400

447401
WaitForCommands(AH,pipefd);
448-
449-
closesocket(pipefd[PIPE_READ]);
450-
closesocket(pipefd[PIPE_WRITE]);
451402
}
452403

453404
#ifdefWIN32
@@ -533,14 +484,22 @@ ParallelBackupStart(ArchiveHandle *AH)
533484
pstate->parallelSlot[i].args= (ParallelArgs*)pg_malloc(sizeof(ParallelArgs));
534485
pstate->parallelSlot[i].args->AH=NULL;
535486
pstate->parallelSlot[i].args->te=NULL;
487+
488+
/* master's ends of the pipes */
489+
pstate->parallelSlot[i].pipeRead=pipeWM[PIPE_READ];
490+
pstate->parallelSlot[i].pipeWrite=pipeMW[PIPE_WRITE];
491+
/* child's ends of the pipes */
492+
pstate->parallelSlot[i].pipeRevRead=pipeMW[PIPE_READ];
493+
pstate->parallelSlot[i].pipeRevWrite=pipeWM[PIPE_WRITE];
494+
536495
#ifdefWIN32
537496
/* Allocate a new structure for every worker */
538497
wi= (WorkerInfo*)pg_malloc(sizeof(WorkerInfo));
539498

540499
wi->worker=i;
541500
wi->AH=AH;
542-
wi->pipeRead=pstate->parallelSlot[i].pipeRevRead=pipeMW[PIPE_READ];
543-
wi->pipeWrite=pstate->parallelSlot[i].pipeRevWrite=pipeWM[PIPE_WRITE];
501+
wi->pipeRead=pipeMW[PIPE_READ];
502+
wi->pipeWrite=pipeWM[PIPE_WRITE];
544503

545504
handle=_beginthreadex(NULL,0, (void*)&init_spawned_worker_win32,
546505
wi,0,&(pstate->parallelSlot[i].threadId));
@@ -556,15 +515,6 @@ ParallelBackupStart(ArchiveHandle *AH)
556515
pipefd[0]=pipeMW[PIPE_READ];
557516
pipefd[1]=pipeWM[PIPE_WRITE];
558517

559-
/*
560-
* Store the fds for the reverse communication in pstate. Actually
561-
* we only use this in case of an error and don't use pstate
562-
* otherwise in the worker process. On Windows we write to the
563-
* global pstate, in Unix we write to our process-local copy but
564-
* that's also where we'd retrieve this information back from.
565-
*/
566-
pstate->parallelSlot[i].pipeRevRead=pipefd[PIPE_READ];
567-
pstate->parallelSlot[i].pipeRevWrite=pipefd[PIPE_WRITE];
568518
pstate->parallelSlot[i].pid=getpid();
569519

570520
/*
@@ -583,7 +533,7 @@ ParallelBackupStart(ArchiveHandle *AH)
583533

584534
/*
585535
* Close all inherited fds for communication of the master with
586-
*the other workers.
536+
*previously-forked workers.
587537
*/
588538
for (j=0;j<i;j++)
589539
{
@@ -611,11 +561,16 @@ ParallelBackupStart(ArchiveHandle *AH)
611561

612562
pstate->parallelSlot[i].pid=pid;
613563
#endif
614-
615-
pstate->parallelSlot[i].pipeRead=pipeWM[PIPE_READ];
616-
pstate->parallelSlot[i].pipeWrite=pipeMW[PIPE_WRITE];
617564
}
618565

566+
/*
567+
* Having forked off the workers, disable SIGPIPE so that master isn't
568+
* killed if it tries to send a command to a dead worker.
569+
*/
570+
#ifndefWIN32
571+
signal(SIGPIPE,SIG_IGN);
572+
#endif
573+
619574
returnpstate;
620575
}
621576

@@ -976,16 +931,13 @@ ListenToWorkers(ArchiveHandle *AH, ParallelState *pstate, bool do_wait)
976931
}
977932
else
978933
exit_horribly(modulename,
979-
"invalid message received from worker: %s\n",msg);
980-
}
981-
elseif (messageStartsWith(msg,"ERROR "))
982-
{
983-
Assert(AH->format==archDirectory||AH->format==archCustom);
984-
pstate->parallelSlot[worker].workerStatus=WRKR_TERMINATED;
985-
exit_horribly(modulename,"%s",msg+strlen("ERROR "));
934+
"invalid message received from worker: \"%s\"\n",
935+
msg);
986936
}
987937
else
988-
exit_horribly(modulename,"invalid message received from worker: %s\n",msg);
938+
exit_horribly(modulename,
939+
"invalid message received from worker: \"%s\"\n",
940+
msg);
989941

990942
/* both Unix and Win32 return pg_malloc()ed space, so we free it */
991943
free(msg);

‎src/bin/pg_dump/parallel.h

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,9 +42,9 @@ typedef struct ParallelSlot
4242
ParallelArgs*args;
4343
T_WorkerStatusworkerStatus;
4444
intstatus;
45-
intpipeRead;
45+
intpipeRead;/* master's end of the pipes */
4646
intpipeWrite;
47-
intpipeRevRead;
47+
intpipeRevRead;/* child's end of the pipes */
4848
intpipeRevWrite;
4949
#ifdefWIN32
5050
uintptr_thThread;

‎src/bin/pg_dump/pg_backup_utils.c

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,23 @@ vwrite_msg(const char *modulename, const char *fmt, va_list ap)
9393
vfprintf(stderr,_(fmt),ap);
9494
}
9595

96+
/*
97+
* Fail and die, with a message to stderr. Parameters as for write_msg.
98+
*
99+
* Note that on_exit_nicely callbacks will get run.
100+
*/
101+
void
102+
exit_horribly(constchar*modulename,constchar*fmt,...)
103+
{
104+
va_listap;
105+
106+
va_start(ap,fmt);
107+
vwrite_msg(modulename,fmt,ap);
108+
va_end(ap);
109+
110+
exit_nicely(1);
111+
}
112+
96113
/* Register a callback to be run when exit_nicely is invoked. */
97114
void
98115
on_exit_nicely(on_exit_nicely_callbackfunction,void*arg)
@@ -106,7 +123,20 @@ on_exit_nicely(on_exit_nicely_callback function, void *arg)
106123

107124
/*
108125
* Run accumulated on_exit_nicely callbacks in reverse order and then exit
109-
* quietly. This needs to be thread-safe.
126+
* without printing any message.
127+
*
128+
* If running in a parallel worker thread on Windows, we only exit the thread,
129+
* not the whole process.
130+
*
131+
* Note that in parallel operation on Windows, the callback(s) will be run
132+
* by each thread since the list state is necessarily shared by all threads;
133+
* each callback must contain logic to ensure it does only what's appropriate
134+
* for its thread. On Unix, callbacks are also run by each process, but only
135+
* for callbacks established before we fork off the child processes. (It'd
136+
* be cleaner to reset the list after fork(), and let each child establish
137+
* its own callbacks; but then the behavior would be completely inconsistent
138+
* between Windows and Unix. For now, just be sure to establish callbacks
139+
* before forking to avoid inconsistency.)
110140
*/
111141
void
112142
exit_nicely(intcode)

0 commit comments

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