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

Commita1a789e

Browse files
committed
In walreceiver, don't try to do ereport() in a signal handler.
This is quite unsafe, even for the case of ereport(FATAL) where we won'treturn control to the interrupted code, and despite this code's use ofa flag to restrict the areas where we'd try to do it. It's possiblefor example that we interrupt malloc or free while that's holding a lockthat's meant to protect against cross-thread interference. Then, anyattempt to do malloc or free within ereport() will result in a deadlock,preventing the walreceiver process from exiting in response to SIGTERM.We hypothesize that this explains some hard-to-reproduce failures seenin the buildfarm.Hence, get rid of the immediate-exit code in WalRcvShutdownHandler,as well as the logic associated with WalRcvImmediateInterruptOK.Instead, we need to take care that potentially-blocking operationsin the walreceiver's data transmission logic (libpqwalreceiver.c)will respond reasonably promptly to the process's latch becomingset and then call ProcessWalRcvInterrupts. Much of the needed codefor that was already present in libpqwalreceiver.c. I refactoredthings a bit so that all the uses of PQgetResult use latch-awarewaiting, but didn't need to do much more.These changes should be enough to ensure that libpqwalreceiver.cwill respond promptly to SIGTERM whenever it's waiting to receivedata. In principle, it could block for a long time while waitingto send data too, and this patch does nothing to guard against that.I think that that hazard is mostly theoretical though: such blockingshould occur only if we fill the kernel's data transmission buffers,and we don't generally send enough data to make that happen withoutwaiting for input. If we find out that the hazard isn't justtheoretical, we could fix it by using PQsetnonblocking, but thatwould require more ticklish changes than I care to make now.This is a bug fix, but it seems like too big a change to push intothe back branches without much more testing than there's time forright now. Perhaps we'll back-patch once we have more confidencein the change.Patch by me; thanks to Thomas Munro for review.Discussion:https://postgr.es/m/20190416070119.GK2673@paquier.xyz
1 parent9c59289 commita1a789e

File tree

3 files changed

+85
-101
lines changed

3 files changed

+85
-101
lines changed

‎src/backend/replication/libpqwalreceiver/libpqwalreceiver.c

Lines changed: 67 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
9999

100100
/* Prototypes for private functions */
101101
staticPGresult*libpqrcv_PQexec(PGconn*streamConn,constchar*query);
102+
staticPGresult*libpqrcv_PQgetResult(PGconn*streamConn);
102103
staticchar*stringlist_to_identifierstr(PGconn*conn,List*strings);
103104

104105
/*
@@ -196,7 +197,7 @@ libpqrcv_connect(const char *conninfo, bool logical, const char *appname,
196197
if (rc&WL_LATCH_SET)
197198
{
198199
ResetLatch(MyLatch);
199-
CHECK_FOR_INTERRUPTS();
200+
ProcessWalRcvInterrupts();
200201
}
201202

202203
/* If socket is ready, advance the libpq state machine */
@@ -456,6 +457,10 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
456457
{
457458
PGresult*res;
458459

460+
/*
461+
* Send copy-end message. As in libpqrcv_PQexec, this could theoretically
462+
* block, but the risk seems small.
463+
*/
459464
if (PQputCopyEnd(conn->streamConn,NULL) <=0||
460465
PQflush(conn->streamConn))
461466
ereport(ERROR,
@@ -472,7 +477,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
472477
* If we had not yet received CopyDone from the backend, PGRES_COPY_OUT is
473478
* also possible in case we aborted the copy in mid-stream.
474479
*/
475-
res=PQgetResult(conn->streamConn);
480+
res=libpqrcv_PQgetResult(conn->streamConn);
476481
if (PQresultStatus(res)==PGRES_TUPLES_OK)
477482
{
478483
/*
@@ -486,7 +491,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
486491
PQclear(res);
487492

488493
/* the result set should be followed by CommandComplete */
489-
res=PQgetResult(conn->streamConn);
494+
res=libpqrcv_PQgetResult(conn->streamConn);
490495
}
491496
elseif (PQresultStatus(res)==PGRES_COPY_OUT)
492497
{
@@ -499,7 +504,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
499504
pchomp(PQerrorMessage(conn->streamConn)))));
500505

501506
/* CommandComplete should follow */
502-
res=PQgetResult(conn->streamConn);
507+
res=libpqrcv_PQgetResult(conn->streamConn);
503508
}
504509

505510
if (PQresultStatus(res)!=PGRES_COMMAND_OK)
@@ -509,7 +514,7 @@ libpqrcv_endstreaming(WalReceiverConn *conn, TimeLineID *next_tli)
509514
PQclear(res);
510515

