libpq_0">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的逻辑：
