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

#define LW_FLAG_HAS_WAITERS ((uint32)1 << 30)
#define LW_FLAG_RELEASE_OK ((uint32)1 << 29)
#define LW_FLAG_LOCKED ((uint32)1 << 28)
 
#define LW_VAL_EXCLUSIVE ((uint32)1 << 24)
#define LW_VAL_SHARED 1 /* 用于标记LWLock的state状态，实现锁的获取和释放*/
 
typedef struct LWLock {
    uint16 tranche;         /* 轻量级锁的ID标识 */
    pg_atomic_uint32 state; /* 锁的状态位 */
    dlist_head waiters;     /* 等锁线程的链表 */
#ifdef LOCK_DEBUG
    pg_atomic_uint32 nwaiters; /* 等锁线程的个数 */
    struct PGPROC *owner;      /* 最后独占锁的持有者 */
#endif
#ifdef ENABLE_THREAD_CHECK
    pg_atomic_uint32 rwlock;
    pg_atomic_uint32 listlock;
#endif
} LWLock;

#define AccessShareLock 1  /* SELECT语句 */
#define RowShareLock 2     /* SELECT FOR UPDATE/FOR SHARE语句 */
#define RowExclusiveLock 3 /* INSERT, UPDATE, DELETE语句 */
#define ShareUpdateExclusiveLock  \
    4 /* VACUUM (non-FULL),ANALYZE, CREATE INDEX CONCURRENTLY语句 */
#define ShareLock 5 /* CREATE INDEX (WITHOUT CONCURRENTLY)语句 */
#define ShareRowExclusiveLock                \
    6 /* 类似于独占模式, 但是允许ROW SHARE模式并发 */
#define ExclusiveLock                  \
    7 /* 阻塞ROW SHARE，如SELECT...FOR UPDATE语句 */
#define AccessExclusiveLock              \
    8 /* ALTER TABLE, DROP TABLE, VACUUM FULL, LOCK TABLE语句 */

typedef struct LOCKTAG {
    uint32 locktag_field1;      /* 32比特位*/
    uint32 locktag_field2;      /* 32比特位*/
    uint32 locktag_field3;      /* 32比特位*/
    uint32 locktag_field4;      /* 32比特位*/
    uint16 locktag_field5;      /* 32比特位*/
    uint8 locktag_type;          /* 详情见枚举类LockTagType*/
    uint8 locktag_lockmethodid; /* 锁方法类型*/
} LOCKTAG;
 
typedef enum LockTagType {
    LOCKTAG_RELATION, /* 表关系*/
    /* LOCKTAG_RELATION的ID信息为所属库的OID+表OID；如果库的OID为0表示此表是共享表，其中OID为openGauss内核通用对象标识符 */
    LOCKTAG_RELATION_EXTEND, /* 扩展表的优先权*/
    /* LOCKTAG_RELATION_EXTEND的ID信息 */
    LOCKTAG_PARTITION,          /* 分区*/
    LOCKTAG_PARTITION_SEQUENCE, /* 分区序列*/
    LOCKTAG_PAGE,               /* 表中的页*/
    /* LOCKTAG_PAGE的ID信息为RELATION信息+BlockNumber（页面号）*/
    LOCKTAG_TUPLE, /* 物理元组*/
    /* LOCKTAG_TUPLE的ID信息为PAGE信息+OffsetNumber（页面上的偏移量） */
    LOCKTAG_TRANSACTION, /* 事务ID (为了等待相应的事务结束) */
    /* LOCKTAG_TRANSACTION的ID信息为事务ID号 */
    LOCKTAG_VIRTUALTRANSACTION, /* 虚拟事务ID */
    /* LOCKTAG_VIRTUALTRANSACTION的ID信息为它的虚拟事务ID号 */
    LOCKTAG_OBJECT, /* 非表关系的数据库对象 */
    /* LOCKTAG_OBJECT的ID信息为数据OID+类OID+对象OID+子ID */
    LOCKTAG_CSTORE_FREESPACE, /* 列存储空闲空间 */
    LOCKTAG_USERLOCK, /* 预留给用户锁的锁对象 */
    LOCKTAG_ADVISORY, /* 用户顾问锁 */
    LOCK_EVENT_NUM
} LockTagType;

