PostgreSQL中插入数据时与WAL相关的处理逻辑是什么？

1.数据结构

静态变量
进程中全局共享

static int num_rdatas; /* 当前使用的条目 *///分配的空间大小 static int max_rdatas; /* 分配的大小 */// 是否由当前后端调用 XLogBeginInsert 函数 static bool begininsert_used = false;static。它针对所有 inserts.p 进行更新；指向最后一条记录的 * end+1，并在我们结束顶级事务或开始新事务时重置；因此它可以用来判断当前事务是否创建了任何 XLOG 记录。 * ProcLastRecPtr 指向当前ba插入的最后一条XLOG记录的开头肯德。 * 所有插入内容均已更新。 * XactLastRecEnd 指向最后一条记录的结束位置+1， * 当顶层事务结束或新事务开始时重置； * 因此，可以用来判断当前事务是否创建了任何XLOG记录。 * * 在并行模式下，这可能不是完全最新的。提交时， * 事务可以假设这涵盖了 * 用户后端或事务期间任何时刻出现的任何并行工作线程写入的所有 xlog 记录。但是当中止时，或者仍然处于并行模式时，其他并行后端可能会在比此处存储的值晚的 LSN 处写入 WAL 记录。必要时，并行领导者会在 WaitForParallelWorkersToFinish 中推进自己的副本。 * 在并行模式下，这可能不是完全最新的。 * 在提交时，可以假设事务已经覆盖了用户后台进程或并行工作进程的所有 xlog 记录交易期间发生的。 * 然而，当中止时，或者仍处于并行模式时，其他并行后台进程可能已在后面的 LSN 中写入 WAL 记录 * 而不是此处存储的值。 * 当需要时，并行处理进程的领导者会在WaitForParallelWorkersToFinish中推进自己的副本。 */XLogRecPtr ProcLastRecPtr = InvalidXLogRecPtr;XLogRecPtr XactLastRecEnd = InvalidXLogRecPtr;XLogRecPtr XactLastCommitEnd = InvalidXLogRecPtr;/* 对于 WALinsertLockAcquire/Release 函数 *///对于 WALinsertLockAcquire/Release 函数 static int MyLockNo = 0;static bool HoldingAllLocks = false;

typedef char* Pointer;//指针 typedef Pointer Page;//Page#define XLOG_HEAP_INSERT 0x00/* * 指向XLOG中某个位置的指针。这些指针是 64 位宽，* 因为我们不希望它们溢出。 * 指向 XLOG 中的位置。 * 这些指针的大小为 64 位，以确保指针不会溢出。 */类型def uint64 XLogRecPtr;/* * 用于访问页眉的附加宏。 （注意参数的多重求值！） */#define PageGetLSN(page) \PageXLogRecPtrGet(((PageHeader) (page))->pd_lsn)#define PageSetLSN(page, lsn) \ PageXLogRecPtrSet(((PageHeader) (page) ))->pd_lsn, lsn)/* 存储压缩版本的备份块映像所需的缓冲区大小 *///存储压缩后的块映像所需的缓存空间大小 #define PGLZ_MAX_BLCKSZ PGLZ_MAX_OUTPUT(BLCKSZ)//-- ---------------------------------- ---------------------------------------- - 锁相关/* * 使用信号量的假自旋锁实现 --- 缓慢且容易 * 违反信号量数量的内核限制，因此不要使用此 * 除非必须！子例程出现在 spin.c 中。与 * 冲突，所以除非必要，否则不要使用它！该子例程出现在 spin.c 中。 */typedef int slock_t;typedef uint32 pg_crc32c;#define SpinLockInit(lock) S_INIT_LOCK(lock)#define SpinLockAcquire(lock) S_LOCK(lock)#define SpinLockRelease(lock) S_UNLOCK(lock)#define SpinLockFree(lock) S_LOCK_FREE(lock)#define XLogSegmentOff set(xlogptr , wal_segsz_bytes) \ ((xlogptr) & ((wal_segsz_bytes) - 1))#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_SHARED1#define LW_LOCK_MASK ((uint32) ( (1 << 25)-1))/* 必须大于 MAX_BACKENDS - 即 2^23-1，所以我们没问题。 */#define LW_SHARED_MASK ((uint32) ((1 << 24)-1))

