@ -20,8 +20,8 @@
* per - transaction state information .
*
* Replication is either synchronous or not synchronous ( async ) . If it is
* async , we just fastpath out of here . If it is sync , then in 9.1 we wait
* for the flush location on the standby before releasing the waiting backend .
* async , we just fastpath out of here . If it is sync , then we wait for
* the write or flush location on the standby before releasing the waiting backend .
* Further complexity in that interaction is expected in later releases .
*
* The best performing way to manage the waiting backends is to have a
@ -67,13 +67,15 @@ char *SyncRepStandbyNames;
static bool announce_next_takeover = true ;
static void SyncRepQueueInsert ( void ) ;
static int SyncRepWaitMode = SYNC_REP_NO_WAIT ;
static void SyncRepQueueInsert ( int mode ) ;
static void SyncRepCancelWait ( void ) ;
static int SyncRepGetStandbyPriority ( void ) ;
# ifdef USE_ASSERT_CHECKING
static bool SyncRepQueueIsOrderedByLSN ( void ) ;
static bool SyncRepQueueIsOrderedByLSN ( int mode ) ;
# endif
/*
@ -120,7 +122,7 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
* be a low cost check .
*/
if ( ! WalSndCtl - > sync_standbys_defined | |
XLByteLE ( XactCommitLSN , WalSndCtl - > lsn ) )
XLByteLE ( XactCommitLSN , WalSndCtl - > lsn [ SyncRepWaitMode ] ) )
{
LWLockRelease ( SyncRepLock ) ;
return ;
@ -132,8 +134,8 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
*/
MyProc - > waitLSN = XactCommitLSN ;
MyProc - > syncRepState = SYNC_REP_WAITING ;
SyncRepQueueInsert ( ) ;
Assert ( SyncRepQueueIsOrderedByLSN ( ) ) ;
SyncRepQueueInsert ( SyncRepWaitMode ) ;
Assert ( SyncRepQueueIsOrderedByLSN ( SyncRepWaitMode ) ) ;
LWLockRelease ( SyncRepLock ) ;
/* Alter ps display to show waiting for sync rep. */
@ -267,18 +269,19 @@ SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
}
/*
* Insert MyProc into SyncRepQueue , maintaining sorted invariant .
* Insert MyProc into the specified SyncRepQueue , maintaining sorted invariant .
*
* Usually we will go at tail of queue , though it ' s possible that we arrive
* here out of order , so start at tail and work back to insertion point .
*/
static void
SyncRepQueueInsert ( void )
SyncRepQueueInsert ( int mode )
{
PGPROC * proc ;
proc = ( PGPROC * ) SHMQueuePrev ( & ( WalSndCtl - > SyncRepQueue ) ,
& ( WalSndCtl - > SyncRepQueue ) ,
Assert ( mode > = 0 & & mode < NUM_SYNC_REP_WAIT_MODE ) ;
proc = ( PGPROC * ) SHMQueuePrev ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
while ( proc )
@ -290,7 +293,7 @@ SyncRepQueueInsert(void)
if ( XLByteLT ( proc - > waitLSN , MyProc - > waitLSN ) )
break ;
proc = ( PGPROC * ) SHMQueuePrev ( & ( WalSndCtl - > SyncRepQueue ) ,
proc = ( PGPROC * ) SHMQueuePrev ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( proc - > syncRepLinks ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
}
@ -298,7 +301,7 @@ SyncRepQueueInsert(void)
if ( proc )
SHMQueueInsertAfter ( & ( proc - > syncRepLinks ) , & ( MyProc - > syncRepLinks ) ) ;
else
SHMQueueInsertAfter ( & ( WalSndCtl - > SyncRepQueue ) , & ( MyProc - > syncRepLinks ) ) ;
SHMQueueInsertAfter ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) , & ( MyProc - > syncRepLinks ) ) ;
}
/*
@ -368,7 +371,8 @@ SyncRepReleaseWaiters(void)
{
volatile WalSndCtlData * walsndctl = WalSndCtl ;
volatile WalSnd * syncWalSnd = NULL ;
int numprocs = 0 ;
int numwrite = 0 ;
int numflush = 0 ;
int priority = 0 ;
int i ;
@ -419,20 +423,28 @@ SyncRepReleaseWaiters(void)
return ;
}
if ( XLByteLT ( walsndctl - > lsn , MyWalSnd - > flush ) )
/*
* Set the lsn first so that when we wake backends they will release
* up to this location .
*/
if ( XLByteLT ( walsndctl - > lsn [ SYNC_REP_WAIT_WRITE ] , MyWalSnd - > write ) )
{
/*
* Set the lsn first so that when we wake backends they will release
* up to this location .
*/
walsndctl - > lsn = MyWalSnd - > flush ;
numprocs = SyncRepWakeQueue ( false ) ;
walsndctl - > lsn [ SYNC_REP_WAIT_WRITE ] = MyWalSnd - > write ;
numwrite = SyncRepWakeQueue ( false , SYNC_REP_WAIT_WRITE ) ;
}
if ( XLByteLT ( walsndctl - > lsn [ SYNC_REP_WAIT_FLUSH ] , MyWalSnd - > flush ) )
{
walsndctl - > lsn [ SYNC_REP_WAIT_FLUSH ] = MyWalSnd - > flush ;
numflush = SyncRepWakeQueue ( false , SYNC_REP_WAIT_FLUSH ) ;
}
LWLockRelease ( SyncRepLock ) ;
elog ( DEBUG3 , " released %d procs up to %X/%X " ,
numprocs ,
elog ( DEBUG3 , " released %d procs up to write %X/%X, %d procs up to flush %X/%X " ,
numwrite ,
MyWalSnd - > write . xlogid ,
MyWalSnd - > write . xrecoff ,
numflush ,
MyWalSnd - > flush . xlogid ,
MyWalSnd - > flush . xrecoff ) ;
@ -507,24 +519,26 @@ SyncRepGetStandbyPriority(void)
}
/*
* Walk queue from head . Set the state of any backends that need to be woken ,
* remove them from the queue , and then wake them . Pass all = true to wake
* whole queue ; otherwise , just wake up to the walsender ' s LSN .
* Walk the specified queue from head . Set the state of any backends that
* need to be woken , remove them from the queue , and then wake them .
* Pass all = true to wake whole queue ; otherwise , just wake up to
* the walsender ' s LSN .
*
* Must hold SyncRepLock .
*/
int
SyncRepWakeQueue ( bool all )
SyncRepWakeQueue ( bool all , int mode )
{
volatile WalSndCtlData * walsndctl = WalSndCtl ;
PGPROC * proc = NULL ;
PGPROC * thisproc = NULL ;
int numprocs = 0 ;
Assert ( SyncRepQueueIsOrderedByLSN ( ) ) ;
Assert ( mode > = 0 & & mode < NUM_SYNC_REP_WAIT_MODE ) ;
Assert ( SyncRepQueueIsOrderedByLSN ( mode ) ) ;
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue ) ,
& ( WalSndCtl - > SyncRepQueue ) ,
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
while ( proc )
@ -532,7 +546,7 @@ SyncRepWakeQueue(bool all)
/*
* Assume the queue is ordered by LSN
*/
if ( ! all & & XLByteLT ( walsndctl - > lsn , proc - > waitLSN ) )
if ( ! all & & XLByteLT ( walsndctl - > lsn [ mode ] , proc - > waitLSN ) )
return numprocs ;
/*
@ -540,7 +554,7 @@ SyncRepWakeQueue(bool all)
* thisproc is valid , proc may be NULL after this .
*/
thisproc = proc ;
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue ) ,
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( proc - > syncRepLinks ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
@ -588,7 +602,12 @@ SyncRepUpdateSyncStandbysDefined(void)
* wants synchronous replication , we ' d better wake them up .
*/
if ( ! sync_standbys_defined )
SyncRepWakeQueue ( true ) ;
{
int i ;
for ( i = 0 ; i < NUM_SYNC_REP_WAIT_MODE ; i + + )
SyncRepWakeQueue ( true , i ) ;
}
/*
* Only allow people to join the queue when there are synchronous
@ -605,16 +624,18 @@ SyncRepUpdateSyncStandbysDefined(void)
# ifdef USE_ASSERT_CHECKING
static bool
SyncRepQueueIsOrderedByLSN ( void )
SyncRepQueueIsOrderedByLSN ( int mode )
{
PGPROC * proc = NULL ;
XLogRecPtr lastLSN ;
Assert ( mode > = 0 & & mode < NUM_SYNC_REP_WAIT_MODE ) ;
lastLSN . xlogid = 0 ;
lastLSN . xrecoff = 0 ;
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue ) ,
& ( WalSndCtl - > SyncRepQueue ) ,
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
while ( proc )
@ -628,7 +649,7 @@ SyncRepQueueIsOrderedByLSN(void)
lastLSN = proc - > waitLSN ;
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue ) ,
proc = ( PGPROC * ) SHMQueueNext ( & ( WalSndCtl - > SyncRepQueue [ mode ] ) ,
& ( proc - > syncRepLinks ) ,
offsetof ( PGPROC , syncRepLinks ) ) ;
}
@ -675,3 +696,20 @@ check_synchronous_standby_names(char **newval, void **extra, GucSource source)
return true ;
}
void
assign_synchronous_commit ( int newval , void * extra )
{
switch ( newval )
{
case SYNCHRONOUS_COMMIT_REMOTE_WRITE :
SyncRepWaitMode = SYNC_REP_WAIT_WRITE ;
break ;
case SYNCHRONOUS_COMMIT_REMOTE_FLUSH :
SyncRepWaitMode = SYNC_REP_WAIT_FLUSH ;
break ;
default :
SyncRepWaitMode = SYNC_REP_NO_WAIT ;
break ;
}
}