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;
}