這篇文章主要介紹“PostgreSQL的后臺進程walsender分析”，在日常操作中，相信很多人在PostgreSQL的后臺進程walsender分析問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”PostgreSQL的后臺進程walsender分析”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！
該進程實質上是streaming replication環境中master節點上普通的backend進程,在standby節點啟動時,standby節點向master發送連接請求,master節點的postmaster進程接收到請求后,啟動該進程與standby節點的walreceiver進程建立通訊連接,用于傳輸WAL Record.
walsender啟動后,使用gdb跟蹤此進程,其調用棧如下:
(gdb) bt #0 0x00007fb6e6390903 in __epoll_wait_nocancel () from /lib64/libc.so.6 #1 0x000000000088e668 in WaitEventSetWaitBlock (set=0x10ac808, cur_timeout=29999, occurred_events=0x7ffd634441b0, nevents=1) at latch.c:1048 #2 0x000000000088e543 in WaitEventSetWait (set=0x10ac808, timeout=29999, occurred_events=0x7ffd634441b0, nevents=1, wait_event_info=83886092) at latch.c:1000 #3 0x000000000088dcec in WaitLatchOrSocket (latch=0x7fb6dcbfc4d4, wakeEvents=27, sock=10, timeout=29999, wait_event_info=83886092) at latch.c:385 #4 0x000000000085405b in WalSndLoop (send_data=0x8547fe <XLogSendPhysical>) at walsender.c:2229 #5 0x0000000000851c93 in StartReplication (cmd=0x10ab750) at walsender.c:684 #6 0x00000000008532f0 in exec_replication_command (cmd_string=0x101dd78 "START_REPLICATION 0/5D000000 TIMELINE 16") at walsender.c:1539 #7 0x00000000008c0170 in PostgresMain (argc=1, argv=0x1049cb8, dbname=0x1049ba8 "", username=0x1049b80 "replicator") at postgres.c:4178 #8 0x000000000081e06c in BackendRun (port=0x103fb50) at postmaster.c:4361 #9 0x000000000081d7df in BackendStartup (port=0x103fb50) at postmaster.c:4033 #10 0x0000000000819bd9 in ServerLoop () at postmaster.c:1706 #11 0x000000000081948f in PostmasterMain (argc=1, argv=0x1018a50) at postmaster.c:1379 #12 0x0000000000742931 in main (argc=1, argv=0x1018a50) at main.c:228
本節首先介紹調用棧中PostgresMain函數.
StringInfo
StringInfoData結構體保存關于擴展字符串的相關信息.
/*-------------------------
* StringInfoData holds information about an extensible string.
* StringInfoData結構體保存關于擴展字符串的相關信息.
* data is the current buffer for the string (allocated with palloc).
* data 通過palloc分配的字符串緩存
* len is the current string length. There is guaranteed to be
* a terminating '\0' at data[len], although this is not very
* useful when the string holds binary data rather than text.
* len 是當前字符串的長度.保證以ASCII 0(\0)結束(data[len] = '\0').
* 雖然如果存儲的是二進制數據而不是文本時不太好使.
* maxlen is the allocated size in bytes of 'data', i.e. the maximum
* string size (including the terminating '\0' char) that we can
* currently store in 'data' without having to reallocate
* more space. We must always have maxlen > len.
* maxlen 以字節為單位已分配的'data'的大小,限定了最大的字符串大小(包括結尾的ASCII 0)
* 小于此尺寸的數據可以直接存儲而無需重新分配.
* cursor is initialized to zero by makeStringInfo or initStringInfo,
* but is not otherwise touched by the stringinfo.c routines.
* Some routines use it to scan through a StringInfo.
* cursor 通過makeStringInfo或initStringInfo初始化為0,但不受stringinfo.c例程的影響.
* 某些例程使用該字段掃描StringInfo
*-------------------------
*/
typedef struct StringInfoData
{
char *data;
int len;
int maxlen;
int cursor;
} StringInfoData;
typedef StringInfoData *StringInfo;PostgresMain
后臺進程postgres的主循環入口 — 所有的交互式或其他形式的后臺進程在這里啟動.
其主要邏輯如下:
1.初始化相關變量
2.初始化進程信息,設置進程狀態,初始化GUC參數
3.解析命令行參數并作相關校驗
4.如為walsender進程,則調用WalSndSignals初始化,否則執行其他信號初始化
5.初始化BlockSig/UnBlockSig/StartupBlockSig
6.非Postmaster,則檢查數據庫路徑/切換路徑/創建鎖定文件等操作
7.調用BaseInit執行基本的初始化
8.調用InitProcess/InitPostgres初始化進程
9.重置內存上下文,處理加載庫和前后臺消息交互等
10.初始化內存上下文
11.進入主循環
11.1切換至MessageContext上下文
11.2初始化輸入的消息
11.3給客戶端發送可以執行查詢等消息
11.4讀取命令
11.5根據命令類型執行相關操作
/* ----------------------------------------------------------------
* PostgresMain
* postgres main loop -- all backends, interactive or otherwise start here
* postgres主循環 -- 所有的交互式或其他形式的后臺進程在這里啟動
*
* argc/argv are the command line arguments to be used. (When being forked
* by the postmaster, these are not the original argv array of the process.)
* dbname is the name of the database to connect to, or NULL if the database
* name should be extracted from the command line arguments or defaulted.
* username is the PostgreSQL user name to be used for the session.
* argc/argv是命令行參數(postmaster fork進程時,不存在原有的進程argv數組).
* dbname是連接的數據庫名稱,如需要從命令行參數中解析或者為默認的數據庫名稱,則為NULL.
* username是PostgreSQL會話的用戶名.
* ----------------------------------------------------------------
*/
/*
輸入：
argc/argv-Main函數的輸入參數
dbname-數據庫名稱
username-用戶名
輸出：
無
*/
void
PostgresMain(int argc, char *argv[],
const char *dbname,
const char *username)
{
int firstchar;//臨時變量，讀取輸入的Command
StringInfoData input_message;//字符串增強結構體
sigjmp_buf local_sigjmp_buf;//系統變量
volatile bool send_ready_for_query = true;//
bool disable_idle_in_transaction_timeout = false;
/* Initialize startup process environment if necessary. */
//如需要,初始化啟動進程環境
if (!IsUnderPostmaster//未初始化？initialized for the bootstrap/standalone case
InitStandaloneProcess(argv[0]);//初始化進程
SetProcessingMode(InitProcessing);//設置進程狀態為InitProcessing
/*
* Set default values for command-line options.
* 設置命令行選項默認值
*/
if (!IsUnderPostmaster)
InitializeGUCOptions();//初始化GUC參數，GUC=Grand Unified Configuration
/*
* Parse command-line options.
* 解析命令行選項
*/
process_postgres_switches(argc, argv, PGC_POSTMASTER, &dbname);//解析輸入參數
/* Must have gotten a database name, or have a default (the username) */
//必須包含數據庫名稱或者存在默認值
if (dbname == NULL)//輸入的dbname為空
{
dbname = username;//設置為用戶名
if (dbname == NULL)//如仍為空，報錯
ereport(FATAL,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("%s: no database nor user name specified",
progname)));
}
/* Acquire configuration parameters, unless inherited from postmaster */
//請求配置參數,除非從postmaster中繼承
if (!IsUnderPostmaster)
{
if (!SelectConfigFiles(userDoption, progname))//讀取配置文件conf/hba文件&定位數據目錄
proc_exit(1);
}
/*
* Set up signal handlers and masks.
* 配置信號handlers和masks.
*
* Note that postmaster blocked all signals before forking child process,
* so there is no race condition whereby we might receive a signal before
* we have set up the handler.
* 注意在fork子進程前postmaster已阻塞了所有信號,
* 因此就算接收到信號,但在完成配置handler前不會存在條件爭用.
*
* Also note: it's best not to use any signals that are SIG_IGNored in the
* postmaster. If such a signal arrives before we are able to change the
* handler to non-SIG_IGN, it'll get dropped. Instead, make a dummy
* handler in the postmaster to reserve the signal. (Of course, this isn't
* an issue for signals that are locally generated, such as SIGALRM and
* SIGPIPE.)
* 同時注意:最好不要使用在postmaster中標記為SIG_IGNored的信號.
* 如果在改變處理器為non-SIG_IGN前,接收到這樣的信號,會被清除.
* 相反,可以在postmaster中創建dummy handler來保留這樣的信號.
* (當然,對于本地產生的信號,比如SIGALRM和SIGPIPE,這不會是問題)
*/
if (am_walsender)//wal sender進程？
WalSndSignals();//如果是，則調用WalSndSignals
else//不是wal sender進程
{
//設置標記,讀取配置文件
pqsignal(SIGHUP, PostgresSigHupHandler); /* set flag to read config
* file */
//中斷信號處理器(中斷當前查詢)
pqsignal(SIGINT, StatementCancelHandler); /* cancel current query */
//終止當前查詢并退出
pqsignal(SIGTERM, die); /* cancel current query and exit */
/*
* In a standalone backend, SIGQUIT can be generated from the keyboard
* easily, while SIGTERM cannot, so we make both signals do die()
* rather than quickdie().
* 在standalone進程,SIGQUIT可很容易的通過鍵盤生成,而SIGTERM則不好生成,
* 因此讓這兩個信號執行die()而不是quickdie().
*/
//bool IsUnderPostmaster = false
if (IsUnderPostmaster)
//悲催時刻,執行quickdie()
pqsignal(SIGQUIT, quickdie); /* hard crash time */
else
//執行die()
pqsignal(SIGQUIT, die); /* cancel current query and exit */
//建立SIGALRM處理器
InitializeTimeouts(); /* establishes SIGALRM handler */
/*
* Ignore failure to write to frontend. Note: if frontend closes
* connection, we will notice it and exit cleanly when control next
* returns to outer loop. This seems safer than forcing exit in the
* midst of output during who-knows-what operation...
* 忽略寫入前端的錯誤.
* 注意:如果前端關閉了連接,會通知并在空中下一次返回給外層循環時退出.
* 這看起來會比在who-knows-what操作期間強制退出安全一些.
*/
pqsignal(SIGPIPE, SIG_IGN);
pqsignal(SIGUSR1, procsignal_sigusr1_handler);
pqsignal(SIGUSR2, SIG_IGN);
pqsignal(SIGFPE, FloatExceptionHandler);
/*
* Reset some signals that are accepted by postmaster but not by
* backend
* 重置一些postmaster接收而后臺進程不會接收的信號
*/
//在某些平臺上,system()需要這個信號
pqsignal(SIGCHLD, SIG_DFL); /* system() requires this on some
* platforms */
}
//初始化BlockSig/UnBlockSig/StartupBlockSig
pqinitmask();//Initialize BlockSig, UnBlockSig, and StartupBlockSig.
if (IsUnderPostmaster)
{
/* We allow SIGQUIT (quickdie) at all times */
//放開SIGQUIT(quickdie)
sigdelset(&BlockSig, SIGQUIT);
}
//除了SIGQUIT,阻塞其他
PG_SETMASK(&BlockSig); /* block everything except SIGQUIT */
if (!IsUnderPostmaster)
{
/*
* Validate we have been given a reasonable-looking DataDir (if under
* postmaster, assume postmaster did this already).
* 驗證已給出了reasonable-looking DataDir
* (如在postmaster下,假定postmaster已完成了這個事情)
*/
checkDataDir();//確認數據庫路徑OK，使用stat命令
/* Change into DataDir (if under postmaster, was done already) */
//切換至數據庫路徑，使用chdir命令
ChangeToDataDir();
/*
* Create lockfile for data directory.
*/
//創建鎖定文件，CreateLockFile(DIRECTORY_LOCK_FILE, amPostmaster, "", true, DataDir);
CreateDataDirLockFile(false);
/* read control file (error checking and contains config ) */
//讀取控制文件(錯誤檢查和包含配置)
LocalProcessControlFile(false);//Read the control file, set respective GUCs.
/* Initialize MaxBackends (if under postmaster, was done already) */
//從配置選項中初始化MaxBackends
InitializeMaxBackends();//Initialize MaxBackends value from config options.
}
/* Early initialization */
BaseInit();//執行基本的初始化
/*
* Create a per-backend PGPROC struct in shared memory, except in the
* EXEC_BACKEND case where this was done in SubPostmasterMain. We must do
* this before we can use LWLocks (and in the EXEC_BACKEND case we already
* had to do some stuff with LWLocks).
* 在共享內存中創建每個backend都有的PGPROC結構體,除了在SubPostmasterMain的EXEC_BACKEND情況.
* 在可以使用LWLocks前必須執行該操作.
* (在EXEC_BACKEND中,已使用了LWLocks執行這個場景)
*/
// initialize a per-process data structure for this backend
#ifdef EXEC_BACKEND
if (!IsUnderPostmaster)
InitProcess();
#else
InitProcess();
#endif
/* We need to allow SIGINT, etc during the initial transaction */
//在初始化事務期間需要允許SIGINT等等
PG_SETMASK(&UnBlockSig);
/*
* General initialization.
* 常規的初始化.
*
* NOTE: if you are tempted to add code in this vicinity, consider putting
* it inside InitPostgres() instead. In particular, anything that
* involves database access should be there, not here.
* 注意:如果希望在此處添加代碼,請考慮將其放入InitPostgres()中.
* 特別的,任何涉及到數據庫訪問的內容都應該在InitPostgres中,而不是在這里.
*/
InitPostgres(dbname, InvalidOid, username, InvalidOid, NULL, false);//Initialize POSTGRES
/*
* If the PostmasterContext is still around, recycle the space; we don't
* need it anymore after InitPostgres completes. Note this does not trash
* *MyProcPort, because ConnCreate() allocated that space with malloc()
* ... else we'd need to copy the Port data first. Also, subsidiary data
* such as the username isn't lost either; see ProcessStartupPacket().
* 如果PostmasterContext仍然存在,回收空間;
* 在InitPostgres完成后,我們不再需要這些空間.
* 注意:這個操作不會回收*MyProcPort,因為ConnCreate()分配
*/
if (PostmasterContext)
{
MemoryContextDelete(PostmasterContext);
PostmasterContext = NULL;
}
//完成初始化后，設置進程模式為NormalProcessing
SetProcessingMode(NormalProcessing);
/*
* Now all GUC states are fully set up. Report them to client if
* appropriate.
*/
//Report GUC
BeginReportingGUCOptions();
/*
* Also set up handler to log session end; we have to wait till now to be
* sure Log_disconnections has its final value.
*/
//設置處理器,用于記錄會話結束;
//等待直至確保Log_disconnections最終有值存在
if (IsUnderPostmaster && Log_disconnections)
on_proc_exit(log_disconnections, 0);//this function adds a callback function to the list of functions invoked by proc_exit()
/* Perform initialization specific to a WAL sender process. */
//為WAL sender進程執行特別的初始化
if (am_walsender)
InitWalSender();//初始化 WAL sender process
/*
* process any libraries that should be preloaded at backend start (this
* likewise can't be done until GUC settings are complete)
* 處理在后臺進程啟動時需要提前預裝載的庫(這個步驟在GUC配置完成后才能夠執行)
*/
process_session_preload_libraries();//加載LIB
/*
* Send this backend's cancellation info to the frontend.
* 發送后端的取消信息到前臺
*/
if (whereToSendOutput == DestRemote)
{
StringInfoData buf;
pq_beginmessage(&buf, 'K');
pq_sendint32(&buf, (int32) MyProcPid);
pq_sendint32(&buf, (int32) MyCancelKey);
pq_endmessage(&buf);
/* Need not flush since ReadyForQuery will do it. */
//不需要flush,因為ReadyForQuery會執行該操作
}
/* Welcome banner for case */
//standalone的歡迎信息
if (whereToSendOutput == DestDebug)
printf("\nPostgreSQL stand-alone backend %s\n", PG_VERSION);
/*
* Create the memory context we will use in the main loop.
* 創建主循環中使用的內存上下文
*
* MessageContext is reset once per iteration of the main loop, ie, upon
* completion of processing of each command message from the client.
* 主循環中,每一次迭代都會重置MessageContext,比如完成了每個命令的處理,已從客戶端返回了信息
*/
//初始化內存上下文：MessageContext
MessageContext = AllocSetContextCreate(TopMemoryContext,
"MessageContext",
ALLOCSET_DEFAULT_SIZES);
/*
* Create memory context and buffer used for RowDescription messages. As
* SendRowDescriptionMessage(), via exec_describe_statement_message(), is
* frequently executed for ever single statement, we don't want to
* allocate a separate buffer every time.
* 創建RowDescription消息的內存上下文和緩存.
* 每一條單獨的語句執行時都會頻繁的通過exec_describe_statement_message()函數
* 調用SendRowDescriptionMessage(),不希望在每次都分配單獨的緩存.
*/
//TODO 傳輸RowDescription messages？
row_description_context = AllocSetContextCreate(TopMemoryContext,
"RowDescriptionContext",
ALLOCSET_DEFAULT_SIZES);
MemoryContextSwitchTo(row_description_context);
initStringInfo(&row_description_buf);
MemoryContextSwitchTo(TopMemoryContext);
/*
* Remember stand-alone backend startup time
* 記錄stand-alone后臺進程的啟動時間
*/
if (!IsUnderPostmaster)
PgStartTime = GetCurrentTimestamp();//記錄啟動時間
/*
* POSTGRES main processing loop begins here
* POSTGRES的主處理循環在這里開始
*
* If an exception is encountered, processing resumes here so we abort the
* current transaction and start a new one.
* 如果出現了異常,處理過程會在這里恢復因此PG可以回滾當前事務并啟動新事務
*
* You might wonder why this isn't coded as an infinite loop around a
* PG_TRY construct. The reason is that this is the bottom of the
* exception stack, and so with PG_TRY there would be no exception handler
* in force at all during the CATCH part. By leaving the outermost setjmp
* always active, we have at least some chance of recovering from an error
* during error recovery. (If we get into an infinite loop thereby, it
* will soon be stopped by overflow of elog.c's internal state stack.)
* 你可能會對這里不用PG_TRY構造一個無限循環感到困惑.
* 理由是這是異常棧的底,因此在這里使用PG_TRY會導致在CATCH部分沒有任何的異常處理器.
* 通過讓最外層的setjmp始終處于活動狀態,我們起碼有機會在錯誤恢復的錯誤中進行恢復.
* (如果進入了無線循環,會很快因為elog's內部狀態棧的溢出而停止)
*
* Note that we use sigsetjmp(..., 1), so that this function's signal mask
* (to wit, UnBlockSig) will be restored when longjmp'ing to here. This
* is essential in case we longjmp'd out of a signal handler on a platform
* where that leaves the signal blocked. It's not redundant with the
* unblock in AbortTransaction() because the latter is only called if we
* were inside a transaction.
* 注意我們使用sigsetjmp(...,1),
* 以便該函數的信號mask(也就是說,UnBlockSig)在longjmp到這里的時候可以被還原.
* 在某個讓信號繼續阻塞的平臺上通過longjmp跳出信號處理器時這樣的處理是需要的.
* 這與AbortTransaction()設置unblock并不多余因為如果我們在事務中保證只執行了一次.
*/
if (sigsetjmp(local_sigjmp_buf, 1) != 0)//
{
/*
* NOTE: if you are tempted to add more code in this if-block,
* consider the high probability that it should be in
* AbortTransaction() instead. The only stuff done directly here
* should be stuff that is guaranteed to apply *only* for outer-level
* error recovery, such as adjusting the FE/BE protocol status.
* 注意:如果你希望在if-block中添加代碼,建議在AbortTransaction()中添加.
* 直接添加在這里的唯一理由是可以應用對高層的錯誤恢復,比如調整FE/BE協議狀態.
*/
/* Since not using PG_TRY, must reset error stack by hand */
//不使用PG_TRY,必須重置錯誤棧
error_context_stack = NULL;
/* Prevent interrupts while cleaning up */
//清理時禁止中斷
HOLD_INTERRUPTS();
/*
* Forget any pending QueryCancel request, since we're returning to
* the idle loop anyway, and cancel any active timeout requests. (In
* future we might want to allow some timeout requests to survive, but
* at minimum it'd be necessary to do reschedule_timeouts(), in case
* we got here because of a query cancel interrupting the SIGALRM
* interrupt handler.) Note in particular that we must clear the
* statement and lock timeout indicators, to prevent any future plain
* query cancels from being misreported as timeouts in case we're
* forgetting a timeout cancel.
* 廢棄正在處理中QueryCancel請求,因為進程會返回到空閑循環中,同時取消所有活動的超時請求.
* (在未來,我們可能希望運行某些超時請求仍然存活,但最起碼有需要執行reschedule_timeouts(),
* 在這種情況下到達這里的原因是查詢取消是通過SIGALRM終端處理器中斷的).
* 注意特別的,必須清理語句和鎖超時提示器,已避免在取消超時后后續的普通查詢出現超時時沒有報告.
*/
disable_all_timeouts(false);
QueryCancelPending = false; /* second to avoid race condition */
stmt_timeout_active = false;
/* Not reading from the client anymore. */
//不再從客戶端讀取信息
DoingCommandRead = false;
/* Make sure libpq is in a good state */
//確保libq狀態OK
pq_comm_reset();
/* Report the error to the client and/or server log */
//向客戶端和/或服務器日志報告錯誤
EmitErrorReport();
/*
* Make sure debug_query_string gets reset before we possibly clobber
* the storage it points at.
* 確保debug_query_string在可能破壞它所指向的存儲之前重置
*/
debug_query_string = NULL;
/*
* Abort the current transaction in order to recover.
* 取消當前事務
*/
AbortCurrentTransaction();
if (am_walsender)
//如為walsender,則執行清理工作
WalSndErrorCleanup();
//錯誤清理
PortalErrorCleanup();
SPICleanup();
/*
* We can't release replication slots inside AbortTransaction() as we
* need to be able to start and abort transactions while having a slot
* acquired. But we never need to hold them across top level errors,
* so releasing here is fine. There's another cleanup in ProcKill()
* ensuring we'll correctly cleanup on FATAL errors as well.
* 在AbortTransaction()中不能釋放replication slots因為需要在持有slot時啟動和回滾事務.
* 但不限于在頂層出現錯誤時持有這些slots,因此在這里釋放這些slots是OK的.
* 這里在ProcKill()中存在另外一個清理確保我們可以在FATAL錯誤中正確的恢復.
*/
if (MyReplicationSlot != NULL)
ReplicationSlotRelease();
/* We also want to cleanup temporary slots on error. */
//出現錯誤時,清理臨時slots
ReplicationSlotCleanup();
//重置JIT
jit_reset_after_error();
/*
* Now return to normal top-level context and clear ErrorContext for
* next time.
* 現在可以回到正常的頂層上下文中并為下次循環清理ErrorContext
*/
MemoryContextSwitchTo(TopMemoryContext);
FlushErrorState();
/*
* If we were handling an extended-query-protocol message, initiate
* skip till next Sync. This also causes us not to issue
* ReadyForQuery (until we get Sync).
* 如果我們正在處理extended-query-protocol消息,啟動跳過直至下次Sync.
* 這同時會導致我們不會觸發ReadyForQuery(直至接收到Sync)
*/
if (doing_extended_query_message)
ignore_till_sync = true;
/* We don't have a transaction command open anymore */
//不再有打開的事務命令
xact_started = false;
/*
* If an error occurred while we were reading a message from the
* client, we have potentially lost track of where the previous
* message ends and the next one begins. Even though we have
* otherwise recovered from the error, we cannot safely read any more
* messages from the client, so there isn't much we can do with the
* connection anymore.
* 如果在讀取客戶端消息時出現錯誤,進程可能已經丟失了上一條消息結束和下一條消息開始的位置.
* 雖然從錯誤中恢復了,但我們仍然不能安全的從客戶端讀取消息,因此我們對該連接已無能無力.
*/
if (pq_is_reading_msg())
ereport(FATAL,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("terminating connection because protocol synchronization was lost")));
/* Now we can allow interrupts again */
//允許中斷
RESUME_INTERRUPTS();
}
/* We can now handle ereport(ERROR) */
//現在可以處理ereport(ERROR)了
PG_exception_stack = &local_sigjmp_buf;
if (!ignore_till_sync)
//錯誤恢復后重新初始化
send_ready_for_query = true; /* initially, or after error */
/*
* Non-error queries loop here.
* 世界清凈了...
*/
for (;;)//主循環
{
/*
* At top of loop, reset extended-query-message flag, so that any
* errors encountered in "idle" state don't provoke skip.
* 在循環的最開始處,重置extended-query-message標記,
* 以便在"idle"狀態遇到錯誤時不會跳過
*/
doing_extended_query_message = false;
/*
* Release storage left over from prior query cycle, and create a new
* query input buffer in the cleared MessageContext.
* 釋放上一個查詢周期中殘余的存儲空間,并在干凈的MessageContext中創建新的查詢輸入緩存
*/
MemoryContextSwitchTo(MessageContext);//切換至MessageContext
MemoryContextResetAndDeleteChildren(MessageContext);
initStringInfo(&input_message);//初始化輸入的信息
/*
* Also consider releasing our catalog snapshot if any, so that it's
* not preventing advance of global xmin while we wait for the client.
* 嘗試釋放catalog snapshot,以便在等待客戶端返回時不會阻礙全局xmin的增加.
*/
InvalidateCatalogSnapshotConditionally();
/*
* (1) If we've reached idle state, tell the frontend we're ready for
* a new query.
* (1) 如果是idle狀態,告訴前臺已準備接受新查詢請求了.
*
* Note: this includes fflush()'ing the last of the prior output.
* 注意:這包含了fflush()'ing前一個輸出的最后一個.
*
* This is also a good time to send collected statistics to the
* collector, and to update the PS stats display. We avoid doing
* those every time through the message loop because it'd slow down
* processing of batched messages, and because we don't want to report
* uncommitted updates (that confuses autovacuum). The notification
* processor wants a call too, if we are not in a transaction block.
* 發送收集的統計信息到collector,正當其時!
* 在每次消息循環時都發送統計信息是需要避免的,
* 因為我不希望報告未提交的更新(這會讓autoacuum出現混亂).
* 如果我們不在事務塊中,那么通知處理器希望調用一次.
*/
if (send_ready_for_query)//I am ready!
{
if (IsAbortedTransactionBlockState())
{
set_ps_display("idle in transaction (aborted)", false);
pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);
/* Start the idle-in-transaction timer */
if (IdleInTransactionSessionTimeout > 0)
{
disable_idle_in_transaction_timeout = true;
enable_timeout_after(IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
IdleInTransactionSessionTimeout);
}
}
else if (IsTransactionOrTransactionBlock())
{
set_ps_display("idle in transaction", false);
pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);
/* Start the idle-in-transaction timer */
if (IdleInTransactionSessionTimeout > 0)
{
disable_idle_in_transaction_timeout = true;
enable_timeout_after(IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
IdleInTransactionSessionTimeout);
}
}
else
{
ProcessCompletedNotifies();
pgstat_report_stat(false);
set_ps_display("idle", false);
pgstat_report_activity(STATE_IDLE, NULL);
}
ReadyForQuery(whereToSendOutput);
send_ready_for_query = false;
}
/*
* (2) Allow asynchronous signals to be executed immediately if they
* come in while we are waiting for client input. (This must be
* conditional since we don't want, say, reads on behalf of COPY FROM
* STDIN doing the same thing.)
* (2) 如果異步信號在等待客戶端輸入時接收到,那么允許馬上執行.
* (這是有條件的,因為我們不希望或者說執行COPY FORM STDIN同樣的動作)
*/
DoingCommandRead = true;
/*
* (3) read a command (loop blocks here)
* (3) 讀取命令(循環塊)
*/
firstchar = ReadCommand(&input_message);//讀取命令
/*
* (4) disable async signal conditions again.
* (4) 再次禁用異步信號條件
*
* Query cancel is supposed to be a no-op when there is no query in
* progress, so if a query cancel arrived while we were idle, just
* reset QueryCancelPending. ProcessInterrupts() has that effect when
* it's called when DoingCommandRead is set, so check for interrupts
* before resetting DoingCommandRead.
* 在處理過程中如無查詢,那么取消查詢被認為是no-op的,
* 因此如果在空閑狀態下接收到查詢取消信號,那么重置QueryCancelPending.
* ProcessInterrupts()函數在DoingCommandRead設置的時候調用會有類似的影響,
* 因此在重置DoingCommandRead前重新檢查中斷.
*/
CHECK_FOR_INTERRUPTS();
DoingCommandRead = false;
/*
* (5) turn off the idle-in-transaction timeout
* (5) 關閉idle-in-transaction超時控制
*/
if (disable_idle_in_transaction_timeout)
{
disable_timeout(IDLE_IN_TRANSACTION_SESSION_TIMEOUT, false);
disable_idle_in_transaction_timeout = false;
}
/*
* (6) check for any other interesting events that happened while we
* slept.
* (6) 在休眠時檢查感興趣的事件
*/
if (ConfigReloadPending)
{
ConfigReloadPending = false;
ProcessConfigFile(PGC_SIGHUP);
}
/*
* (7) process the command. But ignore it if we're skipping till
* Sync.
* (7) 處理命令.但如果我們設置了ignore_till_sync則忽略之.
*/
if (ignore_till_sync && firstchar != EOF)
continue;
switch (firstchar)
{
case 'Q': /* simple query */
{
//--------- 簡單查詢
const char *query_string;
/* Set statement_timestamp() */
//設置時間戳
SetCurrentStatementStartTimestamp();
//SQL語句
query_string = pq_getmsgstring(&input_message);
pq_getmsgend(&input_message);
if (am_walsender)
{
//如為WAL sender,執行exec_replication_command
if (!exec_replication_command(query_string))
exec_simple_query(query_string);
}
else
//普通的后臺進程
exec_simple_query(query_string);//執行SQL語句
send_ready_for_query = true;
}
break;
case 'P': /* parse */
{
//---------- 解析
const char *stmt_name;
const char *query_string;
int numParams;
Oid *paramTypes = NULL;
forbidden_in_wal_sender(firstchar);
/* Set statement_timestamp() */
SetCurrentStatementStartTimestamp();
stmt_name = pq_getmsgstring(&input_message);
query_string = pq_getmsgstring(&input_message);
numParams = pq_getmsgint(&input_message, 2);
if (numParams > 0)
{
int i;
paramTypes = (Oid *) palloc(numParams * sizeof(Oid));
for (i = 0; i < numParams; i++)
paramTypes[i] = pq_getmsgint(&input_message, 4);
}
pq_getmsgend(&input_message);
//執行解析
exec_parse_message(query_string, stmt_name,
paramTypes, numParams);
}
break;
case 'B': /* bind */
//------------- 綁定
forbidden_in_wal_sender(firstchar);
/* Set statement_timestamp() */
SetCurrentStatementStartTimestamp();
/*
* this message is complex enough that it seems best to put
* the field extraction out-of-line
* 該消息看起來比較復雜,看起來最好的做法是提取字段
*/
exec_bind_message(&input_message);
break;
case 'E': /* execute */
{
//------------ 執行
const char *portal_name;
int max_rows;
forbidden_in_wal_sender(firstchar);
/* Set statement_timestamp() */
SetCurrentStatementStartTimestamp();
portal_name = pq_getmsgstring(&input_message);
max_rows = pq_getmsgint(&input_message, 4);
pq_getmsgend(&input_message);
exec_execute_message(portal_name, max_rows);
}
break;
case 'F': /* fastpath function call */
//----------- 函數調用
forbidden_in_wal_sender(firstchar);
/* Set statement_timestamp() */
SetCurrentStatementStartTimestamp();
/* Report query to various monitoring facilities. */
pgstat_report_activity(STATE_FASTPATH, NULL);
set_ps_display("<FASTPATH>", false);
/* start an xact for this function invocation */
start_xact_command();
/*
* Note: we may at this point be inside an aborted
* transaction. We can't throw error for that until we've
* finished reading the function-call message, so
* HandleFunctionRequest() must check for it after doing so.
* Be careful not to do anything that assumes we're inside a
* valid transaction here.
*/
/* switch back to message context */
MemoryContextSwitchTo(MessageContext);
HandleFunctionRequest(&input_message);
/* commit the function-invocation transaction */
finish_xact_command();
send_ready_for_query = true;
break;
case 'C': /* close */
{
//---------- 關閉
int close_type;
const char *close_target;
forbidden_in_wal_sender(firstchar);
close_type = pq_getmsgbyte(&input_message);
close_target = pq_getmsgstring(&input_message);
pq_getmsgend(&input_message);
switch (close_type)
{
case 'S':
if (close_target[0] != '\0')
DropPreparedStatement(close_target, false);
else
{
/* special-case the unnamed statement */
drop_unnamed_stmt();
}
break;
case 'P':
{
Portal portal;
portal = GetPortalByName(close_target);
if (PortalIsValid(portal))
PortalDrop(portal, false);
}
break;
default:
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("invalid CLOSE message subtype %d",
close_type)));
break;
}
if (whereToSendOutput == DestRemote)
pq_putemptymessage('3'); /* CloseComplete */
}
break;
case 'D': /* describe */
{
//------------- 描述比如\d等命令
int describe_type;
const char *describe_target;
forbidden_in_wal_sender(firstchar);
/* Set statement_timestamp() (needed for xact) */
SetCurrentStatementStartTimestamp();
describe_type = pq_getmsgbyte(&input_message);
describe_target = pq_getmsgstring(&input_message);
pq_getmsgend(&input_message);
switch (describe_type)
{
case 'S':
exec_describe_statement_message(describe_target);
break;
case 'P':
exec_describe_portal_message(describe_target);
break;
default:
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("invalid DESCRIBE message subtype %d",
describe_type)));
break;
}
}
break;
case 'H': /* flush */
//--------- flush 刷新
pq_getmsgend(&input_message);
if (whereToSendOutput == DestRemote)
pq_flush();
break;
case 'S': /* sync */
//---------- Sync 同步
pq_getmsgend(&input_message);
finish_xact_command();
send_ready_for_query = true;
break;
/*
* 'X' means that the frontend is closing down the socket. EOF
* means unexpected loss of frontend connection. Either way,
* perform normal shutdown.
*/
case 'X':
case EOF:
/*
* Reset whereToSendOutput to prevent ereport from attempting
* to send any more messages to client.
*/
if (whereToSendOutput == DestRemote)
whereToSendOutput = DestNone;
/*
* NOTE: if you are tempted to add more code here, DON'T!
* Whatever you had in mind to do should be set up as an
* on_proc_exit or on_shmem_exit callback, instead. Otherwise
* it will fail to be called during other backend-shutdown
* scenarios.
*/
proc_exit(0);
case 'd': /* copy data */
case 'c': /* copy done */
case 'f': /* copy fail */
/*
* Accept but ignore these messages, per protocol spec; we
* probably got here because a COPY failed, and the frontend
* is still sending data.
*/
break;
default:
ereport(FATAL,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg("invalid frontend message type %d",
firstchar)));
}
} /* end of input-reading loop */
}在主節點上用gdb跟蹤postmaster,在PostgresMain上設置斷點后啟動standby節點,進入斷點
[xdb@localhost ~]$ ps -ef|grep postgre xdb 1263 1 0 14:20 pts/0 00:00:00 /appdb/xdb/pg11.2/bin/postgres (gdb) b PostgresMain Breakpoint 1 at 0x8bf9df: file postgres.c, line 3660. (gdb) set follow-fork-mode child (gdb) c Continuing. [New process 1332] [Thread debugging using libthread_db enabled] Using host libthread_db library "/lib64/libthread_db.so.1". [Switching to Thread 0x7fb3885d98c0 (LWP 1332)] Breakpoint 1, PostgresMain (argc=1, argv=0x1aa4c78, dbname=0x1aa4b68 "", username=0x1aa4b40 "replicator") at postgres.c:3660 3660 volatile bool send_ready_for_query = true; (gdb)
1.初始化相關變量
注意變量IsUnderPostmaster,如為T則表示該進程為postmaster的子進程
(gdb) p *argv $1 = 0xc27715 "postgres" (gdb) n 3661 bool disable_idle_in_transaction_timeout = false; (gdb) 3664 if (!IsUnderPostmaster) (gdb) p IsUnderPostmaster $2 = true
2.初始化進程信息,設置進程狀態,初始化GUC參數
(gdb) n 3667 SetProcessingMode(InitProcessing); (gdb) 3672 if (!IsUnderPostmaster) (gdb) p InitProcessing $3 = InitProcessing
3.解析命令行參數并作相關校驗
(gdb) n 3678 process_postgres_switches(argc, argv, PGC_POSTMASTER, &dbname); (gdb) 3681 if (dbname == NULL) (gdb) p dbname $4 = 0x1aa4b68 "" (gdb) p username $5 = 0x1aa4b40 "replicator" (gdb) n 3692 if (!IsUnderPostmaster) (gdb)
4.如為walsender進程,則調用WalSndSignals初始化,否則執行其他信號初始化
3712 if (am_walsender) (gdb) 3713 WalSndSignals(); (gdb)
5.初始化BlockSig/UnBlockSig/StartupBlockSig
(gdb) 3751 pqinitmask(); (gdb) 3753 if (IsUnderPostmaster) (gdb) 3756 sigdelset(&BlockSig, SIGQUIT); (gdb) (gdb) 3759 PG_SETMASK(&BlockSig); /* block everything except SIGQUIT */ (gdb)
6.非子進程(仍為postmaster進程),則檢查數據庫路徑/切換路徑/創建鎖定文件等操作
N/A
7.調用BaseInit執行基本的初始化
3785 BaseInit(); (gdb)
8.調用InitProcess/InitPostgres初始化進程
3797 InitProcess(); (gdb) 3801 PG_SETMASK(&UnBlockSig); (gdb) 3810 InitPostgres(dbname, InvalidOid, username, InvalidOid, NULL, false); (gdb)
9.重置內存上下文,處理加載庫和前后臺消息交互等
(gdb)
3819 if (PostmasterContext)
(gdb)
3821 MemoryContextDelete(PostmasterContext);
(gdb) P PostmasterContext
$6 = (MemoryContext) 0x1a78c60
(gdb) P *PostmasterContext
$7 = {type = T_AllocSetContext, isReset = false, allowInCritSection = false, methods = 0xc93260 <AllocSetMethods>,
parent = 0x1a73aa0, firstchild = 0x1a9a700, prevchild = 0x1a7ac70, nextchild = 0x1a75ab0, name = 0xc2622a "Postmaster",
ident = 0x0, reset_cbs = 0x0}
(gdb) n
3822 PostmasterContext = NULL;
(gdb)
3825 SetProcessingMode(NormalProcessing);
(gdb)
3831 BeginReportingGUCOptions();
(gdb)
3837 if (IsUnderPostmaster && Log_disconnections)
(gdb) p Log_disconnections
$8 = false
(gdb) p
$9 = false
(gdb) n
3841 if (am_walsender)
(gdb)
3842 InitWalSender();
(gdb)
3848 process_session_preload_libraries();
(gdb)
3853 if (whereToSendOutput == DestRemote)
(gdb)
3857 pq_beginmessage(&buf, 'K');
(gdb)
3858 pq_sendint32(&buf, (int32) MyProcPid);
(gdb)
3859 pq_sendint32(&buf, (int32) MyCancelKey);
(gdb)
3860 pq_endmessage(&buf);
(gdb)
3865 if (whereToSendOutput == DestDebug)
(gdb)10.初始化內存上下文
(gdb) 3874 MessageContext = AllocSetContextCreate(TopMemoryContext, (gdb) 3884 row_description_context = AllocSetContextCreate(TopMemoryContext, (gdb) 3887 MemoryContextSwitchTo(row_description_context); (gdb) 3888 initStringInfo(&row_description_buf); (gdb) 3889 MemoryContextSwitchTo(TopMemoryContext); (gdb) 3894 if (!IsUnderPostmaster) (gdb) 3919 if (sigsetjmp(local_sigjmp_buf, 1) != 0) (gdb) 4027 PG_exception_stack = &local_sigjmp_buf; (gdb) 4029 if (!ignore_till_sync) (gdb) 4030 send_ready_for_query = true; /* initially, or after error */ (gdb)
11.進入主循環
11.1切換至MessageContext上下文
(gdb) 4042 doing_extended_query_message = false; (gdb) 4048 MemoryContextSwitchTo(MessageContext); (gdb) 4049 MemoryContextResetAndDeleteChildren(MessageContext);
11.2初始化輸入的消息
(gdb)
4051 initStringInfo(&input_message);
(gdb)
4057 InvalidateCatalogSnapshotConditionally();
(gdb) p input_message
$10 = {data = 0x1a78d78 "", len = 0, maxlen = 1024, cursor = 0}
(gdb)11.3給客戶端發送可以執行查詢等消息
(gdb) n
4072 if (send_ready_for_query)
(gdb) p send_ready_for_query
$12 = true
(gdb) n
4074 if (IsAbortedTransactionBlockState())
(gdb)
4087 else if (IsTransactionOrTransactionBlock())
(gdb)
4102 ProcessCompletedNotifies();
(gdb)
4103 pgstat_report_stat(false);
(gdb)
4105 set_ps_display("idle", false);
(gdb)
4106 pgstat_report_activity(STATE_IDLE, NULL);
(gdb)
4109 ReadyForQuery(whereToSendOutput);
(gdb)
4110 send_ready_for_query = false;
(gdb)11.4讀取命令
命令是IDENTIFY_SYSTEM,判斷系統標識是否OK
firstchar -> ASCII 81 —> 字母’Q’
(gdb)
4119 DoingCommandRead = true;
(gdb)
4124 firstchar = ReadCommand(&input_message);
(gdb)
4135 CHECK_FOR_INTERRUPTS();
(gdb) p input_message
$13 = {data = 0x1a78d78 "IDENTIFY_SYSTEM", len = 16, maxlen = 1024, cursor = 0}
(gdb) p firstchar
$14 = 81
(gdb)
$15 = 81
(gdb) n
4136 DoingCommandRead = false;
(gdb)
4141 if (disable_idle_in_transaction_timeout)
(gdb)
4151 if (ConfigReloadPending)
(gdb)
4161 if (ignore_till_sync && firstchar != EOF)
(gdb)11.5根據命令類型執行相關操作
walsender —> 執行exec_replication_command命令
(gdb) 4164 switch (firstchar) (gdb) 4171 SetCurrentStatementStartTimestamp(); (gdb) 4173 query_string = pq_getmsgstring(&input_message); (gdb) 4174 pq_getmsgend(&input_message); (gdb) p query_string $16 = 0x1a78d78 "IDENTIFY_SYSTEM" (gdb) n 4176 if (am_walsender) (gdb) 4178 if (!exec_replication_command(query_string)) (gdb) 4184 send_ready_for_query = true; (gdb) 4186 break; (gdb) 4411 } /* end of input-reading loop */ (gdb)
繼續循環,接收命令,第二個命令是START_REPLICATION
...
(gdb)
4124 firstchar = ReadCommand(&input_message);
(gdb)
4135 CHECK_FOR_INTERRUPTS();
(gdb) p input_message
$18 = {data = 0x1a78d78 "START_REPLICATION 0/5D000000 TIMELINE 16", len = 41, maxlen = 1024, cursor = 0}
(gdb) p firstchar
$19 = 81
...
4164 switch (firstchar)
(gdb) n
4171 SetCurrentStatementStartTimestamp();
(gdb)
4173 query_string = pq_getmsgstring(&input_message);
(gdb)
4174 pq_getmsgend(&input_message);
(gdb)
4176 if (am_walsender)
(gdb) p query_string
$20 = 0x1a78d78 "START_REPLICATION 0/5D000000 TIMELINE 16"
(gdb) p input_message
$21 = {data = 0x1a78d78 "START_REPLICATION 0/5D000000 TIMELINE 16", len = 41, maxlen = 1024, cursor = 41}
(gdb) n
4178 if (!exec_replication_command(query_string))
(gdb)開始執行復制,master節點使用psql連接數據庫,執行sql語句,子進程會接收到相關信號,執行相關處理
執行腳本
[xdb@localhost ~]$ psql -d testdb psql (11.2) Type "help" for help. testdb=# drop table t1;
子進程輸出
(gdb)
Program received signal SIGUSR1, User defined signal 1.
0x00007fb38696c903 in __epoll_wait_nocancel () from /lib64/libc.so.6
(gdb)
Single stepping until exit from function __epoll_wait_nocancel,
which has no line number information.
procsignal_sigusr1_handler (postgres_signal_arg=32766) at procsignal.c:262
262 {
(gdb) n
263 int save_errno = errno;
(gdb)
Program received signal SIGTRAP, Trace/breakpoint trap.
0x00007fb3881eecd0 in __errno_location () from /lib64/libpthread.so.0
(gdb)
Single stepping until exit from function __errno_location,
which has no line number information.
procsignal_sigusr1_handler (postgres_signal_arg=10) at procsignal.c:265
265 if (CheckProcSignal(PROCSIG_CATCHUP_INTERRUPT))
(gdb)DONE!
DEBUG退出gdb后,psql會話crash:(
[xdb@localhost ~]$ psql -d testdb psql (11.2) Type "help" for help. testdb=# drop table t1; WARNING: terminating connection because of crash of another server process DETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. HINT: In a moment you should be able to reconnect to the database and repeat your command. server closed the connection unexpectedly This probably means the server terminated abnormally before or while processing the request. The connection to the server was lost. Attempting reset: Failed. !>
到此，關于“PostgreSQL的后臺進程walsender分析”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
