PostgreSQL中函數AssignTransactionId的實現邏輯是什么

本篇內容介紹了“PostgreSQL中函數AssignTransactionId的實現邏輯是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

一、數據結構

靜態變量
當前事務狀態CurrentTransactionState

/*
 * CurrentTransactionState always points to the current transaction state
 * block.  It will point to TopTransactionStateData when not in a
 * transaction at all, or when in a top-level transaction.
 * CurrentTransactionState通常指向當前事務塊.
 * 如不處于事務中或者處于頂層事務中,則指向TopTransactionStateData
 */
static TransactionStateData TopTransactionStateData = {
    .state = TRANS_DEFAULT,
    .blockState = TBLOCK_DEFAULT,
};
/*
 * unreportedXids holds XIDs of all subtransactions that have not yet been
 * reported in an XLOG_XACT_ASSIGNMENT record.
 * unreportedXids保存所有尚未在XLOG_XACT_ASSIGNMENT記錄的子事務.
 */
static int  nUnreportedXids;
static TransactionId unreportedXids[PGPROC_MAX_CACHED_SUBXIDS];
static TransactionState CurrentTransactionState = &TopTransactionStateData;
/*
 * The subtransaction ID and command ID assignment counters are global
 * to a whole transaction, so we do not keep them in the state stack.
 * subtransaction ID和command ID全局計數器,對事務可見,在state棧中不記錄這些信息.
 */
static SubTransactionId currentSubTransactionId;
static CommandId currentCommandId;
static bool currentCommandIdUsed;

TransactionState
事務狀態結構體

/*
 *  transaction states - transaction state from server perspective
 *  事務狀態枚舉 - 服務器視角的事務狀態
 */
typedef enum TransState
{
    TRANS_DEFAULT,              /* idle 空閑 */
    TRANS_START,                /* transaction starting 事務啟動 */
    TRANS_INPROGRESS,           /* inside a valid transaction 進行中 */
    TRANS_COMMIT,               /* commit in progress 提交中 */
    TRANS_ABORT,                /* abort in progress 回滾中 */
    TRANS_PREPARE               /* prepare in progress 準備中 */
} TransState;
/*
 *  transaction block states - transaction state of client queries
 *  事務塊狀態 - 客戶端查詢的事務狀態
 *
 * Note: the subtransaction states are used only for non-topmost
 * transactions; the others appear only in the topmost transaction.
 * 注意:subtransaction只用于非頂層事務;其他字段用于頂層事務.
 */
typedef enum TBlockState
{
    /* not-in-transaction-block states 未進入事務塊狀態 */
    TBLOCK_DEFAULT,             /* idle 空閑  */
    TBLOCK_STARTED,             /* running single-query transaction 單個查詢事務 */
    /* transaction block states 事務塊狀態 */
    TBLOCK_BEGIN,               /* starting transaction block 開始事務塊 */
    TBLOCK_INPROGRESS,          /* live transaction 進行中 */
    TBLOCK_IMPLICIT_INPROGRESS, /* live transaction after implicit BEGIN 隱式事務,進行中 */
    TBLOCK_PARALLEL_INPROGRESS, /* live transaction inside parallel worker 并行worker中的事務,進行中 */
    TBLOCK_END,                 /* COMMIT received 接收到COMMIT */
    TBLOCK_ABORT,               /* failed xact, awaiting ROLLBACK 失敗,等待ROLLBACK */
    TBLOCK_ABORT_END,           /* failed xact, ROLLBACK received 失敗,已接收ROLLBACK */
    TBLOCK_ABORT_PENDING,       /* live xact, ROLLBACK received 進行中,接收到ROLLBACK */
    TBLOCK_PREPARE,             /* live xact, PREPARE received 進行中,接收到PREPARE */
    /* subtransaction states 子事務狀態 */
    TBLOCK_SUBBEGIN,            /* starting a subtransaction 開啟 */
    TBLOCK_SUBINPROGRESS,       /* live subtransaction 進行中 */
    TBLOCK_SUBRELEASE,          /* RELEASE received 接收到RELEASE */
    TBLOCK_SUBCOMMIT,           /* COMMIT received while TBLOCK_SUBINPROGRESS 進行中,接收到COMMIT */
    TBLOCK_SUBABORT,            /* failed subxact, awaiting ROLLBACK 失敗,等待ROLLBACK */
    TBLOCK_SUBABORT_END,        /* failed subxact, ROLLBACK received 失敗,已接收ROLLBACK */
    TBLOCK_SUBABORT_PENDING,    /* live subxact, ROLLBACK received 進行中,接收到ROLLBACK */
    TBLOCK_SUBRESTART,          /* live subxact, ROLLBACK TO received 進行中,接收到ROLLBACK TO */
    TBLOCK_SUBABORT_RESTART     /* failed subxact, ROLLBACK TO received 失敗,已接收ROLLBACK TO */
} TBlockState;
/*
 *  transaction state structure
 *  事務狀態結構體
 */