511516
/* Verify that there are no more results */
512-
res=PQgetResult(conn->streamConn);
517+
res=libpqrcv_PQgetResult(conn->streamConn);
513518
if (res!=NULL)
514519
ereport(ERROR,
515520
(errmsg("unexpected result after CommandComplete: %s",
@@ -572,12 +577,11 @@ libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
572577
* The function is modeled on PQexec() in libpq, but only implements
573578
* those parts that are in use in the walreceiver api.
574579
*
575-
*Queries are always executed on the connection in streamConn.
580+
*May return NULL, rather than an error result, on failure.
576581
*/
577582
staticPGresult*
578583
libpqrcv_PQexec(PGconn*streamConn,constchar*query)
579584
{
580-
PGresult*result=NULL;
581585
PGresult*lastResult=NULL;
582586

583587
/*
@@ -588,60 +592,26 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
588592
*/
589593

590594
/*
591-
* Submit a query. Since we don't use non-blocking mode, this also can
592-
* block. But its risk is relatively small, so we ignore that for now.
595+
* Submit the query. Since we don't use non-blocking mode, this could
596+
* theoretically block. In practice, since we don't send very long query
597+
* strings, the risk seems negligible.
593598
*/
594599
if (!PQsendQuery(streamConn,query))
595600
returnNULL;
596601

597602
for (;;)
598603
{
599-
/*
600-
* Receive data until PQgetResult is ready to get the result without
601-
* blocking.
602-
*/
603-
while (PQisBusy(streamConn))
604-
{
605-
intrc;
606-
607-
/*
608-
* We don't need to break down the sleep into smaller increments,
609-
* since we'll get interrupted by signals and can either handle
610-
* interrupts here or elog(FATAL) within SIGTERM signal handler if
611-
* the signal arrives in the middle of establishment of
612-
* replication connection.
613-
*/
614-
rc=WaitLatchOrSocket(MyLatch,
615-
WL_EXIT_ON_PM_DEATH |WL_SOCKET_READABLE |
616-
WL_LATCH_SET,
617-
PQsocket(streamConn),
618-
0,
619-
WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
620-
621-
/* Interrupted? */
622-
if (rc&WL_LATCH_SET)
623-
{
624-
ResetLatch(MyLatch);
625-
CHECK_FOR_INTERRUPTS();
626-
}
604+
/* Wait for, and collect, the next PGresult. */
605+
PGresult*result;
627606

628-
/* Consume whatever data is available from the socket */
629-
if (PQconsumeInput(streamConn)==0)
630-
{
631-
/* trouble; drop whatever we had and return NULL */
632-
PQclear(lastResult);
633-
returnNULL;
634-
}
635-
}
607+
result=libpqrcv_PQgetResult(streamConn);
608+
if (result==NULL)
609+
break;/* query is complete, or failure */
636610

637611
/*
638612
* Emulate PQexec()'s behavior of returning the last result when there
639613
* are many. We are fine with returning just last error message.
640614
*/
641-
result=PQgetResult(streamConn);
642-
if (result==NULL)
643-
break;/* query is complete */
644-
645615
PQclear(lastResult);
646616
lastResult=result;
647617

@@ -655,6 +625,51 @@ libpqrcv_PQexec(PGconn *streamConn, const char *query)
655625
returnlastResult;
656626
}
657627

628+
/*
629+
* Perform the equivalent of PQgetResult(), but watch for interrupts.
630+
*/
631+
staticPGresult*
632+
libpqrcv_PQgetResult(PGconn*streamConn)
633+
{
634+
/*
635+
* Collect data until PQgetResult is ready to get the result without
636+
* blocking.
637+
*/
638+
while (PQisBusy(streamConn))
639+
{
640+
intrc;
641+
642+
/*
643+
* We don't need to break down the sleep into smaller increments,
644+
* since we'll get interrupted by signals and can handle any
645+
* interrupts here.
646+
*/
647+
rc=WaitLatchOrSocket(MyLatch,
648+
WL_EXIT_ON_PM_DEATH |WL_SOCKET_READABLE |
649+
WL_LATCH_SET,
650+
PQsocket(streamConn),
651+
0,
652+
WAIT_EVENT_LIBPQWALRECEIVER_RECEIVE);
653+
654+
/* Interrupted? */
655+
if (rc&WL_LATCH_SET)
656+
{
657+
ResetLatch(MyLatch);
658+
ProcessWalRcvInterrupts();
659+
}
660+
661+
/* Consume whatever data is available from the socket */
662+
if (PQconsumeInput(streamConn)==0)
663+
{
664+
/* trouble; return NULL */
665+
returnNULL;
666+
}
667+
}
668+
669+
/* Now we can collect and return the next PGresult */
670+
returnPQgetResult(streamConn);
671+
}
672+
658673
/*
659674
* Disconnect connection to primary, if any.
660675
*/
@@ -716,13 +731,13 @@ libpqrcv_receive(WalReceiverConn *conn, char **buffer,
716731
{
717732
PGresult*res;
718733

719-
res=PQgetResult(conn->streamConn);
734+
res=libpqrcv_PQgetResult(conn->streamConn);
720735
if (PQresultStatus(res)==PGRES_COMMAND_OK)
721736
{
722737
PQclear(res);
723738

724739
/* Verify that there are no more results. */
725-
res=PQgetResult(conn->streamConn);
740+
res=libpqrcv_PQgetResult(conn->streamConn);
726741
if (res!=NULL)
727742
{
728743
PQclear(res);
@@ -886,7 +901,7 @@ libpqrcv_processTuples(PGresult *pgres, WalRcvExecResult *walres,
886901
{
887902
char*cstrs[MaxTupleAttributeNumber];
888903

889-
CHECK_FOR_INTERRUPTS();
904+
ProcessWalRcvInterrupts();
890905

891906
/* Do the allocations in temporary context. */
892907
oldcontext=MemoryContextSwitchTo(rowcontext);

‎src/backend/replication/walreceiver.c

Lines changed: 17 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -112,28 +112,7 @@ static struct
112112
staticStringInfoDatareply_message;
113113
staticStringInfoDataincoming_message;
114114

115-
/*
116-
* About SIGTERM handling:
117-
*
118-
* We can't just exit(1) within SIGTERM signal handler, because the signal
119-
* might arrive in the middle of some critical operation, like while we're
120-
* holding a spinlock. We also can't just set a flag in signal handler and
121-
* check it in the main loop, because we perform some blocking operations
122-
* like libpqrcv_PQexec(), which can take a long time to finish.
123-
*
124-
* We use a combined approach: When WalRcvImmediateInterruptOK is true, it's
125-
* safe for the signal handler to elog(FATAL) immediately. Otherwise it just
126-
* sets got_SIGTERM flag, which is checked in the main loop when convenient.
127-
*
128-
* This is very much like what regular backends do with ImmediateInterruptOK,
129-
* ProcessInterrupts() etc.
130-
*/
131-
staticvolatileboolWalRcvImmediateInterruptOK= false;
132-
133115
/* Prototypes for private functions */
134-
staticvoidProcessWalRcvInterrupts(void);
135-
staticvoidEnableWalRcvImmediateExit(void);
136-
staticvoidDisableWalRcvImmediateExit(void);
137116
staticvoidWalRcvFetchTimeLineHistoryFiles(TimeLineIDfirst,TimeLineIDlast);
138117
staticvoidWalRcvWaitForStartPosition(XLogRecPtr*startpoint,TimeLineID*startpointTLI);
139118
staticvoidWalRcvDie(intcode,Datumarg);
@@ -151,7 +130,20 @@ static void WalRcvShutdownHandler(SIGNAL_ARGS);
151130
staticvoidWalRcvQuickDieHandler(SIGNAL_ARGS);
152131

153132

154-
staticvoid
133+
/*
134+
* Process any interrupts the walreceiver process may have received.
135+
* This should be called any time the process's latch has become set.
136+
*
137+
* Currently, only SIGTERM is of interest. We can't just exit(1) within the
138+
* SIGTERM signal handler, because the signal might arrive in the middle of
139+
* some critical operation, like while we're holding a spinlock. Instead, the
140+
* signal handler sets a flag variable as well as setting the process's latch.
141+
* We must check the flag (by calling ProcessWalRcvInterrupts) anytime the
142+
* latch has become set. Operations that could block for a long time, such as
143+
* reading from a remote server, must pay attention to the latch too; see
144+
* libpqrcv_PQgetResult for example.
145+
*/
146+
void
155147
ProcessWalRcvInterrupts(void)
156148
{
157149
/*
@@ -163,26 +155,12 @@ ProcessWalRcvInterrupts(void)
163155

164156
if (got_SIGTERM)
165157
{
166-
WalRcvImmediateInterruptOK= false;
167158
ereport(FATAL,
168159
(errcode(ERRCODE_ADMIN_SHUTDOWN),
169160
errmsg("terminating walreceiver process due to administrator command")));
170161
}
171162
}
172163

173-
staticvoid
174-
EnableWalRcvImmediateExit(void)
175-
{
176-
WalRcvImmediateInterruptOK= true;
177-
ProcessWalRcvInterrupts();
178-
}
179-
180-
staticvoid
181-
DisableWalRcvImmediateExit(void)
182-
{
183-
WalRcvImmediateInterruptOK= false;
184-
ProcessWalRcvInterrupts();
185-
}
186164

187165
/* Main entry point for walreceiver process */
188166
void
@@ -292,12 +270,10 @@ WalReceiverMain(void)
292270
PG_SETMASK(&UnBlockSig);
293271

294272
/* Establish the connection to the primary for XLOG streaming */
295-
EnableWalRcvImmediateExit();
296273
wrconn=walrcv_connect(conninfo, false,cluster_name[0] ?cluster_name :"walreceiver",&err);
297274
if (!wrconn)
298275
ereport(ERROR,
299276
(errmsg("could not connect to the primary server: %s",err)));
300-
DisableWalRcvImmediateExit();
301277

302278
/*
303279
* Save user-visible connection string. This clobbers the original
@@ -336,7 +312,6 @@ WalReceiverMain(void)
336312
* Check that we're connected to a valid server using the
337313
* IDENTIFY_SYSTEM replication command.
338314
*/
339-
EnableWalRcvImmediateExit();
340315
primary_sysid=walrcv_identify_system(wrconn,&primaryTLI);
341316

342317
snprintf(standby_sysid,sizeof(standby_sysid),UINT64_FORMAT,
@@ -348,7 +323,6 @@ WalReceiverMain(void)
348323
errdetail("The primary's identifier is %s, the standby's identifier is %s.",
349324
primary_sysid,standby_sysid)));
350325
}
351-
DisableWalRcvImmediateExit();
352326

353327
/*
354328
* Confirm that the current timeline of the primary is the same or
@@ -509,6 +483,8 @@ WalReceiverMain(void)
509483
if (rc&WL_LATCH_SET)
510484
{
511485
ResetLatch(walrcv->latch);
486+
ProcessWalRcvInterrupts();
487+
512488
if (walrcv->force_reply)
513489
{
514490
/*
@@ -577,9 +553,7 @@ WalReceiverMain(void)
577553
* The backend finished streaming. Exit streaming COPY-mode from
578554
* our side, too.
579555
*/
580-
EnableWalRcvImmediateExit();
581556
walrcv_endstreaming(wrconn,&primaryTLI);
582-
DisableWalRcvImmediateExit();
583557

584558
/*
585559
* If the server had switched to a new timeline that we didn't
@@ -726,9 +700,7 @@ WalRcvFetchTimeLineHistoryFiles(TimeLineID first, TimeLineID last)
726700
(errmsg("fetching timeline history file for timeline %u from primary server",
727701
tli)));
728702

729-
EnableWalRcvImmediateExit();
730703
walrcv_readtimelinehistoryfile(wrconn,tli,&fname,&content,&len);
731-
DisableWalRcvImmediateExit();
732704

733705
/*
734706
* Check that the filename on the master matches what we
@@ -805,7 +777,7 @@ WalRcvSigUsr1Handler(SIGNAL_ARGS)
805777
errno=save_errno;
806778
}
807779

808-
/* SIGTERM: set flag formain loop, or shutdown immediately if safe */
780+
/* SIGTERM: set flag forProcessWalRcvInterrupts */
809781
staticvoid
810782
WalRcvShutdownHandler(SIGNAL_ARGS)
811783
{
@@ -816,10 +788,6 @@ WalRcvShutdownHandler(SIGNAL_ARGS)
816788
if (WalRcv->latch)
817789
SetLatch(WalRcv->latch);
818790

819-
/* Don't joggle the elbow of proc_exit */
820-
if (!proc_exit_inprogress&&WalRcvImmediateInterruptOK)
821-
ProcessWalRcvInterrupts();
822-
823791
errno=save_errno;
824792
}
825793

‎src/include/replication/walreceiver.h

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -302,6 +302,7 @@ walrcv_clear_result(WalRcvExecResult *walres)
302302

303303
/* prototypes for functions in walreceiver.c */
304304
externvoidWalReceiverMain(void)pg_attribute_noreturn();
305+
externvoidProcessWalRcvInterrupts(void);
305306

306307
/* prototypes for functions in walreceiverfuncs.c */
307308
externSizeWalRcvShmemSize(void);

0 commit comments

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