PostgreSQL数据库PMsignal——后端进程\Postmaster信号通信

PostgreSQL数据库PMsignal——后端进程\Postmaster信号通信

后端进程向Postermaster发信号

src/backend/storage/ipc/pmsignal.c

Postmaster拥有BackendList后端列表，用于存放标识postgres子进程的Backend。AssignPostmasterChildSlot函数为新的postmaster子进程选择未使用的slot，并将其状态设置为ASSIGNED，返回slot number。如果PMSignalState->next_child_flag为0，进入for循环从PMSignalState->num_child_flags数组尾部遍历，–slot为-1小于0，slot更新为PMSignalState->num_child_flags数组元素数量-1。如果该slot未使用则使用PM_CHILD_ASSIGNED，将PMSignalState->next_child_flag赋值为slot，并返回slot+1。

如果PMSignalState->next_child_flag不为0，进入for循环从PMSignalState->num_child_flags数组尾部遍历，–slot不小于0。如果该slot未使用则使用PM_CHILD_ASSIGNED，将PMSignalState->next_child_flag赋值为slot，并返回slot+1。

SendPostmasterSignal函数从子进程向postmaster发送信号，通过kill函数向postmaster发送SIGUSR1信号。支持如下PMSignalReason原由通知postmaster。

/* SendPostmasterSignal - signal the postmaster from a child process */void SendPostmasterSignal(PMSignalReason reason) { /* If called in a standalone backend, do nothing */ if (!IsUnderPostmaster) return; /* Atomically set the proper flag */ PMSignalState->PMSignalFlags[reason] = true; /* Send signal to postmaster */ kill(PostmasterPid, SIGUSR1);}typedef enum { PMSIGNAL_RECOVERY_STARTED, /* recovery has started */ PMSIGNAL_BEGIN_HOT_STANDBY, /* begin Hot Standby */ PMSIGNAL_WAKEN_ARCHIVER, /* send a NOTIFY signal to xlog archiver */ PMSIGNAL_ROTATE_LOGFILE, /* send SIGUSR1 to syslogger to rotate logfile */ PMSIGNAL_START_AUTOVAC_LAUNCHER, /* start an autovacuum launcher */ PMSIGNAL_START_AUTOVAC_WORKER, /* start an autovacuum worker */ PMSIGNAL_BACKGROUND_WORKER_CHANGE, /* background worker state change */ PMSIGNAL_START_WALRECEIVER, /* start a walreceiver */ PMSIGNAL_ADVANCE_STATE_MACHINE, /* advance postmaster's state machine */ NUM_PMSIGNALS /* Must be last value of enum! */} PMSignalReason;

在实现中，后台进程是这样通知Postmaster的：

首先在共享内存中开辟一个数组PMSignalFlags，数组中的每一位对应于一个信号。然后如果后台进程希望向Postmaster发送一个信号，那么后台首先将信号在数组PMSignalFlags中相应的元素置1（逻辑真），然后调用kill函数向postmaster发送SIGUSR1信号。当Postmaster收到SIGUSR1信号后首先检测共享存储中PMSingnalFlags，确认具体的信号是什么。同时将信号在数组PMSignalFlags中相应的元素置0（逻辑假）然后作出相应反应。

Postermaster向后端进程发信号

进程间信号通信

src/backend/storage/ipc/procsignal.c 通过传入MyBackendId获取ProcSignalSlots数组中的对应的元素槽，将信号槽中的pss_pid设置为MyProcPid，最后将该槽的指针赋值给MyProcSignalSlot。也就是一个后端一个slot，和门铃一样，其他进程可以敲击这个门铃slot来通知信号。

typedef struct { pid_t pss_pid; sig_atomic_t pss_signalFlags[NUM_PROCSIGNALS];} ProcSignalSlot;static ProcSignalSlot *ProcSignalSlots = NULL;void ProcSignalInit(int pss_idx) { volatile ProcSignalSlot *slot; Assert(pss_idx >= 1 && pss_idx <= NumProcSignalSlots); slot = &ProcSignalSlots[pss_idx - 1]; /* sanity check */ if (slot->pss_pid != 0) elog(LOG, "process %d taking over ProcSignal slot %d, but it's not empty",MyProcPid, pss_idx); /* Clear out any leftover signal reasons */ MemSet(slot->pss_signalFlags, 0, NUM_PROCSIGNALS * sizeof(sig_atomic_t)); /* Mark slot with my PID */ slot->pss_pid = MyProcPid; /* Remember slot location for CheckProcSignal */ MyProcSignalSlot = slot; /* Set up to release the slot on process exit */ on_shmem_exit(CleanupProcSignalState, Int32GetDatum(pss_idx));}

在共享内存中创建ProcSignalSlot数组

/* ProcSignalShmemSize Compute space needed for procsignal's shared memory */Size ProcSignalShmemSize(void) { return NumProcSignalSlots * sizeof(ProcSignalSlot);}/* ProcSignalShmemInit Allocate and initialize procsignal's shared memory */void ProcSignalShmemInit(void) { Size size = ProcSignalShmemSize(); bool found; ProcSignalSlots = (ProcSignalSlot *)ShmemInitStruct("ProcSignalSlots", size, &found); /* If we're first, set everything to zeroes */ if (!found) MemSet(ProcSignalSlots, 0, size);}