typedef struct TransactionStateData
{
    //事務ID
    TransactionId transactionId;    /* my XID, or Invalid if none */
    //子事務ID
    SubTransactionId subTransactionId;  /* my subxact ID */
    //保存點名稱
    char       *name;           /* savepoint name, if any */
    //保存點級別
    int         savepointLevel; /* savepoint level */
    //低級別的事務狀態
    TransState  state;          /* low-level state */
    //高級別的事務狀態
    TBlockState blockState;     /* high-level state */
    //事務嵌套深度
    int         nestingLevel;   /* transaction nesting depth */
    //GUC上下文嵌套深度
    int         gucNestLevel;   /* GUC context nesting depth */
    //事務生命周期上下文
    MemoryContext curTransactionContext;    /* my xact-lifetime context */
    //查詢資源
    ResourceOwner curTransactionOwner;  /* my query resources */
    //按XID順序保存的已提交的子事務ID
    TransactionId *childXids;   /* subcommitted child XIDs, in XID order */
    //childXids數組大小
    int         nChildXids;     /* # of subcommitted child XIDs */
    //分配的childXids數組空間
    int         maxChildXids;   /* allocated size of childXids[] */
    //上一個CurrentUserId
    Oid         prevUser;       /* previous CurrentUserId setting */
    //上一個SecurityRestrictionContext
    int         prevSecContext; /* previous SecurityRestrictionContext */
    //上一事務是否只讀?
    bool        prevXactReadOnly;   /* entry-time xact r/o state */
    //是否處于Recovery?
    bool        startedInRecovery;  /* did we start in recovery? */
    //XID是否已保存在WAL Record中?
    bool        didLogXid;      /* has xid been included in WAL record? */
    //Enter/ExitParallelMode計數器
    int         parallelModeLevel;  /* Enter/ExitParallelMode counter */
    //父事務狀態
    struct TransactionStateData *parent;    /* back link to parent */
} TransactionStateData;
//結構體指針
typedef TransactionStateData *TransactionState;

二、源碼解讀

AssignTransactionId函數,給定的TransactionState分配一個新的持久化事務號XID,在此函數調用前,不會為事務分配XIDs.

/*
 * AssignTransactionId
 *
 * Assigns a new permanent XID to the given TransactionState.
 * We do not assign XIDs to transactions until/unless this is called.
 * Also, any parent TransactionStates that don't yet have XIDs are assigned
 * one; this maintains the invariant that a child transaction has an XID
 * following its parent's.
 * 為給定的TransactionState分配一個新的持久化事務號XID.
 * 在此函數調用前,我們不會為事務分配XIDs.
 * 同時,所有尚未獲得XIDs的父TransactionStates也會分配事務號,
 *   這可以確保子事務的事務號在父事務之后.
 */
