openGauss数据库源码解析系列文章——事务机制源码解析（二）

typedef struct SnapshotData {    SnapshotSatisfiesFunc satisfies;  /* 判断可见性的函数；通常使用MVCC，即HeapTupleSatisfiesMVCC */    TransactionId xmin; /*当前活跃事务最小值，小于该值的事务说明已结束  */    TransactionId xmax; /*最新提交事务id（latestCompeleteXid）+1，大于等于改值说明事务还未开始，该事务id不可见  */    TransactionId* xip; /*记录当前活跃事务链表，在CSN版本中该值无用  */    TransactionId* subxip; /* 记录缓存子事务活跃链表，在CSN版本中该值无用  */    uint32 xcnt; /* 记录活跃事务的个数（xip中元组数）在CSN版本中该值无用  */    GTM_Timeline timeline; /* openGauss单机中无用  */    uint32 max_xcnt;       /* xip的最大个数，CSN版本中该值无用  */    int32 subxcnt; /* 缓存子事务活跃链表的个数，在CSN版本中该值无用  */    int32 maxsubxcnt;  /* 缓存子事务活跃链表最大个数，在CSN版本中该值无用  */    bool suboverflowed; /* 子事务活跃链表是否已超过共享内存中预分配的上限，在CSN版本中无用。  */     CommitSeqNo snapshotcsn; /* 快照的CSN号，一般为最新提交事务的CSN号+1(NextCommitSeqNo)，CSN号严格小于该值的事务可见。  */     int prepared_array_capacity; /*  单机openGauss无用  */    int prepared_count; /* 单机openGauss无用  */    TransactionId* prepared_array; /* 单机openGauss无用  */     bool takenDuringRecovery;  /* 是否Recovery过程中产生的快照  */    bool copied; /* 该快照是会话级别静态的，还是新分配内存拷贝的  */     CommandId curcid; /*事务块中的命令序列号，即同一事务中，前面插入的数据，后面可见。  */    uint32 active_count; /* ActiveSnapshot stack的refcount */    uint32 regd_count;   /* RegisteredSnapshotList 的refcount*/    void* user_data;     /* 本地多版本快照使用，标记该快照还有线程使用，不能直接释放 */    SnapshotType snapshot_type; /*  openGauss单机无用  */} SnapshotData;