typedef struct LOCK {
    /* 哈希键 */
    LOCKTAG tag; /* 锁对象的唯一标识 */
 
    /* 数据 */
    LOCKMASK grantMask;           /* 已经获取锁对象的位掩码 */
    LOCKMASK waitMask;            /* 等待锁对象的位掩码 */
    SHM_QUEUE procLocks;          /* 与锁关联的PROCLOCK对象链表 */
    PROC_QUEUE waitProcs;         /* 等待锁的PGPROC对象链表 */
    int requested[MAX_LOCKMODES]; /* 请求锁的计数 */
    int nRequested;               /* requested数组总数 */
    int granted[MAX_LOCKMODES];   /* 已获取锁的计数 */
    int nGranted;                 /* granted数组总数 */
} LOCK;

typedef struct PROCLOCK {
    /* 标识 */
    PROCLOCKTAG tag; /* proclock对象的唯一标识 */
 
    /* 数据 */
    LOCKMASK holdMask;    /* 已获取锁类型的位掩码 */
    LOCKMASK releaseMask; /* 预释放锁类型的位掩码 */
    SHM_QUEUE lockLink;   /* 指向锁对象链表的指针 */
    SHM_QUEUE procLink;   /* 指向PGPROC链表的指针 */
} PROCLOCK;

typedef struct {
    ThreadId thread_id;
    uint64 st_sessionid;
} lock_entry_id; 

typedef struct {
    /* 线程ID */
    lock_entry_id entry_id;
 
    /* 轻量级锁检测引用计数 */
    int lw_count;
} lwm_light_detect; 

typedef struct {
    lock_entry_id be_tid;         /* 线程ID */
    int be_idx;                   /* 后台线程的位置 */
    LWLockAddr want_lwlock;         /* 预获取锁的信息 */
    int lwlocks_num;              /* 线程持有的轻量级锁个数 */
    lwlock_id_mode* held_lwlocks; /* 线程持有的轻量级锁数组 */
} lwm_lwlocks;

static void CheckDeadLockRunningTooLong(int depth)
{ /* 每4层检测一下 */
    if (depth > 0 && ((depth % 4) == 0)) {
        TimestampTz now = GetCurrentTimestamp();
        long secs = 0;
        int usecs = 0;
 
        if (now > t_thrd.storage_cxt.deadlock_checker_start_time) {
            TimestampDifference(t_thrd.storage_cxt.deadlock_checker_start_time, now, &secs, &usecs);
            if (secs > 600) { /* 如果从死锁检测开始超过十分钟，则报错处理。*/
#ifdef USE_ASSERT_CHECKING
                DumpAllLocks();/* 在debug版本时，导出所有的锁信息，便于定位问题。*/
#endif
 
                ereport(defence_errlevel(), (errcode(ERRCODE_INTERNAL_ERROR),
                                             errmsg("Deadlock checker runs too long and is greater than 10 minutes.")));
            }
        }
    }
}

typedef struct EDGE {
    PGPROC *waiter;  /* 等待的线程 */
    PGPROC *blocker; /* 阻塞的线程 */
    int pred;        /* 拓扑排序的工作区 */
    int link;        /* 拓扑排序的工作区 */
} EDGE;

typedef struct WAIT_ORDER {
    LOCK *lock;     /* the lock whose wait queue is described */
    PGPROC **procs; /* array of PGPROC *'s in new wait order */
    int nProcs;
} WAIT_ORDER;

typedef struct DEADLOCK_INFO {
    LOCKTAG locktag;   /* 等待锁对象的唯一标识 */
    LOCKMODE lockmode; /* 等待锁对象的锁类型 */
    ThreadId pid;      /* 阻塞线程的线程ID */
} DEADLOCK_INFO;

