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] write) { walsndctl->lsn[SYNC_REP_WAIT_WRITE] = MyWalSnd->write; numwrite = SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE); } if (walsndctl->lsn[SYNC_REP_WAIT_FLUSH] 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] 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)