bool HeapTupleSatisfiesMVCC(HeapTuple htup, Snapshot snapshot, Buffer buffer){    …… /*  初始化变量  */     if (!HeapTupleHeaderXminCommitted(tuple)) { /*  此处先判断用一个bit位记录的hint bit（提示比特位：openGauss判断可见性时，通常需要知道元组xmin和xmax对应的clog的提交状态；为了避免重复访问clog，openGauss内部对可见性判断进行了优化。hint bit是把事务状态直接记录在元组头中，用一个bit位来表示提交和回滚状态。openGauss并不会在事务提交或者回滚时主动更新元组上的 hint bit，而是等到访问该元组并进行可见性判断时，如果发现hint bit没有设置，则从 CLOG 中读取并设置，否则直接读取hint bit的值），防止同一条tuple反复获取事务最终提交状态。如果一次扫描发现该元组的xmin/xmax已经提交，就会打上相应的标记，加速扫描；如果没有标记则继续判断。  */        if (HeapTupleHeaderXminInvalid(tuple)) /*  同样判断hint bit。如果xmin已经标记为invalid说明插入该元组的事务已经回滚，直接返回不可见  */            return false;         if (TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(page, tuple))) { /*  如果是一个事务内部，需要去判断该元组的CID，也即是同一个事务内，后面的查询可以查到当前事务之前插入的扫描结果  */          …….        } else {  /*  如果扫描别的事务，需要根据快照判断事务是否可见  */            visible = XidVisibleInSnapshot(HeapTupleHeaderGetXmin(page, tuple), snapshot, &hintstatus); /*  通过csnlog判断事务是否可见，并且返回该事务的最终提交状态  */            if (hintstatus == XID_COMMITTED) /*  如果该事务提交，则打上提交的hint bit用于加速判断  */                SetHintBits(tuple, buffer, HEAP_XMIN_COMMITTED, HeapTupleHeaderGetXmin(page, tuple));             if (hintstatus == XID_ABORTED) {                … /*  如果事务回滚，则打上回滚标记  */                SetHintBits(tuple, buffer, HEAP_XMIN_INVALID, InvalidTransactionId);            }            if (!visible) {  /*  如果xmin不可见，则该元组不可见，否则表示插入该元组的事务对于该次快照已经提交，继续判断删除该元组的事务是否对该次快照提交  */                    return false;                }            }        }    } else { /* 如果该条元组的xmin已经被打上提交的hint bit，则通过函数接口CommittedXidVisibleInSnapshot判断是否对本次快照可见  */        /* xmin is committed, but maybe not according to our snapshot */        if (!HeapTupleHeaderXminFrozen(tuple) &&            !CommittedXidVisibleInSnapshot(HeapTupleHeaderGetXmin(page, tuple), snapshot)) {           return false;        }    }   …… /*  后续xmax的判断同xmin类似，如果xmax对于本次快照可见，则说明删除该条元组的事务已经提交，则不可见，否则可见，此处不再赘述 */    if (!(tuple->t_infomask & HEAP_XMAX_COMMITTED)) {        if (TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmax(page, tuple))) {            if (HeapTupleHeaderGetCmax(tuple, page) >= snapshot->curcid)                return true; /* 在扫面前该删除事务已经提交 */            else                return false; /* 扫描开始后删除操作的事务才提交 */        }         visible = XidVisibleInSnapshot(HeapTupleHeaderGetXmax(page, tuple), snapshot, &hintstatus);        if (hintstatus == XID_COMMITTED) {            /* 设置xmax的hint bit */            SetHintBits(tuple, buffer, HEAP_XMAX_COMMITTED, HeapTupleHeaderGetXmax(page, tuple));        }        if (hintstatus == XID_ABORTED) {            /* 回滚或者故障 */            SetHintBits(tuple, buffer, HEAP_XMAX_INVALID, InvalidTransactionId);        }        if (!visible) {            return true; /* 快照中xmax对应的事务不可见，则认为该元组仍然活跃 */        }    } else {        /* xmax对应的事务已经提交，但是快照中该事务不可见，认为删除该元组的操作未完成，仍然认为该元组可见 */        if (!CommittedXidVisibleInSnapshot(HeapTupleHeaderGetXmax(page, tuple), snapshot)) {            return true; /* 认为元组可见 */        }    }    return false;}