static inline int32 gs_atomic_add_32(volatile int32* ptr, int32 inc)
{
    return __sync_fetch_and_add(ptr, inc) + inc;
}

static inline int64 gs_atomic_add_64(int64* ptr, int64 inc)
{
    return __sync_fetch_and_add(ptr, inc) + inc;
}

static inline bool gs_compare_and_swap_32(int32* dest, int32 oldval, int32 newval)
{
    if (oldval == newval)
        return true;
 
    volatile bool res = __sync_bool_compare_and_swap(dest, oldval, newval);
 
    return res;
}

static inline bool gs_compare_and_swap_64(int64* dest, int64 oldval, int64 newval)
{
    if (oldval == newval)
        return true;
 
    return __sync_bool_compare_and_swap(dest, oldval, newval);
}

static inline uint128_u arm_compare_and_swap_u128(volatile uint128_u* ptr, uint128_u oldval, uint128_u newval)
{
#ifdef __ARM_LSE
    return __lse_compare_and_swap_u128(ptr, oldval, newval);
#else
    return __excl_compare_and_swap_u128(ptr, oldval, newval);
#endif
}
#endif

static inline uint128_u atomic_compare_and_swap_u128(
    volatile uint128_u* ptr,
    uint128_u oldval = uint128_u{0},
    uint128_u newval = uint128_u{0})
{
#ifdef __aarch64__
    return arm_compare_and_swap_u128(ptr, oldval, newval);
#else
    uint128_u ret;
    ret.u128 = __sync_val_compare_and_swap(&ptr->u128, oldval.u128, newval.u128);
    return ret;
#endif
}

static XLogRecPtr XLogInsertRecordGroup(XLogRecData* rdata, XLogRecPtr fpw_lsn)
{
    …/* 初始化变量以及简单校验  */
    START_CRIT_SECTION();  /* 开启临界区  */
 
    proc->xlogGroupMember = true;
   …
    proc->xlogGroupDoPageWrites = &t_thrd.xlog_cxt.doPageWrites;
 
    nextidx = pg_atomic_read_u32(&t_thrd.shemem_ptr_cxt.LocalGroupWALInsertLocks[groupnum].l.xlogGroupFirst); 
 
    while (true) {
        pg_atomic_write_u32(&proc->xlogGroupNext, nextidx); /* 将上一个成员记录到proc结构体中  */
        /* 防止ARM乱序：保证所有前面的写操作都可见 */
        pg_write_barrier();
 
        if (pg_atomic_compare_exchange_u32(&t_thrd.shemem_ptr_cxt.LocalGroupWALInsertLocks[groupnum].l.xlogGroupFirst,
                &nextidx,
                (uint32)proc->pgprocno)) {
            break;
        } /* 这一步原子操作获取上一个成员的proc no，如果是invalid，说明是leader。*/
    }
     /* 非leader成员不去获取WAL Insert锁，仅仅进行等待，直到被leader唤醒  */
    if (nextidx != INVALID_PGPROCNO) {
        int extraWaits = 0;
 
        for (;;) {
            /* 充当读屏障 */
            PGSemaphoreLock(&proc->sem, false);
            /* 充当读屏障 */
            pg_memory_barrier();
            if (!proc->xlogGroupMember) {
                break;
            }
            extraWaits++;
        }
 
        while (extraWaits-- > 0) {
            PGSemaphoreUnlock(&proc->sem);
        }
        END_CRIT_SECTION();
        return proc->xlogGroupReturntRecPtr;
    }
    /* leader成员持有锁  */
    WALInsertLockAcquire();
    /* 计算每个成员线程的xlog record size  */
    …
    /* leader线程将所有成员线程的xlog record插入到缓冲区  */
    while (nextidx != INVALID_PGPROCNO) {
        localProc = g_instance.proc_base_all_procs[nextidx];
 
        if (unlikely(localProc->xlogGroupIsFPW)) {
            nextidx = pg_atomic_read_u32(&localProc->xlogGroupNext);
            localProc->xlogGroupIsFPW = false;
            continue;
        }
        XLogInsertRecordNolock(localProc->xlogGrouprdata,
            localProc,
            XLogBytePosToRecPtr(StartBytePos),
            XLogBytePosToEndRecPtr(
                StartBytePos + MAXALIGN(((XLogRecord*)(localProc->xlogGrouprdata->data))->xl_tot_len)),
            XLogBytePosToRecPtr(PrevBytePos));
        PrevBytePos = StartBytePos;
        StartBytePos += MAXALIGN(((XLogRecord*)(localProc->xlogGrouprdata->data))->xl_tot_len);
        nextidx = pg_atomic_read_u32(&localProc->xlogGroupNext);
    }
 
    WALInsertLockRelease();  /*  完成工作放锁并唤醒所有成员线程  */
    while (wakeidx != INVALID_PGPROCNO) {
        PGPROC* proc = g_instance.proc_base_all_procs[wakeidx];
 
        wakeidx = pg_atomic_read_u32(&proc->xlogGroupNext);
        pg_atomic_write_u32(&proc->xlogGroupNext, INVALID_PGPROCNO);
        proc->xlogGroupMember = false;
        pg_memory_barrier();
 
        if (proc != t_thrd.proc) {
            PGSemaphoreUnlock(&proc->sem);
        }
    }
 
    END_CRIT_SECTION();
    return proc->xlogGroupReturntRecPtr;
}

