postgresql流复制原理以及流复制和逻辑复制的区别说明

CommitTransaction @ src/backend/access/transam/xact.c
RecordTransactionCommit @ src/backend/access/transam/xact.c

/*
* If we didn’t create XLOG entries, we’re done here; otherwise we
* should trigger flushing those entries the same as a commit record
* would. This will primarily happen for HOT pruning and the like; we
* want these to be flushed to disk in due time.
*/
if (!wrote_xlog) // 没有产生redo的事务，直接返回
goto cleanup;

if (wrote_xlog && markXidCommitted) // 如果产生了redo, 等待同步流复制
SyncRepWaitForLSN(XactLastRecEnd);

/*
* Wait for synchronous replication, if requested by user.
*
* Initially backends start in state SYNC_REP_NOT_WAITING and then
* change that state to SYNC_REP_WAITING before adding ourselves
* to the wait queue. During SyncRepWakeQueue() a WALSender changes
* the state to SYNC_REP_WAIT_COMPLETE once replication is confirmed.
* This backend then resets its state to SYNC_REP_NOT_WAITING.
*/
void
SyncRepWaitForLSN(XLogRecPtr XactCommitLSN)
{
…
/*
* Fast exit if user has not requested sync replication, or there are no
* sync replication standby names defined. Note that those standbys don’t
* need to be connected.
*/
if (!SyncRepRequested() || !SyncStandbysDefined()) // 如果不是同步事务或者没有定义同步流复制节点，直接返回
return;
…
/*
* We don’t wait for sync rep if WalSndCtl->sync_standbys_defined is not
* set. See SyncRepUpdateSyncStandbysDefined.
*
* Also check that the standby hasn’t already replied. Unlikely race
* condition but we’ll be fetching that cache line anyway so it’s likely
* to be a low cost check.
*/
if (!WalSndCtl->sync_standbys_defined ||
XactCommitLSN <= WalSndCtl->lsn[mode]) // 如果没有定义同步流复制节点，或者判断到commit lsn小于已同步的LSN，说明XLOG已经flush了，直接返回。
{
LWLockRelease(SyncRepLock);
return;
}
…

// 进入循环等待状态，说明本地的xlog已经flush了，只是等待同步流复制节点的REDO同步状态。
/*
* Wait for specified LSN to be confirmed.
*
* Each proc has its own wait latch, so we perform a normal latch
* check/wait loop here.
*/
for (;;) // 进入等待状态，检查latch是否满足释放等待的条件（wal sender会根据REDO的同步情况，实时更新对应的latch）
{
int syncRepState;

/* Must reset the latch before testing state. */
ResetLatch(&MyProc->procLatch);

syncRepState = MyProc->syncRepState;
if (syncRepState == SYNC_REP_WAITING)
{
LWLockAcquire(SyncRepLock, LW_SHARED);
syncRepState = MyProc->syncRepState;
LWLockRelease(SyncRepLock);
}
if (syncRepState == SYNC_REP_WAIT_COMPLETE) // 说明XLOG同步完成，退出等待
break;

// 如果本地进程挂了，输出的消息内容是，本地事务信息已持久化，但是远程也许还没有持久化
if (ProcDiePending)
{
ereport(WARNING,
(errcode(ERRCODE_ADMIN_SHUTDOWN),
errmsg(“canceling the wait for synchronous replication and terminating connection due to administrator command”),
errdetail(“The transaction has already committed locally, but might not have been replicated to the standby.”)));
whereToSendOutput = DestNone;
SyncRepCancelWait();
break;
}

// 如果用户主动cancel query，输出的消息内容是，本地事务信息已持久化，但是远程也许还没有持久化
if (QueryCancelPending)
{
QueryCancelPending = false;
ereport(WARNING,
(errmsg(“canceling wait for synchronous replication due to user request”),
errdetail(“The transaction has already committed locally, but might not have been replicated to the standby.”)));
SyncRepCancelWait();
break;
}

// 如果postgres主进程挂了，进入退出流程。
if (!PostmasterIsAlive())
{
ProcDiePending = true;
whereToSendOutput = DestNone;
SyncRepCancelWait();
break;
}

// 等待wal sender来修改对应的latch
/*
* Wait on latch. Any condition that should wake us up will set the
* latch, so no need for timeout.
*/
WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, -1);

StartReplication

WalSndLoop

ProcessRepliesIfAny

ProcessStandbyMessage

ProcessStandbyReplyMessage
if (!am_cascading_walsender) // 非级联流复制节点，那么它将调用SyncRepReleaseWaiters修改backend process等待队列中它们对应的 latch。
SyncRepReleaseWaiters();
SyncRepReleaseWaiters @ src/backend/replication/syncrep.c
/*
* Update the LSNs on each queue based upon our latest state. This
* implements a simple policy of first-valid-standby-releases-waiter.
*
* Other policies are possible, which would change what we do here and what
* perhaps also which information we store as well.
*/
void
SyncRepReleaseWaiters(void)
{
…
// 释放满足条件的等待队列
/*
* Set the lsn first so that when we wake backends they will release up to
* this location.
*/
if (walsndctl->lsn[SYNC_REP_WAIT_WRITE] < MyWalSnd->write)
{
walsndctl->lsn[SYNC_REP_WAIT_WRITE] = MyWalSnd->write;
numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
}
if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < MyWalSnd->flush)
{
walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = MyWalSnd->flush;
numflush = SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
}
…

/*
* 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.
*/
static int
SyncRepWakeQueue(bool all, int mode)
{

…
while (proc) // 修改对应的backend process 的latch
{
/*
* Assume the queue is ordered by LSN
*/
if (!all && walsndctl->lsn[mode] < proc->waitLSN)
return numprocs;

/*
* Move to next proc, so we can delete thisproc from the queue.
* thisproc is valid, proc may be NULL after this.
*/
thisproc = proc;
proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[mode]),
&(proc->syncRepLinks),
offsetof(PGPROC, syncRepLinks));

/*
* Set state to complete; see SyncRepWaitForLSN() for discussion of
* the various states.
*/
thisproc->syncRepState = SYNC_REP_WAIT_COMPLETE; // 满足条件时，改成SYNC_REP_WAIT_COMPLETE
….

synchronous_commit = off # synchronization level;
# off, local, remote_write, or on

$ vi postgresql.conf
synchronous_commit = local
$ pg_ctl reload

postgres=# select pg_cancel_backend(pid) from pg_stat_activity where pid<>pg_backend_pid();

WARNING: canceling wait for synchronous replication due to user request
DETAIL: The transaction has already committed locally, but might not have been replicated to the standby.
COMMIT
postgres=# show synchronous_commit ;
synchronous_commit
——————–
off
(1 row)