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

本文主要是介绍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偏移得到自己的mqmq = (shm_mq *) (error_queue_space +ParallelWorkerNumber * PARALLEL_ERROR_QUEUE_SIZE);// 配置PGPROC到mq上shm_mq_set_sender(mq, MyProc);// mq是单纯的mq抽象，用的时候一般使用mq handle，在这里包装一层成为mqhmqh = shm_mq_attach(mq, seg, NULL);// 配置libpq的消息队列为mqhpq_redirect_to_shm_mq(seg, mqh);// 记录父进程的pid为leader pidpq_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);}elseereport(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的逻辑：