static void
AssignTransactionId(TransactionState s)
{
    bool        isSubXact = (s->parent != NULL);
    ResourceOwner currentOwner;
    bool        log_unknown_top = false;
    /* Assert that caller didn't screw up */
    //確保調用者沒有搞砸
    Assert(!TransactionIdIsValid(s->transactionId));
    Assert(s->state == TRANS_INPROGRESS);
    /*
     * Workers synchronize transaction state at the beginning of each parallel
     * operation, so we can't account for new XIDs at this point.
     * 在每個并行操作前,Parallel Workers同步事務狀態,
     *   因此我們不能在這時候請求XIDs
     */
    if (IsInParallelMode() || IsParallelWorker())
        elog(ERROR, "cannot assign XIDs during a parallel operation");
    /*
     * Ensure parent(s) have XIDs, so that a child always has an XID later
     * than its parent.  Mustn't recurse here, or we might get a stack overflow
     * if we're at the bottom of a huge stack of subtransactions none of which
     * have XIDs yet.
     * 確保父事務已分配XIDs,這樣可以確保子事務的XID后于父事務.
     * 不能在這里遞歸執行,否則如果我們在一個未分配XID子事務大棧的底部,可能會遇到棧溢出
     */
    if (isSubXact && !TransactionIdIsValid(s->parent->transactionId))
    {
        TransactionState p = s->parent;
        TransactionState *parents;
        size_t      parentOffset = 0;
        parents = palloc(sizeof(TransactionState) * s->nestingLevel);
        while (p != NULL && !TransactionIdIsValid(p->transactionId))
        {
            parents[parentOffset++] = p;
            p = p->parent;
        }
        /*
         * This is technically a recursive call, but the recursion will never
         * be more than one layer deep.
         * 遞歸調用,但遞歸不會超過一層
         */
        while (parentOffset != 0)
            AssignTransactionId(parents[--parentOffset]);
        pfree(parents);
    }
    /*
     * When wal_level=logical, guarantee that a subtransaction's xid can only
     * be seen in the WAL stream if its toplevel xid has been logged before.
     * If necessary we log an xact_assignment record with fewer than
     * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set
     * for a transaction even though it appears in a WAL record, we just might
     * superfluously log something. That can happen when an xid is included
     * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in
     * xl_standby_locks.
     * 如wal_level=logical,確保子事務XID在頂層XID已被"日志"的情況只可以被WAL stream看到.
     * 如果有必要，我們將使用小于PGPROC_MAX_CACHED_SUBXIDS的值來記錄xact_assignment記錄.
     * 請注意，即使didLogXid出現在WAL記錄中，如果它沒有為事務設置，也沒有問題，
     *   我們可能只是多余地記錄了一些東西。
     * 當一個xid出現在WAL Record中的某個地方，但不在XLogRecord->xl_xid中
     *   (比如在xl_standby_locks中)時，就會發生這種情況。
     */
    if (isSubXact && XLogLogicalInfoActive() &&
        !TopTransactionStateData.didLogXid)
        log_unknown_top = true;
    /*
     * Generate a new Xid and record it in PG_PROC and pg_subtrans.
     * 生成一個新的XID,并記錄在PG_PROC和pg_subtrans中
     *
     * NB: we must make the subtrans entry BEFORE the Xid appears anywhere in
     * shared storage other than PG_PROC; because if there's no room for it in
     * PG_PROC, the subtrans entry is needed to ensure that other backends see
     * the Xid as "running".  See GetNewTransactionId.
     * 注意:我們必須在Xid出現在除PG_PROC之外的共享存儲之前構造subtrans條目.
     * 因為如果在PG_PROC沒有空閑空間,子事務條目需要確保其他進程看到該XID正在運行.
     * 參考函數GetNewTransactionId說明.
     * 
     */
    s->transactionId = GetNewTransactionId(isSubXact);
    if (!isSubXact)
        XactTopTransactionId = s->transactionId;
    if (isSubXact)
        SubTransSetParent(s->transactionId, s->parent->transactionId);
    /*
     * If it's a top-level transaction, the predicate locking system needs to
     * be told about it too.
     * 如為頂層事務,謂詞鎖系統也需要了解此事務.
     */
    if (!isSubXact)
        RegisterPredicateLockingXid(s->transactionId);
    /*
     * Acquire lock on the transaction XID.  (We assume this cannot block.) We
     * have to ensure that the lock is assigned to the transaction's own
     * ResourceOwner.
     * 請求鎖(我們假定這樣做不好引起阻塞).我們必須確保鎖已被分配給事務自己的ResourceOwner.
     */
    currentOwner = CurrentResourceOwner;
    CurrentResourceOwner = s->curTransactionOwner;
    XactLockTableInsert(s->transactionId);
    CurrentResourceOwner = currentOwner;
    /*
     * Every PGPROC_MAX_CACHED_SUBXIDS assigned transaction ids within each
     * top-level transaction we issue a WAL record for the assignment. We
     * include the top-level xid and all the subxids that have not yet been
     * reported using XLOG_XACT_ASSIGNMENT records.
     * 在每個頂級事務中分配的每個PGPROC_MAX_CACHED_SUBXIDS事務id，
     *   我們都會為分配記錄一條WAL記錄。
     * 該記錄包括頂級的xid和所有尚未使用XLOG_XACT_ASSIGNMENT記錄報告的子xid。
     *
     * This is required to limit the amount of shared memory required in a hot
     * standby server to keep track of in-progress XIDs. See notes for
     * RecordKnownAssignedTransactionIds().
     * 在跟蹤進行中的XIDs的備機上,需要控制共享內存的大小.
     * 參見RecordKnownAssignedTransactionIds()函數說明.
     *
     * We don't keep track of the immediate parent of each subxid, only the
     * top-level transaction that each subxact belongs to. This is correct in
     * recovery only because aborted subtransactions are separately WAL
     * logged.
     * 我們不需要跟蹤父事務的每個子事務,只需要跟蹤子事務歸屬的頂層事務即可.
     * 這樣可行是因為在恢復中,已回滾的子事務是通過WAL單獨記錄的.
     *
     * This is correct even for the case where several levels above us didn't
     * have an xid assigned as we recursed up to them beforehand.
     * 即使在我們之前遞歸到上面的幾個級別時沒有分配xid的情況下，這也是正確的。
     */
    if (isSubXact && XLogStandbyInfoActive())
    {
        unreportedXids[nUnreportedXids] = s->transactionId;
        nUnreportedXids++;
        /*
         * ensure this test matches similar one in
         * RecoverPreparedTransactions()
         * 確保在RecoverPreparedTransactions()中可以匹配到相似的.
         */
        if (nUnreportedXids >= PGPROC_MAX_CACHED_SUBXIDS ||
            log_unknown_top)
        {
            xl_xact_assignment xlrec;
            /*
             * xtop is always set by now because we recurse up transaction
             * stack to the highest unassigned xid and then come back down
             * xtop現在已經設置好了，因為我們將事務堆棧遞歸到最高的未分配的xid，然后再返回
             */
            xlrec.xtop = GetTopTransactionId();
            Assert(TransactionIdIsValid(xlrec.xtop));
            xlrec.nsubxacts = nUnreportedXids;
            XLogBeginInsert();
            XLogRegisterData((char *) &xlrec, MinSizeOfXactAssignment);
            XLogRegisterData((char *) unreportedXids,
                             nUnreportedXids * sizeof(TransactionId));
            (void) XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
            nUnreportedXids = 0;
            /* mark top, not current xact as having been logged */
            //標記為最頂層,而不是當前已記錄日志的xact
            TopTransactionStateData.didLogXid = true;
        }
    }
}

