1，进程启动：

(3)Pg启动过程 - 图1

(3)Pg启动过程 - 图2(3)Pg启动过程 - 图3

1，main->PostMasterMain->ServerLoop：主进程的启动，在此从服务器连接，sender进程启动

selres = select(nSockets, &rmask, NULL, NULL, &timeout);监听文件描述符
selres = select(nSockets, &rmask, NULL, NULL, &timeout);
if (selres > 0){
for (i = 0; i < MAXLISTEN; i++){
//找一个有效的fd
if (ListenSocket[i] == PGINVALID_SOCKET)
break;
if (FD_ISSET(ListenSocket[i], &rmask)){
port = ConnCreate(ListenSocket[i]);->StreamConnection->accept
BackendStartup(port);//fork进程

2，监听到文件描述符(不同需求到达)变化后，在不同的端口创建不同的进程

pid = fork_process();

3，子进程创建后会在postgresMain启动
所有后台进程都在这里启动

PostgresMain
postgres main loop -- all backends, interactive or otherwise start here

2，sender与receiver启动

(3)Pg启动过程 - 图4

2.1，sender启动细节：

注：主进程启动sender后，由sender全权负责所有工作

#1 0x000000000071db41 in WaitEventSetWait ()
#2 0x000000000071df69 in WaitLatchOrSocket ()
#3 0x00000000006f3b64 in WalSndLoop ()
#4 0x00000000006f4320 in exec_replication_command ()
#5 0x000000000073e61d in PostgresMain ()
#6 0x00000000006cf9da in ServerLoop ()
#7 0x00000000006d0898 in PostmasterMain ()
#8 0x000000000048075e in main ()

PostgresMain ()

WalSndSignals(); 信号初始化
MarkPostmasterChildWalSender(); 告诉master进程我是sender 存活时间更长
WalSndErrorCleanup(void) 上一个walsender错误发生需要进行清理
forbidden_in_wal_sender

exec_replication_command ()

case T_StartReplicationCmd: 如果是开始复制的命令
StartReplication(cmd);执行开始复制

walsender主循环：
回调参数：函数指针
WalSndLoop(XLogSendPhysical);``WalSndLoop(WalSndSendDataCallback send_data)

为什么在walsnd断点插不进去，而在exec_replication_command可以

WalSndLoop

1ProcessRepliesIfAny(void)：处理备机的回复

  1.     for (;;)
  2.     {
  3.         pq_startmsgread();
  4.     //判断是否有字符输入，否则break
  5.         r = pq_getbyte_if_available(&firstchar);
  6.     switch (firstchar)
  7.         {
  8.                 /*
  9.                  * 'd' means a standby reply wrapped in a CopyData packet.
  10.                          d为备机的回复
  11.          */
  12.             case 'd':
  13.                 ProcessStandbyMessage();
  14.                 received = true;
  15.                 break;
  17.                 /*
  18.             c为完成流复制的请求
  19.                  * CopyDone means the standby requested to finish streaming.
  20.                  * Reply with CopyDone, if we had not sent that already.
  21.                  */
  22.             case 'c':
  23.                 if (!streamingDoneSending)
  24.                 {
  25.                     pq_putmessage_noblock('c', NULL, 0);
  26.                     streamingDoneSending = true;
  27.                 }
  29.                 streamingDoneReceiving = true;
  30.                 received = true;
  31.                 break;
  33.                 /*
  34.                 备机关闭socket
  35.                  * 'X' means that the standby is closing down the socket.
  36.                  */
  37.             case 'X':
  38.                 proc_exit(0);
  40.             default:
  41.                 ereport(FATAL,
  42.                         (errcode(ERRCODE_PROTOCOL_VIOLATION),
  43.                          errmsg("invalid standby message type \"%c\"",
  44.                                 firstchar)));
  45.         }
  46.     }

1.1ProcessStandbyReplyMessage(void)：接受备机回复wal日志的位置

  1. /*
  2.      * Update shared state for this WalSender process based on reply data from
  3.      * standby.
  5.    更新位置，根据回复的数据
  6.      */
  7.    ProcessStandbyReplyMessage(void)
  8.     {
  9.         WalSnd       *walsnd = MyWalSnd;
  11.         SpinLockAcquire(&walsnd->mutex);
  12.         walsnd->write = writePtr;
  13.         walsnd->flush = flushPtr;
  14.         walsnd->apply = applyPtr;
  15.         if (writeLag != -1 || clearLagTimes)
  16.             walsnd->writeLag = writeLag;
  17.         if (flushLag != -1 || clearLagTimes)
  18.             walsnd->flushLag = flushLag;
  19.         if (applyLag != -1 || clearLagTimes)
  20.             walsnd->applyLag = applyLag;
  21.         walsnd->replyTime = replyTime;
  22.         SpinLockRelease(&walsnd->mutex);
  23.     /*
  24.      * Advance our local xmin horizon when the client confirmed a flush.
  25.     当接受到备机的flush后，更新自己最小的XLOG  
  26.    */
  27.     if (MyReplicationSlot && flushPtr != InvalidXLogRecPtr)
  28.     {
  29.         if (SlotIsLogical(MyReplicationSlot))
  30.             LogicalConfirmReceivedLocation(flushPtr);
  31.         else
  32.             PhysicalConfirmReceivedLocation(flushPtr);
  33.     }
  34.     }

2
问题： 流复制结束，结束进程？ pending data，发送数据的条件

  1.     //流复制结束，结束循环
  2.   if (streamingDoneReceiving && streamingDoneSending &&
  3.             !pq_is_send_pending())
  4.             break;
  6.   发送数据
  7.   /*
  8.          * If we don't have any pending data in the output buffer, try to send
  9.          * some more.  If there is some, we don't bother to call send_data
  10.          * again until we've flushed it ... but we'd better assume we are not
  11.          * caught up.
  12.          */
  13.         if (!pq_is_send_pending())
  14.             send_data();    //walsndloop的参数：类型为函数指针
  15.         else
  16.             WalSndCaughtUp = false;
  18.           /* Check for replication timeout. */
  19.         WalSndCheckTimeOut();  //检测超时
  21.         /* Send keepalive if the time has come */
  22.         WalSndKeepaliveIfNecessary();
  23.     //发送keepalive
  24.         (void) WaitLatchOrSocket(MyLatch, wakeEvents,
  25.                                  MyProcPort->sock, sleeptime,
  26.                                  WAIT_EVENT_WAL_SENDER_MAIN);
  27.                  //发送完毕之后 阻塞

_ 2.1 send_data() ——> _XLogSendPhysical(void)

数据如何流动的，以及全局变量

  1. Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
  2. 将刷盘的wal开始发送
  3. XLogSendPhysical(void){
  4.     //获得刷盘地址
  5.     SendRqstPtr = GetFlushRecPtr();
  6.   //记录尾部和时间线
  7.   LagTrackerWrite()
  8.       resetStringInfo(&output_message);
  9.     pq_sendbyte(&output_message, 'w');
  11.     //发送
  12.     pq_sendint64(&output_message, startptr);    /* dataStart */
  13.     pq_sendint64(&output_message, SendRqstPtr); /* walEnd */
  14.     pq_sendint64(&output_message, 0);    /* sendtime, filled in last */
  15. }

StartReplication(cmd);StartLogicalReplication(cmd);区别

2,2，receiver启动细节

  1. #0  0x00007f06b41b30c3 in __epoll_wait_nocancel () from /lib64/libc.so.6
  2. #1  0x000000000071db41 in WaitEventSetWait ()
  3. #2  0x000000000071df69 in WaitLatchOrSocket ()
  4. #3  0x00000000006f69d8 in WalReceiverMain ()
  5. #4  0x0000000000517a04 in AuxiliaryProcessMain ()
  6. #5  0x00000000006cd15a in StartChildProcess ()
  7. #6  0x00000000006cd9d8 in MaybeStartWalReceiver ()
  8. #7  0x00000000006ce82f in sigusr1_handler ()
  9. #8  <signal handler called>
  10. #9  0x00007f06b41a9b23 in __select_nocancel () from /lib64/libc.so.6
  11. #10 0x00000000006ced35 in ServerLoop ()
  12. #11 0x00000000006d0898 in PostmasterMain ()
  13. #12 0x000000000048075e in main ()
  1. standby：ServerLoop()
  3. //信号 允许receiver 发现了自己为从服务器
  4. if (WalReceiverRequested)
  5. //启动receiver进程
  6.             MaybeStartWalReceiver();
  1. AuxiliaryProcessMain(int argc, char *argv[])
  3. WalReceiverMain(void)
  1. WalReceiverMain.XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1);根据类型处理消息
  2.   switch (type)
  3.     {
  4.   //wal消息
  5.         case 'w':                /* WAL records */
  6.             {
  7.                 /* copy message to StringInfo */
  8.                 hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
  9.                 if (len < hdrlen)
  10.                     ereport(ERROR,
  11.                             (errcode(ERRCODE_PROTOCOL_VIOLATION),
  12.                              errmsg_internal("invalid WAL message received from primary")));
  13.                 appendBinaryStringInfo(&incoming_message, buf, hdrlen);
  15.                 /* read the fields */
  16.                 dataStart = pq_getmsgint64(&incoming_message);
  17.                 walEnd = pq_getmsgint64(&incoming_message);
  18.                 sendTime = pq_getmsgint64(&incoming_message);
  19.                 ProcessWalSndrMessage(walEnd, sendTime);
  21.                 buf += hdrlen;
  22.                 len -= hdrlen;
  23.                 XLogWalRcvWrite(buf, len, dataStart);
  24.                 break;
  25.             }
  26.      //保活消息
  27.         case 'k':                /* Keepalive */
  28.             {
  29.                 /* copy message to StringInfo */
  30.                 hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
  31.                 if (len != hdrlen)
  32.                     ereport(ERROR,
  33.                             (errcode(ERRCODE_PROTOCOL_VIOLATION),
  34.                              errmsg_internal("invalid keepalive message received from primary")));
  35.                 appendBinaryStringInfo(&incoming_message, buf, hdrlen);
  37.                 /* read the fields */
  38.                 walEnd = pq_getmsgint64(&incoming_message);
  39.                 sendTime = pq_getmsgint64(&incoming_message);
  40.                 replyRequested = pq_getmsgbyte(&incoming_message);
  42.                 ProcessWalSndrMessage(walEnd, sendTime);
  44.                 /* If the primary requested a reply, send one immediately */
  45.                 if (replyRequested)
  46.                     XLogWalRcvSendReply(true, false);
  47.                 break;
  48.             }
  51. }
  53. WalReceiverMain.XLogWalRcvSendReply:不停发送自己的log位置时间等
  54. XLogWalRcvFlush(false);//刷盘，并且让startup进程了解

2.3 startup启动细节

  1. #0  0x00007f06b41b30c3 in __epoll_wait_nocancel () from /lib64/libc.so.6
  2. #1  0x000000000071db41 in WaitEventSetWait ()
  3. #2  0x000000000071df69 in WaitLatchOrSocket ()
  4. #3  0x0000000000504eb4 in XLogPageRead ()
  5. #4  0x0000000000512aa2 in ReadPageInternal ()
  6. #5  0x0000000000513370 in XLogReadRecord ()
  7. #6  0x0000000000502c6b in ReadRecord ()
  8. #7  0x00000000005092ba in StartupXLOG ()
  9. #8  0x00000000006d0ff1 in StartupProcessMain ()
  10. #9  0x00000000005179fa in AuxiliaryProcessMain ()
  11. #10 0x00000000006cd15a in StartChildProcess ()
  12. #11 0x00000000006cd27a in PostmasterStateMachine ()
  13. #12 0x00000000006ce215 in reaper ()
  14. #13 <signal handler called>
  15. #14 0x00007f06b41a9b23 in __select_nocancel () from /lib64/libc.so.6
  16. #15 0x00000000006ced35 in ServerLoop ()
  17. #16 0x00000000006d0898 in PostmasterMain ()
  18. #17 0x000000000048075e in main ()
  21. StartupXLOG(void)...