流复制心跳超时问题

基本概念

• walsender 主库数据发送进程
• walreceiver 备库数据接收进程

问题现象

在当前流复制过程中，当walreceiver进程在收到大量数据报文时会出现walsender因keepalive心跳报文超时而导致流复制断开。

原因分析

当数据报文较多时，walreceiver可能因收包或写入时间过长而导致回复keepalive报文不及时。

期望表现

当数据报文较多，walreceiver卡在处理数据报文时，仍然能回复心跳。

keepalive机制分析

流复制的大致流程如下：

1. 主库启动，主库监听相应的端口，并等待备库的连接；
2. 备库启动，并与主库监听的端口建立tcp连接，与主库同步wal日志状态，并发送流复制请求；
3. 主备库建立连接后便开始数据传输，在数据传输的同时，主库会定时发送keepalive来检查备库的状态，备库收到keepalive报文后需进行回复；主库在超时时间内收到keepalive的回复后会重新计算下一次心跳报文的发送时间，若主库未在超时时间内收到keepalive回复报文则认为备库处理超时，会进行拆链；

walsender心跳发送机制

walsender通过wal_sender_timeout确认keepalive超时时间，默认为60s。其生效机制如下：

1. 若收到walreceiver回复的报文，不论是keepalive报文还是数据文本，均会更新last_reply_time;
2. 当last_reply_time与当前时间间隔达到wal_sender_timeout的一半时（默认为30s），walsender会发送keepalive报文；
3. 当last_reply_time与当前时间间隔达到wal_sender_timeout值时，walsender认为walreceiver处理超时，将会断开流复制；

walreceiver心跳发送机制

• 收到心跳报文

  walreceiver收到心跳报文后，会立马回复心跳报文，且不存在写入操作。

• 收到数据报文

  walreceiver收到数据报文后，会在将socket的buffer缓冲区内的数据接收处理完成后回复reply报文，wlasender收到这种报文回复后同样可以重置walsender的last_reply_time;

• 主动发送reply报文

  当walreceiver在wal_receiver_timeout/2时间内没有收到walsender的心跳报文或者数据报文，walreceiver也会主动发送reply报文，并需要walsender回复。

  同样的walreceiver在wal_receiver_timeout时间内没有收到walsender的心跳报文或者数据报文，walreceiver会认为walsender超时而断开连接。

walreceiver报文接收回复及其实现

				/* See if we can read data immediately */
				len = walrcv_receive(wrconn, &buf, &wait_fd);
				if (len != 0)
				{
					/*
					 * Process the received data, and any subsequent data we
					 * can read without blocking.
					 */
					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);
					}

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

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

​