• postgresql源码学习（24）—— 事务日志⑤-日志写入WAL Buffer

    一、 写入过程简介

    1. 写入步骤

    日志写入WAL Buffer的过程分为两步：

    • 预留空间：组装完成后日志记录的长度已经确定，因此可以先计算这个长度，并在WAL Buffer里预留空间，空间预留的过程通过XLogCtl->Insert->insertpos_lck锁保护。也就是说，每个需要写入WAL日志记录的进程在预留空间时都是互斥的。
    • 数据复制：一旦空间预留完成，数据复制的过程是可以并发的，PG通过WALInsertLocks锁来控制并发复制的过程。PG声明了NUM_XLOGINSERT_LOCKS（目前是8）个WALInsertLocks，每个WALInsertLocks由轻量级+日志写入位置组成。不同进程的不同事务在刷入日志时会随机（参照自己的MyProc->pgprocno）获取一个WALInsertLocks。

    2. WALInsertLocks锁

    1. /*
    2.  * pg声明了NUM_XLOGINSERT_LOCKS（目前是8）个用于wal插入的锁WALInsertLock
    3.  * 值越大可以并发插入的进程越多，但是CPU负载会越高。
    4.  */
    5. #define NUM_XLOGINSERT_LOCKS  8
    1. /* 每个WALInsertLock由“轻量锁+日志写入位置”组成
    2.  * 想要进行日志写入时，必须持有一个WALInsertLock（随机获取，哪一个无所谓）
    3. */
    4. typedef struct
    5. {
    6. 	LWLock		lock;   // 轻量锁，当锁释放时，代表日志已经写入WAL Buffer
    7. 	XLogRecPtr	insertingAt;  // 记录当前日志写入WAL Buffer的进展，不需要跨页写入的小记录不会去更新这个值，通常在日志记录较长时才会更新该值。insertingAt这个变量会在进程将WAL由内存刷往磁盘时读取，以确认所有对该区域的写入操作已完成
    8.  
    9. 	XLogRecPtr	lastImportantAt;   // lastImportantAt contains the LSN of the last important WAL record inserted using a given lock.在待插入的日志记录中，有一些记录是和数据一致性无关的，即使丢失也不影响，这种记录不影响lastImportantAt的值
    10. } WALInsertLock;

    这里我们留下两个问题：

    • 为什么数据复制的并发度只设为8，不设更大？
    • 多进程并发复制数据的冲突怎么解决？

            这个问题的答案在WaitXLogInsertionsToFinish函数，下一篇我们会学习它。

            简单来说，每次WAL刷入磁盘，都会调用这个函数，而这个函数需要遍历所有WALInsertLocks，所以NUM_XLOGINSERT_LOCKS不宜过大，目前代码中写死为8。
     

    1. for (i = 0; i < NUM_XLOGINSERT_LOCKS; i++)
    2. {
    3.  ...
    4. }

    二、 XLogInsertRecord函数

    如前所述，这个代码最重要就干两件事：

    • 调用ReserveXLogInsertLocation函数，为之前组装好的XLOG预留空间，返回预留的StartPos（起始位置）和EndPos（结束位置）。
    • 调用CopyXLogRecordToWAL函数，将XLOG数据复制到WAL Buffer
    • 最后返回的是XLOG的EndPos，即当前写入日志已经到哪个位置了

    函数开头是一些检查

    1. XLogRecPtr
    2. XLogInsertRecord(XLogRecData *rdata,
    3. 				 XLogRecPtr fpw_lsn,
    4. 				 uint8 flags,
    5. 				 int num_fpi)
    6. {
    7. 	XLogCtlInsert *Insert = &XLogCtl->Insert;
    8. 	pg_crc32c	rdata_crc;
    9. 	bool		inserted;
    10. 	XLogRecord *rechdr = (XLogRecord *) rdata->data;
    11. 	uint8		info = rechdr->xl_info & ~XLR_INFO_MASK;
    12. 	bool		isLogSwitch = (rechdr->xl_rmid == RM_XLOG_ID &&
    13. 							   info == XLOG_SWITCH);
    14. 	XLogRecPtr	StartPos;
    15. 	XLogRecPtr	EndPos;
    16. 	bool		prevDoPageWrites = doPageWrites;
    17. …
    18. START_CRIT_SECTION();
    19. // WAL日志段切换期间会拿排他锁，此时其他进程不能预留空间
    20. 	if (isLogSwitch)
    21. 		WALInsertLockAcquireExclusive();
    22. 	else
    23. 		WALInsertLockAcquire();
    24.  
    25. // 进程当前copy的RedoRecPtr有没有过期，如果过期了（只会发生在恰好做完checkpoint操作），需要回到调用函数重新计算，因此这种场景下会比其他场景慢。
    26. 	if (RedoRecPtr != Insert->RedoRecPtr)
    27. 	{
    28. 		Assert(RedoRecPtr < Insert->RedoRecPtr);
    29. 		RedoRecPtr = Insert->RedoRecPtr;
    30. 	}
    31. // 另外要检查是否启用了 fullPageWrites 或者 forcePageWrites
    32. 	doPageWrites = (Insert->fullPageWrites || Insert->forcePageWrites);
    33.  
    34. 	if (doPageWrites &&
    35. 		(!prevDoPageWrites ||
    36. 		 (fpw_lsn != InvalidXLogRecPtr && fpw_lsn <= RedoRecPtr)))
    37. 	{
    38. 		/*
    39. 		 * Oops, some buffer now needs to be backed up that the caller didn't
    40. 		 * back up.  Start over.如果人家配了但你没做全页写，需要回炉重做，直接报错返回
    41. 		 */
    42. 		WALInsertLockRelease();
    43. 		END_CRIT_SECTION();
    44. 		return InvalidXLogRecPtr;
    45. 	}

    预留空间部分

    1. 	/*
    2. 	 * Reserve space for the record in the WAL. This also sets the xl_prev pointer.
    3.      * 预留空间，这步也会设置xl_prev指针
    4. 	 */
    5. 	if (isLogSwitch)
    6.     // 如果是日志切换记录，恰好需要做日志切换，则可能StartPos和EndPos相同，也就是说不需要记这个WAL日志记录
    7. 		inserted = ReserveXLogSwitch(&StartPos, &EndPos, &rechdr->xl_prev);
    8. 	else
    9. 	{
    10. 		ReserveXLogInsertLocation(rechdr->xl_tot_len, &StartPos, &EndPos,
    11. 								  &rechdr->xl_prev);
    12. 		inserted = true;
    13. 	}

    数据复制部分

    1. // 预留空间之后，开始做数据复制。inserted 为true，表示非日志切换记录
    2. 	if (inserted)
    3. 	{
    4. 		/*
    5. 		 * Now that xl_prev has been filled in, calculate CRC of the record header.目前xl_prev已经填充了，对记录头做cdc校验
    6. 		 */
    7. 		rdata_crc = rechdr->xl_crc;
    8. 		COMP_CRC32C(rdata_crc, rechdr, offsetof(XLogRecord, xl_crc));
    9. 		FIN_CRC32C(rdata_crc);
    10. 		rechdr->xl_crc = rdata_crc;
    11.  
    12. 		/*
    13. 		 * All the record data, including the header, is now ready to be
    14. 		 * inserted. Copy the record in the space reserved. 将日志记录复制到WAL Buffer
    15. 		 */
    16. 		CopyXLogRecordToWAL(rechdr->xl_tot_len, isLogSwitch, rdata,
    17. 							StartPos, EndPos);
    18.  
    19. 		/*
    20. 		 * Unless record is flagged as not important, update LSN of last
    21. 		 * important record in the current slot. When holding all locks, just
    22. 		 * update the first one.除非是一些被标记为不重要的数据，否则都需要更新当前槽位的lastImportantAt值，如果holdingAllLocks为真，则更新第一个值
    23. 		 */
    24. 		if ((flags & XLOG_MARK_UNIMPORTANT) == 0)
    25. 		{
    26. 			int			lockno = holdingAllLocks ? 0 : MyLockNo;
    27.  
    28. 			WALInsertLocks[lockno].l.lastImportantAt = StartPos;
    29. 		}
    30. 	}
    31. 	else // inserted 为false，表示日志切换记录
    32. 	{
    33. 		/*
    34. 		 * This was an xlog-switch record, but the current insert location was
    35. 		 * already exactly at the beginning of a segment, so there was no need
    36. 		 * to do anything. 这是一条日志切换记录，但当前插入位置正好在段的开始位置，因此什么都不用干（因为没东西可以复制）。
    37. 		 */
    38. 	}
    39.  
    40. 	/*
    41. 	 * Done! Let others know that we're finished.操作完成，释放锁
    42. 	 */
    43. 	WALInsertLockRelease();
    44.  
    45. 	MarkCurrentTransactionIdLoggedIfAny();
    46.  
    47. 	END_CRIT_SECTION();
    48. …
    49. 	/*
    50. 	 * Update our global variables
    51. 	 */
    52. 	ProcLastRecPtr = StartPos;
    53. 	XactLastRecEnd = EndPos;
    54. …
    55. 	return EndPos;
    56. }

    三、 空间预留函数 ReserveXLogInsertLocation

    • 为WAL记录（在WAL Buffer中）预留适当大小的空间。
    • StartPos是保留位置的开头，*EndPos是保留位置结尾+1（end+1）,*PrevPtr是前一条记录的开头位置，它用于设置该记录的 xl_prev
    • 这部分对于XLogInsert函数的性能非常重要，因为这是只能串行执行的，而其余部分基本都可以并发处理。因此要确保这部分尽量简短，insertpos_lck在繁忙系统上可能遇到激烈竞争。
    • 注意：这里的空间计算必须与后面的数据复制函数 CopyXLogRecordToWAL中的代码相匹配。

    1. static void
    2. ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos, XLogRecPtr *EndPos,
    3.                           XLogRecPtr *PrevPtr)
    4. {
    5.     XLogCtlInsert *Insert = &XLogCtl->Insert;
    6.     uint64      startbytepos;
    7.     uint64      endbytepos;
    8.     uint64      prevbytepos;
    9.  
    10.     size = MAXALIGN(size);
    11.  
    12.     /* All (non xlog-switch) records should contain data. */
    13.     Assert(size > SizeOfXLogRecord);
    14.  
    15.     /*
    16.      * 这部分是核心，也是真正串行执行的部分，务必要快
    17.      */
    18.     SpinLockAcquire(&Insert->insertpos_lck);
    19.  
    20.     startbytepos = Insert->CurrBytePos;
    21.     endbytepos = startbytepos + size;
    22.     prevbytepos = Insert->PrevBytePos;
    23.     Insert->CurrBytePos = endbytepos;
    24.     Insert->PrevBytePos = startbytepos;
    25.  
    26.     SpinLockRelease(&Insert->insertpos_lck);
    27.  
    28.     *StartPos = XLogBytePosToRecPtr(startbytepos);
    29.     *EndPos = XLogBytePosToEndRecPtr(endbytepos);
    30.     *PrevPtr = XLogBytePosToRecPtr(prevbytepos);
    31.  
    32.     /*
    33.      * Check that the conversions between "usable byte positions" and
    34.      * XLogRecPtrs work consistently in both directions.
    35.      */
    36.     Assert(XLogRecPtrToBytePos(*StartPos) == startbytepos);
    37.     Assert(XLogRecPtrToBytePos(*EndPos) == endbytepos);
    38.     Assert(XLogRecPtrToBytePos(*PrevPtr) == prevbytepos);
    39. }

    四、 数据复制函数 CopyXLogRecordToWAL

    将WAL记录数据复制到WAL Buffer中预留好的空间。

    函数参数：

    • write_len：XLOG的总长度，用于做校验。
    • isLogSwitch：该记录是否是日志切换记录
    • rdata：XLogRecData链表，存放了XLOG的数据。
    • StartPos：XLOG的写入开始位置
    • EndPos：XLOG的结束位置，用于做校验
    1. static void
    2. CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata,
    3.                     XLogRecPtr StartPos, XLogRecPtr EndPos)
    4. {
    5.     char       *currpos;
    6.     int         freespace;
    7.     int         written;
    8.     XLogRecPtr  CurrPos;
    9.     XLogPageHeader pagehdr;
    10.  
    11.     /*
    12.      * Get a pointer to the right place in the right WAL buffer to start
    13.      * inserting to.复制操作的起点
    14.      */
    15.     CurrPos = StartPos;
    16.     currpos = GetXLogBuffer(CurrPos);
    17.     freespace = INSERT_FREESPACE(CurrPos);
    18.  
    19.     /*
    20.      * there should be enough space for at least the first field (xl_tot_len) on this page.
    21.      */
    22.     Assert(freespace >= sizeof(uint32));
    23.  
    24.     /* Copy record data，核心代码，循环复制rdata数组中每个元素的数据 */
    25.     written = 0;
    26.     while (rdata != NULL)
    27.     {
    28.         char       *rdata_data = rdata->data;
    29.         int         rdata_len = rdata->len;
    30.  
    31. /* 用于处理当前需要写入的XLOG长度大于WAL Buffer中当前page的可用空间的情况，此时需要先将XLOG一部分写入当前page，然后再切换到下一个page。 */
    32.         while (rdata_len > freespace)
    33.         {
    34.             /*
    35.              * Write what fits on this page, and continue on the next page.
    36.              */
    37.             Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || freespace == 0);
    38.             memcpy(currpos, rdata_data, freespace);
    39.             rdata_data += freespace;
    40.             rdata_len -= freespace;
    41.             written += freespace;
    42.             CurrPos += freespace;
    43.  
    44.             /*
    45.              *获取下一个page开头位置的指针，并在页头设置xlp_rem_len
    46.              */
    47.             currpos = GetXLogBuffer(CurrPos);
    48.             pagehdr = (XLogPageHeader) currpos;
    49.             pagehdr->xlp_rem_len = write_len - written;
    50.             pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD;
    51.  
    52.             /* skip over the page header，跳过页头部分 */
    53.             if (XLogSegmentOffset(CurrPos, wal_segment_size) == 0)
    54.             {
    55.                 CurrPos += SizeOfXLogLongPHD;
    56.                 currpos += SizeOfXLogLongPHD;
    57.             }
    58.             else
    59.             {
    60.                 CurrPos += SizeOfXLogShortPHD;
    61.                 currpos += SizeOfXLogShortPHD;
    62.             }
    63.             freespace = INSERT_FREESPACE(CurrPos);
    64.         }
    65.  
    66.         Assert(CurrPos % XLOG_BLCKSZ >= SizeOfXLogShortPHD || rdata_len == 0);
    67.         memcpy(currpos, rdata_data, rdata_len);
    68.         currpos += rdata_len;
    69.         CurrPos += rdata_len;
    70.         freespace -= rdata_len;
    71.         written += rdata_len;
    72.  
    73.         rdata = rdata->next;
    74.     }
    75.     Assert(written == write_len);
    76.     …
    77.     if (CurrPos != EndPos)
    78.         elog(PANIC, "space reserved for WAL record does not match what was written");
    79. }

    参考

    《PostgreSQL技术内幕：事务处理深度探索》第4章

    https://blog.csdn.net/obvious__/article/details/119242661?spm=1001.2014.3001.5502

    PostgreSQL的wal日志并发写入机制 - 知乎

    https://blog.csdn.net/asmartkiller/article/details/121375548

    https://icode.best/i/12479444350651

  • 相关阅读:
    GitLab CI/CD 自动化部署-springBoot-demo示例
    PS抠图后有毛边怎么处理？
    五、Spring Boot 整合持久层技术（4）
    这 20 道 Redis 经典面试题你还不会，就别去面试了！
    StarRocks 易用性全面提升：数据导入可以如此简单
    致敬技术与创新·20231024程序员节
    java学习之springcloud之服务注册与发现篇
    ROS 工作空间
    Android 使用DownloadMananger下载图片
    【系统架构设计】架构核心知识： 3.6 负载均衡和Session
  • 原文地址：https://blog.csdn.net/Hehuyi_In/article/details/125447500