PostgreSQL数据库复制——后台一等公民进程WalReceiver pg_stat_wal_receiver视图

pg_stat_wal_receiver视图

pg_stat_wal_receiver视图只包含一行，显示来自该接收器连接的服务器 WAL 接收器的统计信息。（The pg_stat_wal_receiver view will contain only one row, showing statistics about the WAL receiver from that receiver’s connected server.）

pg_stat_wal_receiver视图定义如下所示。其中需要关注：

• receive_start_lsn pg_lsnWAL 接收器启动时使用的第一个预写日志位置和receive_start_tli integerWAL 接收器启动时使用的第一个时间线编号（startup进程设置receiveStart和receiveStartTLI，要求WalReceiver进行流复制）；
• written_lsn pg_lsn已收到最后一个预写日志位置并将其写入磁盘但未刷新（XLogWalRcvWrite函数调用pg_write之后设置的）；
• flushed_lsn pg_lsn已收到最后一个预写日志位置并刷新到磁盘，该字段的初始值是 WAL 接收器启动时使用的第一个日志位置（XLogWalRcvFlush函数调用fsync之后设置的）；
• received_tli integer收到并刷新到磁盘的最后一个预写日志位置的时间线编号，该字段的初始值是 WAL 接收器启动时使用的第一个日志位置的时间线编号；
• last_msg_send_time timestamp with time zoneWalReceiver进程从WAL sender收到的最后一条消息中包含的发送时间；
• last_msg_receipt_time timestamp with time zoneWalReceiver进程从 WAL sender收到的最后一条消息的WalReceiver自己的接收时间；
• latest_end_lsn pg_lsnWalReceiver进程从WAL sender收到的当前最后一个预写日志位置；
• latest_end_time timestamp with time zoneWalReceiver进程从WAL sender收到的当前最后一个预写日志位置的发送时间。

CREATE VIEW pg_stat_wal_receiver AS
    SELECT s.pid, s.status, s.receive_start_lsn, s.receive_start_tli, s.written_lsn, s.flushed_lsn,
           s.received_tli, s.last_msg_send_time, s.last_msg_receipt_time, s.latest_end_lsn, s.latest_end_time,
           s.slot_name, s.sender_host, s.sender_port, s.conninfo
    FROM pg_stat_get_wal_receiver() s
    WHERE s.pid IS NOT NULL;

pg_stat_get_wal_receiver函数

pg_stat_get_wal_receiver函数会为上面的字段获取相应的值，其中需要关注：

• written_lsn pg_lsn已收到最后一个预写日志位置并将其写入磁盘但未刷新，为XLogWalRcvWrite函数设置的WalRcv->writtenUpto（pg_atomic_uint64 CAS变量，原子读写，多个进程可以并发读写），而该值就是调用pg_pwrite之后更新的LogstreamResult.Write（Write为XLogRecPtr类型，仅由WalReceiver使用）值；
• flushed_lsn pg_lsn已收到最后一个预写日志位置并刷新到磁盘，该字段的初始值是 WAL 接收器启动时使用的第一个日志位置XLogWalRcvFlush函数调用fsync之后设置的WalRcv->flushedUpto，而该值就是调用issue_xlog_fsync之后更新的LogstreamResult.Flush（Flush和flushedUpto都为XLogRecPtr类型，更新是需要获取walrcv->mutex自旋锁的），调用pg_stat_get_wal_receiver函数获取时时没有获取walrcv->mutex自旋锁的；
• received_tli integer收到并刷新到磁盘的最后一个预写日志位置的时间线编号，该字段的初始值是 WAL 接收器启动时使用的第一个日志位置的时间线编号；
• last_msg_receipt_time timestamp with time zoneWalReceiver进程从 WAL sender收到的最后一条消息的WalReceiver自己的接收时间；last_msg_receipt_time就是获取WalRcv->lastMsgReceiptTime，该变量在ProcessWalSndrMessage函数中更新，就是WalReceiver接收到数据时自己的时间，也就是WalReceiver端的接收时间；
• latest_end_time timestamp with time zoneWalReceiver进程从WAL sender收到的最后一个预写日志位置的时间，wal sender端发送数据时设置到数据包中的发送时间。也就是在接收到新XLog数据包时更新的发送时间。
• last_msg_send_time timestamp with time zoneWalReceiver进程从WAL sender收到的当前最后一条消息中包含的发送时间；last_msg_send_time就是获取的WalRcv->lastMsgSendTime，该变量在ProcessWalSndrMessage函数中更新，就是WalReceiver接收到数据中包含的时间（即wal sender端发送数据时设置到数据包中的发送时间）；也就是在接收到心跳包和XLog数据包时都更新的发送时间。