三、跟蹤分析

執行txid_current,觸發函數調用

11:10:36 (xdb@[local]:5432)testdb=# begin;
BEGIN
11:40:20 (xdb@[local]:5432)testdb=#* select txid_current_if_assigned();
 txid_current_if_assigned 
--------------------------
(1 row)
11:40:43 (xdb@[local]:5432)testdb=#* select txid_current();

啟動gdb,設置斷點

(gdb) b AssignTransactionId
Breakpoint 5 at 0x546a4c: file xact.c, line 491.
(gdb) c
Continuing.
Breakpoint 5, AssignTransactionId (s=0xf9c720 <TopTransactionStateData>) at xact.c:491
491     bool        isSubXact = (s->parent != NULL);
(gdb)

查看調用棧

(gdb) bt
#0  AssignTransactionId (s=0xf9c720 <TopTransactionStateData>) at xact.c:491
#1  0x000000000054693d in GetTopTransactionId () at xact.c:392
#2  0x00000000009fe1f3 in txid_current (fcinfo=0x25835a0) at txid.c:443
#3  0x00000000006cfebd in ExecInterpExpr (state=0x25834b8, econtext=0x25831a8, isnull=0x7ffe3d4a31f7)
    at execExprInterp.c:654
#4  0x00000000006d1ac6 in ExecInterpExprStillValid (state=0x25834b8, econtext=0x25831a8, isNull=0x7ffe3d4a31f7)
    at execExprInterp.c:1786
#5  0x00000000007140dd in ExecEvalExprSwitchContext (state=0x25834b8, econtext=0x25831a8, isNull=0x7ffe3d4a31f7)
    at ../../../src/include/executor/executor.h:303
#6  0x000000000071414b in ExecProject (projInfo=0x25834b0) at ../../../src/include/executor/executor.h:337
#7  0x0000000000714323 in ExecResult (pstate=0x2583090) at nodeResult.c:136
#8  0x00000000006e4c30 in ExecProcNodeFirst (node=0x2583090) at execProcnode.c:445
#9  0x00000000006d9974 in ExecProcNode (node=0x2583090) at ../../../src/include/executor/executor.h:237
#10 0x00000000006dc22d in ExecutePlan (estate=0x2582e78, planstate=0x2583090, use_parallel_mode=false, 
    operation=CMD_SELECT, sendTuples=true, numberTuples=0, direction=ForwardScanDirection, dest=0x24cd0a0, 
    execute_once=true) at execMain.c:1723
