pg_walreceiver

启动流程

启动receiver进程

连接主库

wal_receiver整体流程

wal_receiver整体流程由三层循环组成，第一层循环使用launch机制，第二层和第三层则使用for(;;)实现。

• 第一层循环是wal_receiver进程的控制循环，负责挂起和监听socket是否有数据到来
• 第二层循环是流复制流程，从一次流复制开始到流复制结束或流复制出现错误
• 第三层循环是wal数据和心跳数据接收，不断的从socket的buffer中收取数据，直到buffer中没有数据

wal数据接收流程

1. wal数据接收在walreceiver的第三层循环，通过循环调用walrcv_receive收取数据，当返回的数据长度小于等于0时退出；其中等于0时表示没有数据，小于0表示收包遇到错误；

   for (;;)
   {
       if (len > 0)
       {
           /*
            * Something was received from primary, so reset
            * timeout
            */
           last_recv_timestamp = GetCurrentTimestamp();
           ping_sent = false;
           XLogWalRcvProcessMsg(buf[0], &buf[1], len - 1,
                                startpointTLI);
       }
       else if (len == 0)
           break;
       else if (len < 0)
       {
           ereport(LOG,
                   (errmsg("replication terminated by primary server"),
                    errdetail("End of WAL reached on timeline %u at %X/%X.",
                              startpointTLI,
                              LSN_FORMAT_ARGS(LogstreamResult.Write))));
           endofwal = true;
           break;
       }
       len = walrcv_receive(wrconn, &buf, &wait_fd);
   }

2. 当收到数据后，会根据报文的第一个字节区分wal数据和心跳，并进行区别处理；

   switch (type)
   {
       case 'w':                /* WAL records */
           {
               /* copy message to StringInfo */
               hdrlen = sizeof(int64) + sizeof(int64) + sizeof(int64);
               if (len < hdrlen)
                   ereport(ERROR,
                           (errcode(ERRCODE_PROTOCOL_VIOLATION),
                            errmsg_internal("invalid WAL message received from primary")));
               appendBinaryStringInfo(&incoming_message, buf, hdrlen);
      
               /* read the fields */
               dataStart = pq_getmsgint64(&incoming_message);
               walEnd = pq_getmsgint64(&incoming_message);
               sendTime = pq_getmsgint64(&incoming_message);
               ProcessWalSndrMessage(walEnd, sendTime);
      
               buf += hdrlen;
               len -= hdrlen;
               XLogWalRcvWrite(buf, len, dataStart, tli);
               break;
           }
       case 'k':                /* Keepalive */
           {
               /* copy message to StringInfo */
               hdrlen = sizeof(int64) + sizeof(int64) + sizeof(char);
               if (len != hdrlen)
                   ereport(ERROR,
                           (errcode(ERRCODE_PROTOCOL_VIOLATION),
                            errmsg_internal("invalid keepalive message received from primary")));
               appendBinaryStringInfo(&incoming_message, buf, hdrlen);
      
               /* read the fields */
               walEnd = pq_getmsgint64(&incoming_message);
               sendTime = pq_getmsgint64(&incoming_message);
               replyRequested = pq_getmsgbyte(&incoming_message);
      
               ProcessWalSndrMessage(walEnd, sendTime);
      
               /* If the primary requested a reply, send one immediately */
               if (replyRequested)
                   XLogWalRcvSendReply(true, false);
               break;
           }
       default:
           ereport(ERROR,
                   (errcode(ERRCODE_PROTOCOL_VIOLATION),
                    errmsg_internal("invalid replication message type %d",
                                    type)));
   }

   如果是wal报文，则进行写入；如果是心跳报文需要进一步读取报文的最后一个字节是否为1，为1表示需要回复心跳报文。

3. 当一次数据收取结束后（socket的buffer中没有数据），此时walreceiver也需要进行回复，通知walsender收到了数据；

   /* Let the primary know that we received some data. */
   XLogWalRcvSendReply(false, false);

4. 同时如果写入了数据，需要将数据flush到磁盘

   /*
    * If we've written some records, flush them to disk and
    * let the startup process and primary server know about
    * them.
    */
   XLogWalRcvFlush(false, startpointTLI);