PostgreSQL中的ProcessRepliesIfAny函数分析

2025-03-12 技术教程

本篇内容主要讲解“PostgreSQL中的ProcessRepliesIfAny函数分析”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“PostgreSQL中的ProcessRepliesIfAny函数分析”吧!

调用栈如下:

(gdb)bt#00x00007fb6e6390903in__epoll_wait_nocancel()from/lib64/libc.so.6#10x000000000088e668inWaitEventSetWaitBlock(set=0x10ac808,cur_timeout=29999,occurred_events=0x7ffd634441b0,nevents=1)atlatch.c:1048#20x000000000088e543inWaitEventSetWait(set=0x10ac808,timeout=29999,occurred_events=0x7ffd634441b0,nevents=1,wait_event_info=83886092)atlatch.c:1000#30x000000000088dcecinWaitLatchOrSocket(latch=0x7fb6dcbfc4d4,wakeEvents=27,sock=10,timeout=29999,wait_event_info=83886092)atlatch.c:385#40x000000000085405binWalSndLoop(send_data=0x8547fe<XLogSendPhysical>)atwalsender.c:2229#50x0000000000851c93inStartReplication(cmd=0x10ab750)atwalsender.c:684#60x00000000008532f0inexec_replication_command(cmd_string=0x101dd78"START_REPLICATION0/5D000000TIMELINE16")atwalsender.c:1539#70x00000000008c0170inPostgresMain(argc=1,argv=0x1049cb8,dbname=0x1049ba8"",username=0x1049b80"replicator")atpostgres.c:4178#80x000000000081e06cinBackendRun(port=0x103fb50)atpostmaster.c:4361#90x000000000081d7dfinBackendStartup(port=0x103fb50)atpostmaster.c:4033#100x0000000000819bd9inServerLoop()atpostmaster.c:1706#110x000000000081948finPostmasterMain(argc=1,argv=0x1018a50)atpostmaster.c:1379#120x0000000000742931inmain(argc=1,argv=0x1018a50)atmain.c:228一、数据结构

N/A

二、源码解读

ProcessRepliesIfAny
在streaming期间,处理接收到的消息,同时检查远程终端是否关闭了连接,执行相关处理.
代码不多也不复杂,可自行阅读.