#11 0x00000000006d9f5c in standard_ExecutorRun (queryDesc=0x256b0c8, direction=ForwardScanDirection, count=0, 
    execute_once=true) at execMain.c:364
#12 0x00000000006d9d7f in ExecutorRun (queryDesc=0x256b0c8, direction=ForwardScanDirection, count=0, execute_once=true)
    at execMain.c:307
#13 0x00000000008ccf5a in PortalRunSelect (portal=0x250c748, forward=true, count=0, dest=0x24cd0a0) at pquery.c:932
#14 0x00000000008ccbf3 in PortalRun (portal=0x250c748, count=9223372036854775807, isTopLevel=true, run_once=true, 
    dest=0x24cd0a0, altdest=0x24cd0a0, completionTag=0x7ffe3d4a3570 "") at pquery.c:773
#15 0x00000000008c6b1e in exec_simple_query (query_string=0x24a6ec8 "select txid_current();") at postgres.c:1145
#16 0x00000000008cae70 in PostgresMain (argc=1, argv=0x24d2dc8, dbname=0x24d2c30 "testdb", username=0x24a3ba8 "xdb")
    at postgres.c:4182
    #17 0x000000000082642b in BackendRun (port=0x24c8c00) at postmaster.c:4361
---Type <return> to continue, or q <return> to quit---
#18 0x0000000000825b8f in BackendStartup (port=0x24c8c00) at postmaster.c:4033
#19 0x0000000000821f1c in ServerLoop () at postmaster.c:1706
#20 0x00000000008217b4 in PostmasterMain (argc=1, argv=0x24a1b60) at postmaster.c:1379
#21 0x00000000007488ef in main (argc=1, argv=0x24a1b60) at main.c:228

輸入參數TransactionState(全局變量,指向TopTransactionStateData)

(gdb) p s
$13 = (TransactionState) 0xf9c720 <TopTransactionStateData>
(gdb) p *s
$14 = {transactionId = 0, subTransactionId = 1, name = 0x0, savepointLevel = 0, state = TRANS_INPROGRESS, 
  blockState = TBLOCK_INPROGRESS, nestingLevel = 1, gucNestLevel = 1, curTransactionContext = 0x2523850, 
  curTransactionOwner = 0x24d4868, childXids = 0x0, nChildXids = 0, maxChildXids = 0, prevUser = 10, prevSecContext = 0, 
  prevXactReadOnly = false, startedInRecovery = false, didLogXid = false, parallelModeLevel = 0, parent = 0x0}
(gdb)

初始化部分變量并驗證

(gdb) n
493     bool        log_unknown_top = false;
(gdb) 
496     Assert(!TransactionIdIsValid(s->transactionId));
(gdb) 
497     Assert(s->state == TRANS_INPROGRESS);
(gdb)

獲取事務號

(gdb) 
503     if (IsInParallelMode() || IsParallelWorker())
(gdb) 
512     if (isSubXact && !TransactionIdIsValid(s->parent->transactionId))
(gdb) 
545     if (isSubXact && XLogLogicalInfoActive() &&
(gdb) 
557     s->transactionId = GetNewTransactionId(isSubXact);
(gdb) 
558     if (!isSubXact)
(gdb) p s->transactionId
$15 = 2407
(gdb)

注冊,并設置其他信息

(gdb) n
559         XactTopTransactionId = s->transactionId;
(gdb) 
561     if (isSubXact)
(gdb) 
568     if (!isSubXact)
(gdb) 
569         RegisterPredicateLockingXid(s->transactionId);
(gdb) 
576     currentOwner = CurrentResourceOwner;
(gdb) 
577     CurrentResourceOwner = s->curTransactionOwner;
(gdb) 
579     XactLockTableInsert(s->transactionId);
(gdb) 
581     CurrentResourceOwner = currentOwner;
(gdb) 
601     if (isSubXact && XLogStandbyInfoActive())
(gdb) 
635 }

完成調用

(gdb) 
GetTopTransactionId () at xact.c:393
393     return XactTopTransactionId;
(gdb)

“PostgreSQL中函數AssignTransactionId的實現邏輯是什么”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！