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)