SendProcSignal向一个Postgres进程发送信号，如果信号发送出去了，即成功，返回0；错误返回-1并设置errno.

/* SendProcSignal Send a signal to a Postgres process * Providing backendId is optional, but it will speed up the operation. * On success (a signal was sent), zero is returned. * On error, -1 is returned, and errno is set (typically to ESRCH or EPERM). * Not to be confused with ProcSendSignal */int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId){ volatile ProcSignalSlot *slot; if (backendId != InvalidBackendId){ slot = &ProcSignalSlots[backendId - 1]; /* Note: Since there's no locking, it's possible that the target * process detaches from shared memory and exits right after this * test, before we set the flag and send signal. And the signal slot * might even be recycled by a new process, so it's remotely possible * that we set a flag for a wrong process. That's OK, all the signals * are such that no harm is done if they're mistakenly fired. */ if (slot->pss_pid == pid){ /* Atomically set the proper flag */ slot->pss_signalFlags[reason] = true; /* Send signal */ return kill(pid, SIGUSR1); } }else{ /* BackendId not provided, so search the array using pid. We search * the array back to front so as to reduce search overhead. Passing * InvalidBackendId means that the target is most likely an auxiliary * process, which will have a slot near the end of the array. */ int i; for (i = NumProcSignalSlots - 1; i >= 0; i--){ slot = &ProcSignalSlots[i]; if (slot->pss_pid == pid){ /* the above note about race conditions applies here too */ /* Atomically set the proper flag */ slot->pss_signalFlags[reason] = true; /* Send signal */ return kill(pid, SIGUSR1); } } } errno = ESRCH; return -1;}

信号处理函数

/* * procsignal_sigusr1_handler - handle SIGUSR1 signal. */voidprocsignal_sigusr1_handler(SIGNAL_ARGS){ int save_errno = errno; if (CheckProcSignal(PROCSIG_CATCHUP_INTERRUPT)) HandleCatchupInterrupt(); if (CheckProcSignal(PROCSIG_NOTIFY_INTERRUPT)) HandleNotifyInterrupt(); if (CheckProcSignal(PROCSIG_PARALLEL_MESSAGE)) HandleParallelMessageInterrupt(); if (CheckProcSignal(PROCSIG_WALSND_INIT_STOPPING)) HandleWalSndInitStopping(); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_DATABASE)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_DATABASE); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_TABLESPACE)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_TABLESPACE); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_LOCK)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_LOCK); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_SNAPSHOT); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK); if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN)) RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN); SetLatch(MyLatch); latch_sigusr1_handler(); errno = save_errno;}

业界 RPC 框架大致分为两大流派，一种侧重跨语言调用，另一种是偏重服务治理。

跨语言调用型的 RPC 框架有 Thrift、gRPC、Hessian、Hprose 等。这类 RPC 框架侧重于服务的跨语言调用，能够支持大部分的语言进行语言无关的调用，非常适合多语言调用场景。但这类框架没有服务发现相关机制，实际使用时需要代理层进行请求转发和负载均衡策略控制。

其中，gRPC 是 Google 开发的高性能、通用的开源 RPC 框架，其由 Google 主要面向移动应用开发并基于 HTTP/2 协议标准而设计，基于 ProtoBuf（Protocol Buffers）序列化协议开发，且支持众多开发语言。本身它不是分布式的，所以要实现框架的功能需要进一步的开发。

Hprose（High Performance Remote Object Service Engine）是一个 MIT 开源许可的新型轻量级跨语言跨平台的面向对象的高性能远程动态通讯中间件。

服务治理型的 RPC 框架的特点是功能丰富，提供高性能的远程调用、服务发现及服务治理能力，适用于大型服务的服务解耦及服务治理，对于特定语言(Java)的项目可以实现透明化接入。缺点是语言耦合度较高，跨语言支持难度较大。国内常见的冶理型 RPC 框架如下：

Dubbo：Dubbo 是阿里巴巴公司开源的一个 Java 高性能优秀的服务框架，使得应用可通过高性能的 RPC 实现服务的输出和输入功能，可以和 Spring 框架无缝集成。当年在淘宝内部，Dubbo 由于跟淘宝另一个类似的框架 HSF 有竞争关系，导致 Dubbo 团队解散，最近又活过来了，有专职同学投入。 DubboX：DubboX 是由当当在基于 Dubbo 框架扩展的一个 RPC 框架，支持 REST 风格的远程调用、Kryo/FST 序列化，增加了一些新的feature。Motan：Motan 是新浪微博开源的一个 Java 框架。它诞生的比较晚，起于 2013 年，2016 年 5 月开源。Motan 在微博平台中已经广泛应用，每天为数百个服务完成近千亿次的调用。 rpcx：rpcx 是一个类似阿里巴巴 Dubbo 和微博 Motan 的分布式的 RPC 服务框架，基于 Golang net/rpc 实现。但是 rpcx 基本只有一个人在维护，没有完善的社区，使用前要慎重，之前做 Golang 的 RPC 选型时也有考虑这个，最终还是放弃了，选择了 gRPC，如果想自己自研一个 RPC 框架，可以参考学习一下。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。