Postgresql源码（117）libpq的两套实现（socket/shm_mq）

libpq的通信方式

libpq提供了两套通信方式

• socket
• shm_mq

分别实现在下面两个文件中

• pqcomm.c
• pqmq.c

什么时候用socket通信？

除了下述并行场景，其他场景全部使用socket通信。

static const PQcommMethods PqCommSocketMethods = {
	.comm_reset = socket_comm_reset,
	.flush = socket_flush,
	.flush_if_writable = socket_flush_if_writable,
	.is_send_pending = socket_is_send_pending,
	.putmessage = socket_putmessage,
	.putmessage_noblock = socket_putmessage_noblock
};

什么时候使用mq通信？

并行框架中会将子进程的libpq的通信改成mq通信，用于子进程给父进程发送错误信息。

static const PQcommMethods PqCommMqMethods = {
	.comm_reset = mq_comm_reset,
	.flush = mq_flush,
	.flush_if_writable = mq_flush_if_writable,
	.is_send_pending = mq_is_send_pending,
	.putmessage = mq_putmessage,
	.putmessage_noblock = mq_putmessage_noblock
};

使用MQ通信需要用pq_redirect_to_shm_mq函数指定使用的dsm和mq。

注意这个pq_mq_handle是申请在dsm上的，专门用于并行框架。

void
pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
{
	PqCommMethods = &PqCommMqMethods;
	pq_mq_handle = mqh;
	whereToSendOutput = DestRemote;
	FrontendProtocol = PG_PROTOCOL_LATEST;
	on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0);
}

使用位置在并行框架子进程入口

void ParallelWorkerMain(...)
{
  ...
  ...
  // 拿到父进程在共享内存中申请mq的内存其实地址
  error_queue_space = shm_toc_lookup(toc, PARALLEL_KEY_ERROR_QUEUE, false);
  
  // 用自己的woker num偏移得到自己的mq
	mq = (shm_mq *) (error_queue_space +
					 ParallelWorkerNumber * PARALLEL_ERROR_QUEUE_SIZE);
  
  // 配置PGPROC到mq上
	shm_mq_set_sender(mq, MyProc);
  // mq是单纯的mq抽象，用的时候一般使用mq handle，在这里包装一层成为mqh
	mqh = shm_mq_attach(mq, seg, NULL);
  // 配置libpq的消息队列为mqh
	pq_redirect_to_shm_mq(seg, mqh);
  // 记录父进程的pid为leader pid
	pq_set_parallel_leader(fps->parallel_leader_pid,
						   fps->parallel_leader_backend_id);
}

配置好后，子进程已经记录了父进程的pid，在子进程中需要发送消息时：

int
mq_putmessage(...)
{
  ...
  for (;;)
	{
		// 先把书库放入mq中，flush到共享内存
		result = shm_mq_sendv(pq_mq_handle, iov, 2, true, true);

		if (pq_mq_parallel_leader_pid != 0)
		{
				... // 这里只有子进程能走进来，通知父进程读取
				SendProcSignal(pq_mq_parallel_leader_pid,
							   PROCSIG_PARALLEL_MESSAGE,
							   pq_mq_parallel_leader_backend_id);
        ...
			}
		}

		if (result != SHM_MQ_WOULD_BLOCK)
			break;

		(void) WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, 0,
						 WAIT_EVENT_MESSAGE_QUEUE_PUT_MESSAGE);
		ResetLatch(MyLatch);
		CHECK_FOR_INTERRUPTS();
	}
  ...
}

子进程发完了，信息会留存在mq中。然后给父进程发信号。

父进程收到kill过来的信号，进入信号处理函数（函数已经绑定sigusr1了），标记ParallelMessagePending

void
procsignal_sigusr1_handler(SIGNAL_ARGS)
{
	...
	if (CheckProcSignal(PROCSIG_PARALLEL_MESSAGE))
		HandleParallelMessageInterrupt();
	...
}

void
HandleParallelMessageInterrupt(void)
{
	InterruptPending = true;
	ParallelMessagePending = true;
	SetLatch(MyLatch);
}

等下次调用CHECK_FOR_INTERRUPTS宏，执行ProcessInterrupts时处理具体的消息。

在函数中会shm_mq_receive接受子进程发到mq中的消息。

void
HandleParallelMessages(void)
{
	...
	HOLD_INTERRUPTS();
	...
  ...
				res = shm_mq_receive(pcxt->worker[i].error_mqh, &nbytes,
									 &data, true);
				if (res == SHM_MQ_WOULD_BLOCK)
					break;
				else if (res == SHM_MQ_SUCCESS)
				{
					StringInfoData msg;

					initStringInfo(&msg);
					appendBinaryStringInfo(&msg, data, nbytes);
					HandleParallelMessage(pcxt, i, &msg);
					pfree(msg.data);
				}
				else
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
							 errmsg("lost connection to parallel worker")));
			}
		}
	}

	...

	RESUME_INTERRUPTS();
}

elog如何发送错误日志？

无论是并发时的子进程，还是普通进程，调用elog发送日志都会经过两步：

errstart(int elevel, const char *domain)    

做一些初始化和配置

errfinish(const char *filename, int lineno, const char *funcname)

调用EmitErrorReport发送错误

EmitErrorReport负责将日志发送到client和server log

EmitErrorReport
  /* Send to server log, if enabled */
	if (edata->output_to_server)
		send_message_to_server_log(edata);

	/* Send to client, if enabled */
	if (edata->output_to_client)
		send_message_to_frontend(edata);

这里发送到client的日志send_message_to_frontend中，会走libpq的逻辑：

1. 普通场景libpq使用PqCommSocketMethods的实现，将日志发送给客户端。
2. 并发场景子进程中libpq使用PqCommMqMethods的实现，将日志发送给父进程。