#ifdef __aarch64__
#define LWLOCK_PADDED_SIZE PG_CACHE_LINE_SIZE(128)
#else
#define LWLOCK_PADDED_SIZE (sizeof(LWLock) <= 32 ? 32 : 64)
#endif
typedef union LWLockPadded
{
LWLocklock;
charpad[LWLOCK_PADDED_SIZE];
} LWLockPadded;

typedef union {
    uint128   u128;
    uint64    u64[2];
    uint32    u32[4];
} uint128_u; * 为了代码可读及操作，将u128设计成union的联合结构体，内存位置进行64位数值的赋值。*/
static void ReserveXLogInsertLocation(uint32 size, XLogRecPtr* StartPos, XLogRecPtr* EndPos, XLogRecPtr* PrevPtr)
{
    volatile XLogCtlInsert* Insert = &t_thrd.shemem_ptr_cxt.XLogCtl->Insert;
    uint64 startbytepos;
    uint64 endbytepos;
    uint64 prevbytepos;
 
    size = MAXALIGN(size);
 
#if defined(__x86_64__) || defined(__aarch64__)
    uint128_u compare;
    uint128_u exchange;
    uint128_u current;
 
    compare = atomic_compare_and_swap_u128((uint128_u*)&Insert->CurrBytePos);
 
loop1:
    startbytepos = compare.u64[0];
    endbytepos = startbytepos + size;
 
    exchange.u64[0] = endbytepos;  * 此处为了代码可读，将uint128设置成一个union的联合结构体。将起始和结束位置写入到exchange中。*/
    exchange.u64[1] = startbytepos;
 
    current = atomic_compare_and_swap_u128((uint128_u*)&Insert->CurrBytePos, compare, exchange);
    if (!UINT128_IS_EQUAL(compare, current)) { * 如果被其他线程并发更新，重新循环*/
        UINT128_COPY(compare, current);
        goto loop1;
    }
    prevbytepos = compare.u64[1];
 
#else
    SpinLockAcquire(&Insert->insertpos_lck); /* 其余平台使用自旋锁原子锁来保护变量更新 */
    startbytepos = Insert->CurrBytePos;
    prevbytepos = Insert->PrevBytePos;
    endbytepos = startbytepos + size;
    Insert->CurrBytePos = endbytepos;
    Insert->PrevBytePos = startbytepos;
 
    SpinLockRelease(&Insert->insertpos_lck);
#endif * __x86_64__|| __aarch64__ */
    *StartPos = XLogBytePosToRecPtr(startbytepos);
    *EndPos = XLogBytePosToEndRecPtr(endbytepos);
    *PrevPtr = XLogBytePosToRecPtr(prevbytepos);
}