/**Processanyincomingmessageswhilestreaming.Alsochecksiftheremote*endhasclosedtheconnection.*在streaming期间,处理接收到的消息.*同时检查远程终端是否关闭了连接,执行相关处理.*/staticvoidProcessRepliesIfAny(void){unsignedcharfirstchar;intr;boolreceived=false;//当前时间last_processing=GetCurrentTimestamp();for(;;){//----------循环接收相关消息pq_startmsgread();r=pq_getbyte_if_available(&firstchar);if(r<0){/*unexpectederrororEOF*///未知异常或者EOFereport(COMMERROR,(errcode(ERRCODE_PROTOCOL_VIOLATION),errmsg("unexpectedEOFonstandbyconnection")));//进程退出proc_exit(0);}if(r==0){/*nodataavailablewithoutblocking*///已无阻塞的消息数据,退出pq_endmsgread();break;}/*Readthemessagecontents*///读取消息内容resetStringInfo(&reply_message);if(pq_getmessage(&reply_message,0)){ereport(COMMERROR,(errcode(ERRCODE_PROTOCOL_VIOLATION),errmsg("unexpectedEOFonstandbyconnection")));proc_exit(0);}/**IfwealreadyreceivedaCopyDonefromthefrontend,thefrontend*shouldnotsendusanythinguntilwe'veclosedourendoftheCOPY.*XXX:Intheory,thefrontendcouldalreadysendthenextcommand*beforereceivingtheCopyDone,butlibpqdoesn'tcurrentlyallow*that.*如果已在前台接收到CopyDone消息,前台不应该再发送消息,直至关闭COPY.*XXX:理论上来说,在接收到CopyDone前,前台可能已经发送了下一个命令,但libpq不允许这种情况发生*/if(streamingDoneReceiving&&firstchar!='X')ereport(FATAL,(errcode(ERRCODE_PROTOCOL_VIOLATION),errmsg("unexpectedstandbymessagetype\"%c\",afterreceivingCopyDone",firstchar)));/*Handletheverylimitedsubsetofcommandsexpectedinthisphase*///处理有限几个命令switch(firstchar){/**'d'meansastandbyreplywrappedinaCopyDatapacket.*'d'意味着standby节点的应答封装了CopyData包*/case'd':ProcessStandbyMessage();received=true;break;/**CopyDonemeansthestandbyrequestedtofinishstreaming.*ReplywithCopyDone,ifwehadnotsentthatalready.*CopyDone意味着standby节点请求结束streaming.*如尚未发送,则使用CopyDone应答.*/case'c':if(!streamingDoneSending){pq_putmessage_noblock('c',NULL,0);streamingDoneSending=true;}streamingDoneReceiving=true;received=true;break;/**'X'meansthatthestandbyisclosingdownthesocket.*'X'意味着standby节点正在关闭socket*/case'X':proc_exit(0);default:ereport(FATAL,(errcode(ERRCODE_PROTOCOL_VIOLATION),errmsg("invalidstandbymessagetype\"%c\"",firstchar)));}}/**Savethelastreplytimestampifwe'vereceivedatleastonereply.*如接收到至少一条应答信息,则保存最后的应答时间戳.*/if(received){last_reply_timestamp=last_processing;waiting_for_ping_response=false;}}二、跟踪分析

在主节点上用gdb跟踪postmaster,在PostgresMain上设置断点后启动standby节点,进入断点

(gdb)setfollow-fork-modechild(gdb)bProcessRepliesIfAnyBreakpoint2at0x85343b:filewalsender.c,line1597.(gdb)cContinuing.Breakpoint2,ProcessRepliesIfAny()atwalsender.c:15971597boolreceived=false;(gdb)

查看进程信息

[xdb@localhost~]$ps-ef|greppostgresxdb13761014:16?00:00:00/appdb/xdb/pg11.2/bin/postgresxdb13771376014:16?00:00:00postgres:loggerxdb15501376016:53?00:00:00postgres:checkpointerxdb15511376016:53?00:00:00postgres:backgroundwriterxdb15521376016:53?00:00:00postgres:walwriterxdb15531376016:53?00:00:00postgres:autovacuumlauncherxdb15541376016:53?00:00:00postgres:archiverxdb15551376016:53?00:00:00postgres:statscollectorxdb15561376016:53?00:00:00postgres:logicalreplicationlauncherxdb16331376017:26?00:00:00postgres:walsenderreplicator192.168.26.26(40528)idle

循环接收相关消息

(gdb)n1599last_processing=GetCurrentTimestamp();(gdb)1603pq_startmsgread();(gdb)1604r=pq_getbyte_if_available(&firstchar);(gdb)1605if(r<0)(gdb)pr$1=1(gdb)pfirstchar$2=100'd'(gdb)

命令是’d’,执行相关处理

(gdb)n1613if(r==0)(gdb)1621resetStringInfo(&reply_message);(gdb)1622if(pq_getmessage(&reply_message,0))(gdb)1637if(streamingDoneReceiving&&firstchar!='X')(gdb)1644switch(firstchar)(gdb)1650ProcessStandbyMessage();(gdb)1651received=true;(gdb)1652break;(gdb)1681}(gdb)

设置断点

(gdb)bwalsender.c:1643Breakpoint3at0x8535b6:filewalsender.c,line1643.(gdb)bwalsender.c:1672Breakpoint4at0x85361a:filewalsender.c,line1672.(gdb)cContinuing.Breakpoint3,ProcessRepliesIfAny()atwalsender.c:16441644switch(firstchar)(gdb)Continuing....Breakpoint4,ProcessRepliesIfAny()atwalsender.c:16731673proc_exit(0);(gdb)

进程即将退出,查看进程信息

进程退出(PID=1633),启动了新的进程(PID=1637)

(gdb)n[Inferior2(process1633)exitednormally](gdb)

到此，相信大家对“PostgreSQL中的ProcessRepliesIfAny函数分析”有了更深的了解，不妨来实际操作一番吧！这里是亿速云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！