HTSV_Result HeapTupleSatisfiesVacuum(HeapTuple htup, TransactionId OldestXmin, Buffer buffer){    ……  /* 初始化变量    if (!HeapTupleHeaderXminCommitted(tuple)) {  /* hint bit标记加速，与MVCC的逻辑相同。  */        if (HeapTupleHeaderXminInvalid(tuple))  /* 如果xmin未提交，则返回该元组死亡，可以清理。 */            return HEAPTUPLE_DEAD;        xidstatus = TransactionIdGetStatus(HeapTupleGetRawXmin(htup), false);  /* 通过CSNLog来获取当前的事务状态  */        if (xidstatus == XID_INPROGRESS) {            if (tuple->t_infomask & HEAP_XMAX_INVALID)  /* 如果xmax还没有，说明没有人删除，此时判断该元组正在插入过程中，否则在删除过程中  */                return HEAPTUPLE_INSERT_IN_PROGRESS;            return HEAPTUPLE_DELETE_IN_PROGRESS;  /* 返回正在删除的过程中  */        } else if (xidstatus == XID_COMMITTED) {  /* 如果xmin提交了，打上hint bit，后面继续看xmax是否提交。  */            SetHintBits(tuple, buffer, HEAP_XMIN_COMMITTED, HeapTupleGetRawXmin(htup));        } else {   ….  /* 事务结束了且未提交，可能是abort或者是crash的事务，一般返回死亡，可删除；单机情形 t_thrd.xact_cxt.useLocalSnapshot没有作用，恒为false。 */            SetHintBits(tuple, buffer, HEAP_XMIN_INVALID, InvalidTransactionId);            return ((!t_thrd.xact_cxt.useLocalSnapshot || IsInitdb) ? HEAPTUPLE_DEAD : HEAPTUPLE_LIVE);        }    }  /* 接着判断xmax。如果还没有设置xmax说明没有人删除该元组，返回元组存活，不可删除。  */    if (tuple->t_infomask & HEAP_XMAX_INVALID)        return HEAPTUPLE_LIVE;……    if (!(tuple->t_infomask & HEAP_XMAX_COMMITTED)) {   /*如果xmax提交，则看xmax是否比oldesxmin小。小的话说明没有未结束的事务会访问该元组，可以删除。 */        xidstatus = TransactionIdGetStatus(HeapTupleGetRawXmax(htup), false);        if (xidstatus == XID_INPROGRESS)            return HEAPTUPLE_DELETE_IN_PROGRESS;         else if (xidstatus == XID_COMMITTED)            SetHintBits(tuple, buffer, HEAP_XMAX_COMMITTED, HeapTupleGetRawXmax(htup));        else {…  /*  xmax对应的事务abort或者crash */            SetHintBits(tuple, buffer, HEAP_XMAX_INVALID, InvalidTransactionId);            return HEAPTUPLE_LIVE;        }    }   /*判断该元组是否可以删除，xmax<OldestXmin可以删除。  */    if (!TransactionIdPrecedes(HeapTupleGetRawXmax(htup), OldestXmin))        return ((!t_thrd.xact_cxt.useLocalSnapshot || IsInitdb) ? HEAPTUPLE_RECENTLY_DEAD : HEAPTUPLE_LIVE);     /* 该元组可以认为已经DEAD，不被任何活跃事务访问，可以删除。 */    return ((!t_thrd.xact_cxt.useLocalSnapshot || IsInitdb) ? HEAPTUPLE_DEAD : HEAPTUPLE_LIVE);}

