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

Commitff44fba

Browse files
committed
Replace walsender's latch with the general shared latch.
Relying on the normal shared latch simplifies interrupt/signalhandling because we can rely on all signal handlers setting the proclatch. That in turn allows us to avoid the use ofImmediateInterruptOK, which arguably isn't correct becauseWaitLatchOrSocket isn't declared to be immediately interruptible.Also change sections that wait on the walsender's latch to noticeinterrupts quicker/more reliably and make them more consistent witheach other.This is part of a larger "get rid of ImmediateInterruptOK" series.Discussion: 20150115020335.GZ5245@awork2.anarazel.de
1 parent20af53d commitff44fba

File tree

3 files changed

+55
-47
lines changed

3 files changed

+55
-47
lines changed

‎src/backend/replication/basebackup.c

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1294,15 +1294,21 @@ throttle(size_t increment)
12941294
/* Only sleep if the transfer is faster than it should be. */
12951295
if (sleep>0)
12961296
{
1297-
ResetLatch(&MyWalSnd->latch);
1297+
ResetLatch(MyLatch);
1298+
1299+
/* We're eating a potentially set latch, so check for interrupts */
1300+
CHECK_FOR_INTERRUPTS();
12981301

12991302
/*
13001303
* (TAR_SEND_SIZE / throttling_sample * elapsed_min_unit) should be
13011304
* the maximum time to sleep. Thus the cast to long is safe.
13021305
*/
1303-
wait_result=WaitLatch(&MyWalSnd->latch,
1306+
wait_result=WaitLatch(MyLatch,
13041307
WL_LATCH_SET |WL_TIMEOUT |WL_POSTMASTER_DEATH,
13051308
(long) (sleep /1000));
1309+
1310+
if (wait_result&WL_LATCH_SET)
1311+
CHECK_FOR_INTERRUPTS();
13061312
}
13071313
else
13081314
{

‎src/backend/replication/walsender.c

Lines changed: 44 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -1081,6 +1081,11 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
10811081
if (!PostmasterIsAlive())
10821082
exit(1);
10831083

1084+
/* Clear any already-pending wakeups */
1085+
ResetLatch(MyLatch);
1086+
1087+
CHECK_FOR_INTERRUPTS();
1088+
10841089
/* Process any requests or signals received recently */
10851090
if (got_SIGHUP)
10861091
{
@@ -1092,9 +1097,6 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
10921097
/* Check for input from the client */
10931098
ProcessRepliesIfAny();
10941099

1095-
/* Clear any already-pending wakeups */
1096-
ResetLatch(&MyWalSnd->latch);
1097-
10981100
/* Try to flush pending output to the client */
10991101
if (pq_flush_if_writable()!=0)
11001102
WalSndShutdown();
@@ -1117,15 +1119,12 @@ WalSndWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
11171119
WL_SOCKET_WRITEABLE |WL_SOCKET_READABLE |WL_TIMEOUT;
11181120

11191121
/* Sleep until something happens or we time out */
1120-
ImmediateInterruptOK= true;
1121-
CHECK_FOR_INTERRUPTS();
1122-
WaitLatchOrSocket(&MyWalSnd->latch,wakeEvents,
1122+
WaitLatchOrSocket(MyLatch,wakeEvents,
11231123
MyProcPort->sock,sleeptime);
1124-
ImmediateInterruptOK= false;
11251124
}
11261125

11271126
/* reactivate latch so WalSndLoop knows to continue */
1128-
SetLatch(&MyWalSnd->latch);
1127+
SetLatch(MyLatch);
11291128
}
11301129

11311130
/*
@@ -1165,6 +1164,11 @@ WalSndWaitForWal(XLogRecPtr loc)
11651164
if (!PostmasterIsAlive())
11661165
exit(1);
11671166

1167+
/* Clear any already-pending wakeups */
1168+
ResetLatch(MyLatch);
1169+
1170+
CHECK_FOR_INTERRUPTS();
1171+
11681172
/* Process any requests or signals received recently */
11691173
if (got_SIGHUP)
11701174
{
@@ -1176,9 +1180,6 @@ WalSndWaitForWal(XLogRecPtr loc)
11761180
/* Check for input from the client */
11771181
ProcessRepliesIfAny();
11781182

1179-
/* Clear any already-pending wakeups */
1180-
ResetLatch(&MyWalSnd->latch);
1181-
11821183
/* Update our idea of the currently flushed position. */
11831184
if (!RecoveryInProgress())
11841185
RecentFlushPtr=GetFlushRecPtr();
@@ -1244,15 +1245,12 @@ WalSndWaitForWal(XLogRecPtr loc)
12441245
wakeEvents |=WL_SOCKET_WRITEABLE;
12451246

12461247
/* Sleep until something happens or we time out */
1247-
ImmediateInterruptOK= true;
1248-
CHECK_FOR_INTERRUPTS();
1249-
WaitLatchOrSocket(&MyWalSnd->latch,wakeEvents,
1248+
WaitLatchOrSocket(MyLatch,wakeEvents,
12501249
MyProcPort->sock,sleeptime);
1251-
ImmediateInterruptOK= false;
12521250
}
12531251

12541252
/* reactivate latch so WalSndLoop knows to continue */
1255-
SetLatch(&MyWalSnd->latch);
1253+
SetLatch(MyLatch);
12561254
returnRecentFlushPtr;
12571255
}
12581256

@@ -1813,6 +1811,11 @@ WalSndLoop(WalSndSendDataCallback send_data)
18131811
if (!PostmasterIsAlive())
18141812
exit(1);
18151813

1814+
/* Clear any already-pending wakeups */
1815+
ResetLatch(MyLatch);
1816+
1817+
CHECK_FOR_INTERRUPTS();
1818+
18161819
/* Process any requests or signals received recently */
18171820
if (got_SIGHUP)
18181821
{
@@ -1821,14 +1824,9 @@ WalSndLoop(WalSndSendDataCallback send_data)
18211824
SyncRepInitConfig();
18221825
}
18231826

1824-
CHECK_FOR_INTERRUPTS();
1825-
18261827
/* Check for input from the client */
18271828
ProcessRepliesIfAny();
18281829

1829-
/* Clear any already-pending wakeups */
1830-
ResetLatch(&MyWalSnd->latch);
1831-
18321830
/*
18331831
* If we have received CopyDone from the client, sent CopyDone
18341832
* ourselves, and the output buffer is empty, it's time to exit
@@ -1912,11 +1910,8 @@ WalSndLoop(WalSndSendDataCallback send_data)
19121910
wakeEvents |=WL_SOCKET_WRITEABLE;
19131911

19141912
/* Sleep until something happens or we time out */
1915-
ImmediateInterruptOK= true;
1916-
CHECK_FOR_INTERRUPTS();
1917-
WaitLatchOrSocket(&MyWalSnd->latch,wakeEvents,
1913+
WaitLatchOrSocket(MyLatch,wakeEvents,
19181914
MyProcPort->sock,sleeptime);
1919-
ImmediateInterruptOK= false;
19201915
}
19211916
}
19221917
return;
@@ -1959,9 +1954,9 @@ InitWalSenderSlot(void)
19591954
walsnd->pid=MyProcPid;
19601955
walsnd->sentPtr=InvalidXLogRecPtr;
19611956
walsnd->state=WALSNDSTATE_STARTUP;
1957+
walsnd->latch=&MyProc->procLatch;
19621958
SpinLockRelease(&walsnd->mutex);
19631959
/* don't need the lock anymore */
1964-
OwnLatch((Latch*)&walsnd->latch);
19651960
MyWalSnd= (WalSnd*)walsnd;
19661961

19671962
break;
@@ -1986,19 +1981,14 @@ WalSndKill(int code, Datum arg)
19861981

19871982
Assert(walsnd!=NULL);
19881983

1989-
/*
1990-
* Clear MyWalSnd first; then disown the latch. This is so that signal
1991-
* handlers won't try to touch the latch after it's no longer ours.
1992-
*/
19931984
MyWalSnd=NULL;
19941985

1995-
DisownLatch(&walsnd->latch);
1996-
1997-
/*
1998-
* Mark WalSnd struct no longer in use. Assume that no lock is required
1999-
* for this.
2000-
*/
1986+
SpinLockAcquire(&walsnd->mutex);
1987+
/* clear latch while holding the spinlock, so it can safely be read */
1988+
walsnd->latch=NULL;
1989+
/* Mark WalSnd struct as no longer being in use. */
20011990
walsnd->pid=0;
1991+
SpinLockRelease(&walsnd->mutex);
20021992
}
20031993