如上，关于接收和发送时间的设置逻辑，请查看ProcessWalSndrMessage函数开头的相关代码。关于WAL日志written和flushed的位置，请查看XLogWalRcvWrite和XLogWalRcvFlush函数的逻辑。

Datum pg_stat_get_wal_receiver(PG_FUNCTION_ARGS) {
    
	TupleDesc	tupdesc;
	Datum	   *values;
	bool	   *nulls;
	int			pid;
	bool		ready_to_display;
	WalRcvState state;
	XLogRecPtr	receive_start_lsn, written_lsn, flushed_lsn, latest_end_lsn;
	TimeLineID	receive_start_tli, received_tli;
	TimestampTz last_send_time, last_receipt_time, latest_end_time;
	char		sender_host[NI_MAXHOST], slotname[NAMEDATALEN], conninfo[MAXCONNINFO];
	int			sender_port = 0;
	
	SpinLockAcquire(&WalRcv->mutex);  /* Take a lock to ensure value consistency */
	pid = (int) WalRcv->pid;
	ready_to_display = WalRcv->ready_to_display;
	state = WalRcv->walRcvState;
	receive_start_lsn = WalRcv->receiveStart;
	receive_start_tli = WalRcv->receiveStartTLI;
	flushed_lsn = WalRcv->flushedUpto;
	received_tli = WalRcv->receivedTLI;
	last_send_time = WalRcv->lastMsgSendTime;
	last_receipt_time = WalRcv->lastMsgReceiptTime;
	latest_end_lsn = WalRcv->latestWalEnd;
	latest_end_time = WalRcv->latestWalEndTime;
	strlcpy(slotname, (char *) WalRcv->slotname, sizeof(slotname));
	strlcpy(sender_host, (char *) WalRcv->sender_host, sizeof(sender_host));
	sender_port = WalRcv->sender_port;
	strlcpy(conninfo, (char *) WalRcv->conninfo, sizeof(conninfo));
	SpinLockRelease(&WalRcv->mutex);

	/* No WAL receiver (or not ready yet), just return a tuple with NULL values */
	if (pid == 0 || !ready_to_display) PG_RETURN_NULL();  // wal receiver进程pid还没设置，或者ready_to_display还没有设置

	/* Read "writtenUpto" without holding a spinlock. Note that it may not be consistent with the other shared variables of the WAL receiver protected by a spinlock, but this should not be used for data integrity checks. */
	written_lsn = pg_atomic_read_u64(&WalRcv->writtenUpto);

	/* determine result type */
	if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE) elog(ERROR, "return type must be a row type");
	values = palloc0(sizeof(Datum) * tupdesc->natts);
	nulls = palloc0(sizeof(bool) * tupdesc->natts);
	/* Fetch values */
	values[0] = Int32GetDatum(pid);

	if (!is_member_of_role(GetUserId(), ROLE_PG_READ_ALL_STATS)) {
     /* Only superusers and members of pg_read_all_stats can see details. Other users only get the pid value to know whether it is a WAL receiver, but no details. */
		MemSet(&nulls[1], true, sizeof(bool) * (tupdesc->natts - 1));
	}else{
    
		values[1] = CStringGetTextDatum(WalRcvGetStateString(state));

		if (XLogRecPtrIsInvalid(receive_start_lsn)) nulls[2] = true;  // 非法设置为null
		else values[2] = LSNGetDatum(receive_start_lsn);
		values[3] = Int32GetDatum(receive_start_tli);
		if (XLogRecPtrIsInvalid(written_lsn)) nulls[4] = true;
		else values[4] = LSNGetDatum(written_lsn);
		if (XLogRecPtrIsInvalid(flushed_lsn)) nulls[5] = true;
		else values[5] = LSNGetDatum(flushed_lsn);
		values[6] = Int32GetDatum(received_tli);
		
		if (last_send_time == 0) nulls[7] = true;
		else values[7] = TimestampTzGetDatum(last_send_time);
		if (last_receipt_time == 0) nulls[8] = true;
		else values[8] = TimestampTzGetDatum(last_receipt_time);
		if (XLogRecPtrIsInvalid(latest_end_lsn)) nulls[9] = true;
		else values[9] = LSNGetDatum(latest_end_lsn);
		if (latest_end_time == 0) nulls[10] = true;
		else values[10] = TimestampTzGetDatum(latest_end_time);
			
		if (*slotname == '\0') nulls[11] = true;
		else values[11] = CStringGetTextDatum(slotname);
		if (*sender_host == '\0') nulls[12] = true;
		else values[12] = CStringGetTextDatum(sender_host);
		if (sender_port == 0) nulls[13] = true;
		else values[13] = Int32GetDatum(sender_port);
		if (*conninfo == '\0') nulls[14] = true;
		else  values[14] = CStringGetTextDatum(conninfo);
	}	
	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls))); /* Returns the record as Datum */
}