LWLock
lwlock.c之外的代码不应直接操作内容这个结构，但我们必须声明该结构才能将 LWLock 合并到其他数据结构中。

/* * lwlock.c 之外的代码不应该直接操作这个 * 结构的内容，但是我们必须在这里声明它以允许 LWLocks * 合并到其他数据结构中。 *&nlwlock.c 之外的代码不应直接操作 o 的内容f 这个结构，* 但我们必须声明该结构才能将 LWLock 合并到其他数据结构中。 * / typedef struct LWLock { uint16 tranche; /* tranche ID */ //独占/非独占储物柜状态 pg_atomic_uint32 state; /* 独占/非独占锁柜的状态 */ //等待PGPROCs链表 proclist_head waiters; /* 等待的 PGPROC 列表 */#ifdef LOCK_DEBUG//用于 DEBUG // 等待者数量 pg_atomic_uint32 nwaiters; /* 等待者数量 */ //锁的最后独占拥有者 struct PGPROC *owner; /* 锁的最后独占所有者 */#endif} LWLock;

2.源码解读

heap_insert
主要实现逻辑就是将元组插入到堆中，有一部分是处理WAL（XLog）的。
请参见 PostgreSQL 源代码解读（104）- WAL#1（Insert & WAL-heap_insert 函数#1）

XLogInsert/XLogInsertRecord
插入指定的 XLOG 记录RMID 和信息字节，主体 of 是先前通过 XLogRegister* 调用注册的数据和缓冲区引用。
参见PostgreSQL源码解读(106)-WAL#3(Insert&WAL-heap_insert函数#3)

WALInsertLockXXX
包括WALInsertLockAcquireExclusive、WALInsertLockAcquire和WALInsertLockRelease 等待