20041994
/*
@@ -2570,8 +2560,8 @@ WalSndSigHupHandler(SIGNAL_ARGS)
25702560
intsave_errno=errno;
25712561

25722562
got_SIGHUP= true;
2573-
if (MyWalSnd)
2574-
SetLatch(&MyWalSnd->latch);
2563+
2564+
SetLatch(MyLatch);
25752565

25762566
errno=save_errno;
25772567
}
@@ -2603,8 +2593,7 @@ WalSndLastCycleHandler(SIGNAL_ARGS)
26032593
kill(MyProcPid,SIGTERM);
26042594

26052595
walsender_ready_to_stop= true;
2606-
if (MyWalSnd)
2607-
SetLatch(&MyWalSnd->latch);
2596+
SetLatch(MyLatch);
26082597

26092598
errno=save_errno;
26102599
}
@@ -2668,7 +2657,6 @@ WalSndShmemInit(void)
26682657
WalSnd*walsnd=&WalSndCtl->walsnds[i];
26692658

26702659
SpinLockInit(&walsnd->mutex);
2671-
InitSharedLatch(&walsnd->latch);
26722660
}
26732661
}
26742662
}
@@ -2685,7 +2673,21 @@ WalSndWakeup(void)
26852673
inti;
26862674

26872675
for (i=0;i<max_wal_senders;i++)
2688-
SetLatch(&WalSndCtl->walsnds[i].latch);
2676+
{
2677+
Latch*latch;
2678+
WalSnd*walsnd=&WalSndCtl->walsnds[i];
2679+
2680+
/*
2681+
* Get latch pointer with spinlock held, for the unlikely case that
2682+
* pointer reads aren't atomic (as they're 8 bytes).
2683+
*/
2684+
SpinLockAcquire(&walsnd->mutex);
2685+
latch=walsnd->latch;
2686+
SpinLockRelease(&walsnd->mutex);
2687+
2688+
if (latch!=NULL)
2689+
SetLatch(latch);
2690+
}
26892691
}
26902692

26912693
/* Set state for current walsender (only called in walsender) */

‎src/include/replication/walsender_private.h

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,10 +51,10 @@ typedef struct WalSnd
5151
slock_tmutex;
5252

5353
/*
54-
*Latch usedby backends to wake up this walsender when it has work to
55-
*do.
54+
*Pointer to the walsender's latch. Usedby backends to wake up this
55+
*walsender when it has work to do. NULL if the walsender isn't active.
5656
*/
57-
Latchlatch;
57+
Latch*latch;
5858

5959
/*
6060
* The priority order of the standby managed by this WALSender, as listed

0 commit comments

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