typedef struct PGXACT {    GTM_TransactionHandle handle; /* 单机模式无用参数 */    TransactionId xid;              /* 该线程持有的xid号,如果没有则为0 */    TransactionId prepare_xid;    /* 准备阶段的xid号*/ TransactionId xmin; /* 当前事务开启时最小的活跃xid，vaccum操作不会删除那些xid大于等于xmin的元组。  */    CommitSeqNo csn_min;   /* 当前事务开启时最小的活跃CSN号*/    TransactionId next_xid; /* 单机模式无用参数*/    int nxids;  /* 子事物个数*/    uint8 vacuumFlags; /* vacuum操作相关的flag */ bool needToSyncXid; /* 单机模式无用参数*/bool delayChkpt; /* 如果该线程需要checkpoint线程延迟等待，此值为true#ifdef __aarch64__ */    char padding[PG_CACHE_LINE_SIZE - PGXACT_PAD_OFFSET]; /* 为了性能考虑的结构体对齐*/#endif} PGXACT; struct PGPROC {    SHM_QUEUE links; /* 链表中的指针 */     PGSemaphoreData sem; /* 休眠等待的信号量 */    int waitStatus;      /* 等待状态 */     Latch procLatch; /* 线程的通用闩锁 */     LocalTransactionId lxid; /* 当前线程本地顶层事务ID */    ThreadId pid;            /* 线程的PID */     ThreadId sessMemorySessionid;    uint64 sessionid; /* 线程池模式下当前的会话ID */    int logictid;     /* 逻辑线程ID */    TransactionId gtt_session_frozenxid;    /* 会话级全局临时表的冻结XID */     int pgprocno;    int nodeno;     /* 线程启动时下面这些数据结构为0 */    BackendId backendId; /* 线程的后台ID */    Oid databaseId;      /* 当前访问数据库的OID */    Oid roleId;          /* 当前用户的OID */     /* 版本号，用于升级过程中新老版本的判断 */    uint32 workingVersionNum;     /*热备模式下，标记当前事务是否收到冲突信号。设置该值时需要持有ProcArray锁。 */    bool recoveryConflictPending;     /* 线程等待的轻量级锁信息. */    bool lwWaiting;        /* 当等待轻量级锁时，为真 */    uint8 lwWaitMode;      /* 预获取锁的模式 */    bool lwIsVictim;       /* 强制放弃轻量级锁 */    dlist_node lwWaitLink; /* 等待在相同轻量级锁对象的下一个等待者 */ /* 线程等待的常规锁信息 */    LOCK* waitLock;         /* 等待的常规锁对象 */    PROCLOCK* waitProcLock; /* 等待常规锁对象的持有者 */    LOCKMODE waitLockMode;  /* 预获取常规锁对象的模式 */    LOCKMASK heldLocks;     /* 本线程获取锁对象模式的位掩码 */     /* 等待主备机回放日志同步的信息 */    XLogRecPtr waitLSN;     /* 等待的lsn*/    int syncRepState;       /* 等待主备同步的状态 */    bool syncRepInCompleteQueue;   /* 是否等待在完成队列中 */    SHM_QUEUE syncRepLinks; /* 指向同步队列的指针 */     DataQueuePtr waitDataSyncPoint; /* 数据页复制的数据同步点 */    int dataSyncRepState;           /* 数据页复制的同步状态 */    SHM_QUEUE dataSyncRepLinks;     /* 指向数据页同步队列的指针*/     MemoryContext topmcxt; /* 本线程的顶层内存上下文 */    char myProgName[64];    pg_time_t myStartTime;    syscalllock deleMemContextMutex;     SHM_QUEUE myProcLocks[NUM_LOCK_PARTITIONS];     /* 以下结构为了实现XID批量提交 */    /* 是否为XID批量提交中的成员 */    bool procArrayGroupMember;    /* XID批量提交中的下一个成员 */    pg_atomic_uint32 procArrayGroupNext;    /* 父事务XID和子事物XID中的最大者 */    TransactionId procArrayGroupMemberXid;     /* 提交序列号 */    CommitSeqNo commitCSN;     /* 以下结构为了实现CLOG批量提交 */    bool clogGroupMember;  /* 是否为CLOG批量提交中的成员*/    pg_atomic_uint32 clogGroupNext;  /* CLOG批量提交中的下一个成员*/    TransactionId clogGroupMemberXid;  /* CLOG批量提交的事务ID */    CLogXidStatus clogGroupMemberXidStatus; /* CLOG批量提交的事务状态 */    int64 clogGroupMemberPage;  /* CLOG批量提交对应的CLOG页面 */    XLogRecPtr clogGroupMemberLsn;  /* CLOG批量提交成员的提交回放日志位置 */#ifdef __aarch64__      /* 以下结构体是为了实现ARM架构下回放日志批量插入 */    bool xlogGroupMember;    pg_atomic_uint32 xlogGroupNext;    XLogRecData* xlogGrouprdata;    XLogRecPtr xlogGroupfpw_lsn;    XLogRecPtr* xlogGroupProcLastRecPtr;    XLogRecPtr* xlogGroupXactLastRecEnd;    void* xlogGroupCurrentTransactionState;    XLogRecPtr* xlogGroupRedoRecPtr;    void* xlogGroupLogwrtResult;    XLogRecPtr xlogGroupReturntRecPtr;    TimeLineID xlogGroupTimeLineID;    bool* xlogGroupDoPageWrites;    bool xlogGroupIsFPW;    uint64 snap_refcnt_bitmap;#endif     LWLock* subxidsLock;struct XidCache subxids; /* 子事物XID */     LWLock* backendLock; /* 每个线程的轻量级锁，用于保护以下数据结构的并发访问 */     /* Lock manager data, recording fast-path locks taken by this backend. */    uint64 fpLockBits;  /* 快速路径锁的持有模式 */    FastPathTag fpRelId[FP_LOCK_SLOTS_PER_BACKEND]; /* 表对象的槽位 */    bool fpVXIDLock;  /* 是否获得本地XID的快速路径锁 */    LocalTransactionId fpLocalTransactionId; /* 本地的XID */};

typedef struct _snapxid {    TransactionId xmin;    TransactionId xmax;    CommitSeqNo snapshotcsn;    TransactionId localxmin;    bool takenDuringRecovery;    ref_cnt_t ref_cnt[NREFCNT]; /* 该快照的引用计数，如果为0则可复用 */} snapxid_t;  /*多版本快照内容，在openGauss CSN方案下，仅需要记录xmin、xmax、snapshotcsn等关键信息即可。*/ static snapxid_t* g_snap_buffer = NULL;     /*   缓冲区队列内存区指针  */static snapxid_t* g_snap_buffer_copy = NULL; /*   缓冲区队列内存的浅拷贝 */static size_t g_bufsz = 0;static bool g_snap_assigned = false;  /*多版本快照buffer队列是否已初始化 */ #define SNAP_SZ sizeof(snapxid_t) /*  每一个多版本快照的size大小  */#define MaxNumSnapVersion 64    /* 多版本快照队列的大小，64个版本  */ static volatile snapxid_t* g_snap_current = NULL; /*  当前的快照指针  */static volatile snapxid_t* g_snap_next = NULL; /*  下一个可用槽位的快照指针  */

static inline snapxid_t* GetNextSnapXid(){return g_snap_buffer ? (snapxid_t*)g_snap_next : NULL;}

static void SetNextSnapXid(){    if (g_snap_buffer != NULL) {        g_snap_current = g_snap_next; /* 将最新的多版本快照更新到最新。*/        pg_write_barrier(); /* 此处是防止buffer ring初始化时的ARM乱序问题。*/        g_snap_assigned = true;        snapxid_t* ret = (snapxid_t*)g_snap_current;        size_t idx = SNAPXID_INDEX(ret);    loop: /* 主循环，整体思路是不停遍历多版本槽位信息，一直找到一个refcout为0的可重用槽位。*/        do {            ++idx;            /* 如果发生回卷，那么重头再找 */            if (idx == g_bufsz)                idx = 0;            ret = SNAPXID_AT(idx);            if (IsZeroRefCount(ret)) {                g_snap_next = ret;                return;            }        } while (ret != g_snap_next);        ereport(WARNING, (errmsg("snapshot ring buffer overflow.")));/* 当前多版本快照个数为64个，理论上可能是会出现槽位被占满，如果没有空闲槽位，重新遍历即可。 */        goto loop;    }}

void CalculateLocalLatestSnapshot(bool forceCalc){    …/* 初始化变量  */     snapxid_t* snapxid = GetNextSnapXid(); /*设置下一个空闲多版本快照槽位信息 */     /* 初始化xmax为 latestCompletedXid + 1 */    xmax = t_thrd.xact_cxt.ShmemVariableCache->latestCompletedXid;    TransactionIdAdvance(xmax);      /*并不是每个事务提交都会重新计算xmin和oldestxmin，只有每1000个事务或者每隔1s才会计算，此时xmin及oldestxmin一般偏小，但是不影响可见性判断。  */    currentTimeStamp = GetCurrentTimestamp();    if (forceCalc || ((++snapshotPendingCnt == MAX_PENDING_SNAPSHOT_CNT) ||                         (TimestampDifferenceExceeds(snapshotTimeStamp, currentTimeStamp, CALC_SNAPSHOT_TIMEOUT)))) {        snapshotPendingCnt = 0;        snapshotTimeStamp = currentTimeStamp;         /* 初始化xmin */        globalxmin = xmin = xmax;         int* pgprocnos = arrayP->pgprocnos;        int numProcs;         /*          循环遍历proc并计算快照相应值          */        numProcs = arrayP->numProcs;         /*主要流程，遍历proc_base_all_xacts，将其中pgxact->xid的最小值记为xmin，其中pgxact->xmin的最小值记为oldestxmin。  */        for (index = 0; index < numProcs; index++) {            int pgprocno = pgprocnos[index];            volatile PGXACT* pgxact = &g_instance.proc_base_all_xacts[pgprocno];            TransactionId xid;             if (pgxact->vacuumFlags & PROC_IN_LOGICAL_DECODING)                continue;             /* 对于autovacuum的xmin，跳过，避免长VACUUM阻塞脏元组回收 */            if (pgxact->vacuumFlags & PROC_IN_VACUUM)                continue;             /* 用最小的xmin来更新globalxmin */            xid = pgxact->xmin;             if (TransactionIdIsNormal(xid) && TransactionIdPrecedes(xid, globalxmin))                globalxmin = xid;             xid = pgxact->xid;             if (!TransactionIdIsNormal(xid))                xid = pgxact->next_xid;             if (!TransactionIdIsNormal(xid) || !TransactionIdPrecedes(xid, xmax))                continue;             if (TransactionIdPrecedes(xid, xmin))                xmin = xid;        }         if (TransactionIdPrecedes(xmin, globalxmin))            globalxmin = xmin;         t_thrd.xact_cxt.ShmemVariableCache->xmin = xmin;        t_thrd.xact_cxt.ShmemVariableCache->recentLocalXmin = globalxmin;    }     /* 此处给多版本快照信息赋值，xmin、oldestxmin因为不是及时计算故可能偏小，xmax、CSN号都是当前的准确值，注意计算快照的时候必须持有排他锁。  */    snapxid->xmin = t_thrd.xact_cxt.ShmemVariableCache->xmin;    snapxid->xmax = xmax;    snapxid->localxmin = t_thrd.xact_cxt.ShmemVariableCache->recentLocalXmin;    snapxid->snapshotcsn = t_thrd.xact_cxt.ShmemVariableCache->nextCommitSeqNo;    snapxid->takenDuringRecovery = RecoveryInProgress();    SetNextSnapXid(); /*设置当前多版本快照  */}（5） GetLocalSnapshotData：获取最新的多版本快照供事务使用。函数代码如下：Snapshot GetLocalSnapshotData(Snapshot snapshot){      /* 检查是否有多版本快照。在recover启动之前，是没有计算出多版本快照的，此时直接返回。 */    if (!g_snap_assigned || (g_snap_buffer == NULL)) {        ereport(DEBUG1, (errmsg("Falling back to origin GetSnapshotData: not assigned yet or during shutdown\n")));        return NULL;    }    pg_read_barrier(); /*为了防止ringBuffer初始化时的ARM乱序问题*/    snapxid_t* snapxid = GetCurrentSnapXid();  /* 将当前的多版本快照refcount++，避免被并发计算新快照的事务重用。  */     snapshot->user_data = snapxid;     …  /* 此处将多版本快照snapxid中的信息赋值给快照，注意此处是深拷贝，因为多版本快照仅有几个变量的关键信息，直接赋值即可，之后就可以将相应的多版本快照refcount释放。  */    u_sess->utils_cxt.RecentXmin = snapxid->xmin;    snapshot->xmin = snapxid->xmin;    snapshot->xmax = snapxid->xmax;    snapshot->snapshotcsn = snapxid->snapshotcsn;    …    ReleaseSnapshotData(snapshot);  /* 将多版本快照的refcount释放，以便可以被重用。  */    return snapshot;}