//---------------------------------------------------- --------------------- WALInsertLockAcquireExclusive/* * 获取所有WAL插入锁，防止其他后端向WAL插入*。 * 请求所有WAL插入锁，以防止其他后端插入 * 到WAL。后台进程向WAL中插入数据 */static voidWALInsertLockAcquireExclusive(void){ int i; /* * 当持有所有锁时，除最后一个锁之外的所有锁的 insertingAt * 指示符都设置为 0xFFFFFFFFFFFFFFFF，该值高于任何实际 * XLogRecPtr 值，以确保没有人阻塞等待这些锁。 * 持有 对于所有锁，除了la的insertingAt指示符st 锁，** 其余的设置为 0xFFFFFFFFFFFFFFFF， * 大于所有实际的 XLogRecPtr 以确保这些锁不被阻塞。 。 */ for (i = 0; i < NUM_XLOGINSERT_LOCKS - 1; i++)//NUM_XLOGINSERT_LOCKS { LWLockAcquire(&WALInsertLocks[i].l.lock, LW_EXCLUSIVE); LWLockUpdateVar(&WALInsertLocks[i].l.lock, &WALInsertLocks[i].l.insertingAt, PG_UINT64_MAX); 。为0 lwlockacquire(&walinsrtLocks[i].l.lock, lw_exClusive); // 设置markingallLocks = TRUE;}/**Lwlockacquire-在指定模式下获取轻量级Lo CK*lwlockacquire-在指定模式下获取轻量级锁* *如果锁不可用，则休眠直到它可用。如果锁立即可用则返回 true，如果我们必须睡觉则返回 false。 * 如果没有锁可用，睡眠直到可用。 * 如果锁立即可用，则返回 T，需要 Sleep 返回 F * * 副作用：取消/死亡中断将被推迟，直到锁释放。 ;//PGPROC数据结构 bool result = true; int extraWaits = 0;#ifdef LWLOCK_STATS lwlock_stats *lwstats; lwstats = get_lwlock_stats_entry(lock);//获取锁统计条目#endif //模式验证 AssertArg(mode == LW_SHARED || mode == LW_EXCLUSIVE); PRINT_LWDEBUG("LWLockAcquire", lock, mode);#ifdef LWLOCK_STATS /* 计算锁获取尝试次数 */ if (mode == LW_EXCLUSIVE) lwstats->ex_acquire_count++; else lwstats->sh_acquire_count++;#endif /* LWLOCK_STATS */ /* * 如果我们还没有 PGPROC，我们就不能等待。这应该仅在引导或共享内存初始化期间发生*。在此处添加断言*以捕获不安全的编码实践。确保发现不安全的编码行为。 */ Assert(!(proc == NULL && IsUnderPostmaster)); /* 确保我们有空间记住锁 */ // 确保我们有空间有足够的空间来存储锁 if (num_held_lwlocks > = MAX_SIMUL_LWLOCKS) elog(ERROR, "too much LWLocks take"); /* * 锁定取消/死亡中断，直到我们退出受 LWLock 保护的代码段。这确保了中断不会干扰共享内存中数据结构的操作。 * 仅当退出受LWLock锁保护的实现逻辑时才允许取消或中断。 * 这保证了中断不会与共享内存中的数据结构管理逻辑相关联。 */ HOLD_INTERRUPTS(); /* * 每次我们收到 LWLockRelease 信号后循环尝试获取锁。 * 每次我们收到 LWLockRelease 信号后，循环尝试获取锁。当 LWLockRelease 信号产生时获取锁。 * * * 注意：让 LWLockRelease 实际上授予我们锁似乎更好，而不是重试并可能不得不返回睡眠状态。但*在实践中这并不好，因为它意味着当两个或多个进程继续获取同一个锁时，每次获取锁时都会进行进程交换。由于 LWLock 通常用于保护不是很长的计算部分，因此进程需要能够在单个 CPU 时间片内多次获取和释放同一锁，即使存在争用也是如此。能够做到这一点的效率超过了有时浪费进程分派的低效率，因为当释放的等待者最终运行时锁尚未释放。请参阅 2001 年 12 月 29 日的 pgsql-hackers 档案。 * 注：似乎由 LWLockRelease 的实际所有者授予我们锁会比不断地重入和休眠更好， * 但在工程实践中， * 这种做法并不好，因为这意味着当两个或多个进程竞争时对于同一个锁，每个锁都会有一个进程交换。 * 由于LWLock一般用于保护不需要太长时间的计算逻辑， *甚至当发生争用时，进程需要能够在一个CPU时间片内多次获取和释放同一个锁。 * 这样做的回报有时会导致进程调度效率低下， * 因为当一个被释放的进程最终准备好运行时，并没有获取锁。 */ for (;;) { bool 必须等待; /* * 第一次尝试获取锁，我们还没有/不再在等待队列中。 * 第一次尝试获取锁，我们已经不在等待队列中了。 */        必须等待 = LWLockAttemptLock(lock, mode); if (!mustwait) { LOG_LWDEBUG("LWLockAcquire", lock, "立即获取锁");休息; /* 成功！获得锁 */ } /*       * 好的，此时我们无法在第一次尝试时获取锁。我们 * 不能简单地将自己排队到列表的末尾并等待 * 被唤醒，因为现在锁可能早已被释放了。 * 而是将我们添加到队列中并尝试再次抢锁。如果我们成功了，我们需要恢复排队并感到高兴，否则我们*重新检查锁。如果我们仍然无法获取它，我们就知道*其他锁在释放时会看到我们的队列条目，因为它们*在我们检查锁之前就已经存在。*此时，我们无法第一次获取锁。 * 我们不能简单地在链表末尾排队等待唤醒，因为锁可能已经释放很长时间了。 * 相反，我们需要重新加入队列。尝试再次获取锁。 * 如果成功，我们需要翻转队列，否则我们需要重新检查锁。 * 如果我们仍然无法获取锁，我们就知道其他lockers在释放时可以看到我们的队列条目， * 因为当我们检查锁时它们已经存在。 */ /* 添加到队列 */ //添加到队列 LWLockQueueSelf(lock, mode); /* 我们现在保证在必要时被唤醒 */   //确保在需要时可以被唤醒 Mustwait = LWLockAttemptLock(lock, mode);取消队列 if (!mustwait) { LOG_LWDEBUG("LWLockAcquire", lock, "已获取, un正在执行队列"); LWLockDequeueSelf(lock);//出队中断;                                                                                                                                                         bsp;* 等待唤醒。 * 等待唤醒 * * 由于我们与常规锁管理器和 ProcWaitForSignal 共享进程等待信号量，因此我们可能需要获取L WLock *当其中一个处于待处理状态时，我们可能会因为 LWLockRelease 发出信号以外的原因而被唤醒。如果是这样，*循环并再次等待。一旦我们获得了 LWLock，*重新将信号量增加接收到的附加信号的数量，*以便锁管理器或信号管理器在下次等待时将看到接收到的*信号。         *由于我们使用常规锁管理和 ProcWaitForSignal 信号共享进程等待信号量，*我们其中一个挂掉时可能需要获取LWLock， * 原因可能是由于被唤醒以外的其他原因d 通过 LWLockRelease 信号。 * 如果是这种情况，则继续循环等待。 * 一旦获得LWLock，根据收到的附加信号数量再次增加信号量，         *   方便锁管理器或者信号管理器在下一个等待时看到已接收到的信号就可以。 */        LOG_LWDEBUG("LWLockAcquire", lock, "waiting");#ifdef LWLOCK_STATS        lwstats->block_count++;//统计# endif LWLockReportWaitStart(lock);//报告等待 TRACE_POSTGRESQL_LWLOCK_WAIT_START(T_NAME(lock), mode); for (;;) { PGSemaphoreLock(proc->sem); if (!proc->lwWaiting)//如果不是LWLock等待，则跳出Loops break; extraWaits++;//ExtraWaits++;//额外等待                                                                                                                                                                                                                                                    (&lock->state, LW_FLAG_RELEASE_OK);#ifdef LOCK_DEBUG                                                       * 不再等待 */ //无需等待ait uint32 nwaiters PG_USED_FOR_ASSERTS_ONLY = pg_atomic_fetch_sub_u32(&lock->nwaiters, 1);断言(nwaiters < MAX_BACKENDS); }#endif TRACE_POSTGRESQL_LWLOCK_WAIT_DONE(T_NAME(锁), 模式); quire", lock, "awakened"); //再次循环以再次请求锁 /* 现在循环返回并尝试再次获取锁。 */ result = false; } TRACE_POSTGRESQL_LWLOCK_ACQUIRE(T_NAME(lock), mode); //获取成功！ /* Add lock to list of lock by this backend */ //向后台进程持有的锁列表中添加锁held_lwlocks[num_held_lwlocks].lock = lock;held_lwlocks[num_held_lwlocks++].mode = mode;/ * * 修复进程等待信号量的任何吸收唤醒的计数。 * 修复进程如何等待信号量计数的其他吸收唤醒。 */ while (extraWaits--> 0) PGSemaphoreUnlock(proc->sem); return result ;} /* * 尝试以原子方式获取传入的 lwlock 的内部函数 * in mode. * 尝试使用指定的模式以原子方式获取 LWLock。锁的内部功能。 * * 该函数不会阻塞等待锁释放 - 这是调用者的工作。 * 该函数不会阻塞等待锁释放的进程——这是调用者的工作。 * * 如果锁未释放并且我们需要等待，则返回 true。 * 如果锁还没有释放，还需要等待，则返回T */static boolLWLockAttemptLock(LWLock *lock, LWLockMode mode){ uint32 old_state; AssertArg(模式 == LW_EXCLUSIVE || 模式 == LW_SHARED); /* * 在循环外读取一次，稍后的迭代将通过比较和交换获得更新的值。 * 循环外读取一次，然后通过比较和交换就可以得到更新的值 */ old_state = pg_atomic_read_u32(&lock->state); /* 循环直到我们确定是否可以获取锁 */ //循环指针我们已经确定是否可以获取锁位置 while (true) { uint32desired_state;布尔lock_free； ​​desired_state = old_state;如果（米ode == LW_EXCLUSIVE) // 独占                                                              sp; desired_state+= lw_val_exClusive;} else {//非独占lock_free = (old_state & lw_exclusive) == 0; if (lock_free) design_State+= lw_va L_shared;}/**尝试交换到我们正在执行的状态。如果我们没有看到 * lock 是空闲的，那只是旧值。如果我们认为它是免费的，*我们将尝试将其标记为已获取。我们总是在值中交换 * 的原因是，这同时也是内存屏障。我们可以尝试*变得更聪明，并且仅在我们看到锁是空闲的情况下才交换值，*但到目前为止基准测试尚未表明它是有益的。 * 尝试以我们想要的状态进行交换。 * 如果你没有看到锁被释放，那么这次是旧值。 * 如果锁被释放，尝试将锁标记为已获取。 * 我们通常交换值的原因是它会使用双倍的内存屏障。 * 我们尝试改变甚至更好：只释放我们看到的交换锁d，但压力测试显示性能提升甚微。 * 如果自从我们上次查看以来该值发生了变化，请重试。 * 如果自从我们上次查看以来该值发生了变化，请重试。尝试 */ if (pg_atomic_compare_exchange_u32(&lock->state,                                                                                                                            ！#ifdef LOCK_DEBUG                                                                                                                               return false; else has return true; /* 别人有锁 */ ---- ------------- ------------------------------------------------- ---- WALInsertLockAcquire/* * 获取一个WAL插入锁，用于插入到 WAL。我们获取了哪个 WAL 插入锁，所以尝试 * 我们上次使用的那个。如果系统不是特别繁忙，那么 * 一个很好的选择是它仍然可用，并且最好有一些* 与特定锁的关联性，这样您就不会在进程之间不必要地反弹 * 缓存线没有争议。 * 我们请求哪一个WAL插入锁并不重要，所以我们得到最后一个使用的锁。 * 如果系统不忙，那么运气好的话它仍然可用， * 最好与特定的锁保持一定的亲和力，这样 * * 可以避免在没有争用的情况下在进程之间不必要地切换缓存行。 * * 如果这是第一次进入此后端，则（半）随机地开锁。如果您有很多非常短的连接，这允许均匀地使用锁。第一次获取时，随机获取锁。 * 如果有很多很短的连接，这样可以均匀的使用锁。 */ 静态 int lockToTry = -1; if (lockToTry == -1) lockToTry = MyProc->pgprocno % NUM_XLOGINSERT_LOCKS; MyLockNo = lockToTry; /* * insertingAt 值最初设置为 0，因为我们还不知道 * 插入位置。 * insertAt 值初始化为 0，因为我们还不知道将其插入到哪里。 .lock, LW_EXCLUSIVE);如果(!immed) { /* * 如果我们无法立即获得锁，请下次尝试另一个锁。在插入锁多于并发 * 插入的系统上，这会导致所有插入器最终迁移到没有其他人使用的 * 锁。在插入器*多于锁的系统上，它仍然有助于将插入器均匀地分布在锁上。锁一下，下次再试试别的锁。 * 在插入锁多于并发插入器的系统中，*                                                                                                                                                                                                                                                                                       ​锁。 */ lockToTry = (lockToTry + 1) % NUM_XLOGINSERT_LOCKS; }}//---------------------------------------------- ---- --------------- WALInsertLockRelease/* * 释放我们的插入锁（或者多个锁，如果我们持有它们的话）。 * 释放插入锁 * * 注意：重新设置t 所有变量都为 0，因此它们会导致 LWLockWaitForVar 在下次获取锁时阻塞。 * 注意：将所有变量重置为 0，以便它们在下次获取锁时导致 LWLockWaitForVar 阻塞。 */static voidWALInsertLockRelease(void){ if (holdingAllLocks) //如果所有锁都被持有 {                                                                                                                                  ; LWLockReleaseClearVar(&WALInsertLocks[i].l.lock,                                                                                                                holdingAllLocks = false; } else {LWLockReleaseClearVar(&WALInsertLocks[MyLockNo].l.lock, ti​​ngAt,                                                                                                                                        /*                                                                                                                                               与 相对于 ar(LWLock *lock, uint64 *valptr, uint64 val){ LWLockWaitListLock( 锁); /* * 在释放锁之前设置变量的值，以防止竞争 * 一种竞争条件，其中新的锁获取锁，但尚未设置变量值。 * 在释放锁之前设置变量的值。当新的locker在没有设置变量值的情况下获取锁时，这可以防止争用。 */ *valptr = val; LWLockWaitListUnlock(锁定); LWLockRelease(lock) ;}/** 针对并发活动锁定 LWLock 的等待列表。* 针对并发活动锁定 LWLock 等待列表** 注意：即使等待列表被锁定，非冲突的锁定操作*仍可能并发发生。 * 注意：虽然等待列表被锁定，但非冲突的锁定操作*仍然可能并发发生。等待链表上锁，非冲突的锁操作仍有可能并发发生** 持有互斥量的时间应该短！* 持有互斥量的时间应该尽可能短ble*/static voidLWLockWaitListLock(LWLock *lock){ uint32 old_state; #ifdef LWLOCK_STATS lwlock_stats *lwstats; uint32 延迟 = 0； lwstats = get_lwlock_stats_entry(lock);#endif while (true) { /* 总是尝试一次直接获取锁 */ //第一次尝试直接获取锁 old_state = pg_atomic_fetch_or_u32(&lock- >state, LW_FLAG_LOCKED); if (!(old_state & LW_FLAG_LOCKED)) 中断;旋转直到锁被释放 lock ;ays;#endif                                                                                                                                                                                                                                                          又来了。 * 再试一次，尝试重新请求时可能会获取锁。 */ }#ifdef LWLOCK_STATS lwstats->spin_delay_count += delays;//Delay count#endif} /* * 解锁LWLock的等待列表。 * 解锁 LWLock 的等待列表 * * 请注意，操作标志并释放 * 中的锁会更有效单个原子操作。 * 请注意，在单个原子操作中操作标志和释放锁可能会更有效。 */static voidLWLockWaitListUnlock(LWLock *lock){ uint32 old_state PG_USED_FOR_ASSERTS_ONLY; old_state = pg_atomic_fetch_and_u32(&lock->state, ~LW_FLAG_LOCKED); Assert(old_state & LW_FLAG_LOC KED);}/** LWLockRelease - 释放先前获取的锁* LWLockRelease - 释放先前获取的锁 获取锁 */voidLWLockRelease(LWLock *lock){ LWLockMode mode; uint32 旧状态；布尔检查服务员； int i;/* * 从持有的锁列表中删除锁。通常，但并非总是，它将是最新获取的锁；所以向后搜索数组。 （总是），最后请求的锁被清除，因此从后到前搜索数组。 */ for (i = num_held_lwlocks; --i >= 0;) if (lock ==held_lwlocks[i].lock) 中断 ; if (i < 0) elog(ERROR, "锁 %s 未被持有", T_NAME(lock)); mode =held_lwlocks[i].mode;//Mode   num_held_lwlocks--;//减一 for (; i < num_held_lwlocks； i++) 持有_lwlocks[i] = 持有_lwlocks[i + 1]; PRINT_LWDEBUG("LWLockRelease", 锁, 模式); /* * 释放我对锁的持有，之后它可以立即被其他人获取，即使我们仍然需要唤醒其他服务员。 * 释放“I”持有的锁， */ if (mode == LW_EXCLUSIVE) oldstate = pg_atomic_sub_fetch_u32(&lock->state, LW_VAL_EXCLUSIVE); else oldstate = pg_atomic_sub_fetch_u32(&lock->state, LW_VAL_SHARED); /* 没有其他人可以拥有这种锁 */ //我还能做谁！断言（！（旧状态＆LW_VAL_EXCLUSIVE））; /* * 我们仍在等待后端被调度，不要再次唤醒它们。 FLAG_HAS_WAITERS | FLAG_HAS_WAITERS | LW_FLAG_RELEASE_OK ) && (oldstate & LW_LOCK_MASK) == 0)   check_waiters = true;否则 check_waiters = false; /* * 由于唤醒服务员需要获取自旋锁，因此只有在必要时才这样做。 * 因为唤醒服务员需要获取自旋锁，只有在必要的时候才可以。 */ if (check_waiters) { /* XXX: 提交前删除？ */ //XXX: 提交前清除？ LOG_LWDEBUG("LWLockRelease", lock, "释放等待者");唤醒（锁定）； TRACE_POSTGRESQL_LWLOCK_RELEASE(T_NAME(锁)); /* * 现在可以允许取消/死亡中断。因此，相信大家对《PostgreSQL中插入数据时与WAL相关的处理逻辑是什么》有了更深入的了解，不妨实践一下吧！这是网站。更多相关内容，您可以进入相关渠道进行查询。关注我们并继续学习！