database/postgres
1
0
Fork 0
mirror of https://github.com/postgres/postgres.git synced 2025-11-10 17:42:29 +03:00
Code Activity

Use latch instead of select() in walreceiver

Browse Source 
Replace use of poll()/select() by WaitLatchOrSocket(), which is more
portable and flexible.

Also change walreceiver to use its procLatch instead of a custom latch.

From: Petr Jelinek <petr@2ndquadrant.com>
This commit is contained in:
Peter Eisentraut
2016-11-30 12:00:00 -05:00
parent b999c247a5
commit 597a87ccc9
6 changed files with 43 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/backend/replication/walreceiver.c
View File

@@ -261,7 +261,7 @@ WalReceiverMain(void)
/* Arrange to clean up at walreceiver exit */
on_shmem_exit(WalRcvDie, 0);
OwnLatch(&walrcv->latch);
walrcv->latch = &MyProc->procLatch;
/* Properly accept or ignore signals the postmaster might send us */
pqsignal(SIGHUP, WalRcvSigHupHandler); /* set flag to read config
@@ -483,7 +483,7 @@ WalReceiverMain(void)
* avoiding some system calls.
*/
Assert(wait_fd != PGINVALID_SOCKET);
rc = WaitLatchOrSocket(&walrcv->latch,
rc = WaitLatchOrSocket(walrcv->latch,
WL_POSTMASTER_DEATH | WL_SOCKET_READABLE |
WL_TIMEOUT | WL_LATCH_SET,
wait_fd,
@@ -491,7 +491,7 @@ WalReceiverMain(void)
WAIT_EVENT_WAL_RECEIVER_MAIN);
if (rc & WL_LATCH_SET)
{
ResetLatch(&walrcv->latch);
ResetLatch(walrcv->latch);
if (walrcv->force_reply)
{
/*
@@ -652,7 +652,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
WakeupRecovery();
for (;;)
{
ResetLatch(&walrcv->latch);
ResetLatch(walrcv->latch);
/*
* Emergency bailout if postmaster has died. This is to avoid the
@@ -687,7 +687,7 @@ WalRcvWaitForStartPosition(XLogRecPtr *startpoint, TimeLineID *startpointTLI)
}
SpinLockRelease(&walrcv->mutex);
WaitLatch(&walrcv->latch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
WaitLatch(walrcv->latch, WL_LATCH_SET | WL_POSTMASTER_DEATH, 0,
WAIT_EVENT_WAL_RECEIVER_WAIT_START);
}
@@ -763,7 +763,7 @@ WalRcvDie(int code, Datum arg)
/* Ensure that all WAL records received are flushed to disk */
XLogWalRcvFlush(true);
DisownLatch(&walrcv->latch);
walrcv->latch = NULL;
SpinLockAcquire(&walrcv->mutex);
Assert(walrcv->walRcvState == WALRCV_STREAMING ||
@@ -812,7 +812,8 @@ WalRcvShutdownHandler(SIGNAL_ARGS)
got_SIGTERM = true;
SetLatch(&WalRcv->latch);
if (WalRcv->latch)
SetLatch(WalRcv->latch);
/* Don't joggle the elbow of proc_exit */
if (!proc_exit_inprogress && WalRcvImmediateInterruptOK)
@@ -1297,7 +1298,8 @@ void
WalRcvForceReply(void)
{
WalRcv->force_reply = true;
SetLatch(&WalRcv->latch);
if (WalRcv->latch)
SetLatch(WalRcv->latch);
}
/*