/* CLOG分区*/
#define NUM_CLOG_PARTITIONS 256  /*分区打散的个数*/
/* CLOG轻量级分区锁*/
#define CBufHashPartition(hashcode) \
    ((hashcode) % NUM_CLOG_PARTITIONS)
#define CBufMappingPartitionLock(hashcode) \
    (&t_thrd.shemem_ptr_cxt.mainLWLockArray[FirstCBufMappingLock + CBufHashPartition(hashcode)].lock)
#define CBufMappingPartitionLockByIndex(i) \
    (&t_thrd.shemem_ptr_cxt.mainLWLockArray[FirstCBufMappingLock + i].lock)

#ifdef __USE_NUMA
    if (nNumaNodes > 1) {
        ereport(INFO, (errmsg("InitProcGlobal nNumaNodes: %d, inheritThreadPool: %d, groupNum: %d",
            nNumaNodes, g_instance.numa_cxt.inheritThreadPool,
            (g_threadPoolControler ? g_threadPoolControler->GetGroupNum() : 0))));
 
        int groupProcCount = (TotalProcs + nNumaNodes - 1)  nNumaNodes;
        size_t allocSize = groupProcCount * sizeof(PGPROC);
        for (int nodeNo = 0; nodeNo < nNumaNodes; nodeNo++) {
            initProcs[nodeNo] = (PGPROC *)numa_alloc_onnode(allocSize, nodeNo);
            if (!initProcs[nodeNo]) {
                ereport(FATAL, (errcode(ERRCODE_OUT_OF_MEMORY),
                                errmsg("InitProcGlobal NUMA memory allocation in node %d failed.", nodeNo)));
            }
            add_numa_alloc_info(initProcs[nodeNo], allocSize);
            int ret = memset_s(initProcs[nodeNo], groupProcCount * sizeof(PGPROC), 0, groupProcCount * sizeof(PGPROC));
            securec_check_c(ret, "\0", "\0");
        }
    } else {
#endif

   WALInsertLockPadded** insertLockGroupPtr =
        (WALInsertLockPadded**)CACHELINEALIGN(palloc0(nNumaNodes * sizeof(WALInsertLockPadded*) + PG_CACHE_LINE_SIZE));
#ifdef __USE_NUMA
    if (nNumaNodes > 1) {
        size_t allocSize = sizeof(WALInsertLockPadded) * g_instance.xlog_cxt.num_locks_in_group + PG_CACHE_LINE_SIZE;
        for (int i = 0; i < nNumaNodes; i++) {
            char* pInsertLock = (char*)numa_alloc_onnode(allocSize, i);
            if (pInsertLock == NULL) {
                ereport(PANIC, (errmsg("XLOGShmemInit could not alloc memory on node %d", i)));
            }
            add_numa_alloc_info(pInsertLock, allocSize);
            insertLockGroupPtr[i] = (WALInsertLockPadded*)(CACHELINEALIGN(pInsertLock));
        }
    } else {
#endif
        char* pInsertLock = (char*)CACHELINEALIGN(palloc(
            sizeof(WALInsertLockPadded) * g_instance.attr.attr_storage.num_xloginsert_locks + PG_CACHE_LINE_SIZE));
        insertLockGroupPtr[0] = (WALInsertLockPadded*)(CACHELINEALIGN(pInsertLock));
#ifdef __USE_NUMA
    }
#endif

typedef struct
{
LWLock lock;
#ifdef __aarch64__
pg_atomic_uint32xlogGroupFirst;
#endif
XLogRecPtrinsertingAt;
} WALInsertLock;