深入理解PostgreSQL_Buffer_Pin机制

// src/include/storage/buf_internals.h:29-33
/*
 * Buffer state is a single 32-bit variable where following data is combined.
 *
 * - 18 bits refcount      <- Pin 计数
 * - 4 bits usage count    <- 使用频率
 * - 10 bits of flags      <- 状态标志
 */
#define BUF_REFCOUNT_ONE 1
#define BUF_REFCOUNT_MASK ((1U << 18) - 1)  // 最大值 262,143

// src/backend/storage/buffer/bufmgr.c:88, 197-201

#define REFCOUNT_ARRAY_ENTRIES 8  // 快速数组大小

static struct PrivateRefCountEntry PrivateRefCountArray[REFCOUNT_ARRAY_ENTRIES];
static HTAB *PrivateRefCountHash = NULL;  // 溢出哈希表

// bufmgr.c:81-85
typedef struct PrivateRefCountEntry
{
    Buffer      buffer;     // Buffer编号（1-based）
    int32       refcount;   // 本进程的pin次数
} PrivateRefCountEntry;

// bufmgr.c:197-201

// 快速数组：存放最常用的8个buffer
static PrivateRefCountEntry PrivateRefCountArray[REFCOUNT_ARRAY_ENTRIES];

// 溢出哈希表：超过8个时使用
static HTAB *PrivateRefCountHash = NULL;

// 溢出计数
static int32 PrivateRefCountOverflowed = 0;

// 时钟指针：用于选择要移入哈希表的数组项
static uint32 PrivateRefCountClock = 0;

// 预留的空闲条目
static PrivateRefCountEntry *ReservedRefCountEntry = NULL;

static bool
PinBuffer(BufferDesc *buf, BufferAccessStrategy strategy)
{
    Buffer      b = BufferDescriptorGetBuffer(buf);
    bool        result;
    PrivateRefCountEntry *ref;

    // 步骤1: 查找私有引用计数条目
    ref = GetPrivateRefCountEntry(b, true);

    if (ref == NULL)
    {
        // ═══════════════════════════════════════════════════
        // 情况A: 首次pin此buffer，需要更新共享引用计数
        // ═══════════════════════════════════════════════════
        uint32      buf_state;
        uint32      old_buf_state;

        // 步骤2: 预留私有引用计数空间
        ReservePrivateRefCountEntry();

        // 步骤3: 创建新的私有引用计数条目
        ref = NewPrivateRefCountEntry(b);

        // 步骤4: CAS循环更新共享引用计数
        old_buf_state = pg_atomic_read_u32(&buf->state);
        for (;;)
        {
            // 如果buffer被锁定，等待解锁
            if (old_buf_state & BM_LOCKED)
                old_buf_state = WaitBufHdrUnlocked(buf);

            buf_state = old_buf_state;

            // 增加共享引用计数
            buf_state += BUF_REFCOUNT_ONE;

            // 更新usage_count（用于替换算法）
            if (strategy == NULL)
            {
                // 默认策略：增加usage_count直到最大值
                if (BUF_STATE_GET_USAGECOUNT(buf_state) < BM_MAX_USAGE_COUNT)
                    buf_state += BUF_USAGECOUNT_ONE;
            }
            else
            {
                // Ring buffer策略：只设置为1，避免驱逐其他buffer
                if (BUF_STATE_GET_USAGECOUNT(buf_state) == 0)
                    buf_state += BUF_USAGECOUNT_ONE;
            }

            // CAS原子更新
            if (pg_atomic_compare_exchange_u32(&buf->state,
                                               &old_buf_state, buf_state))
            {
                result = (buf_state & BM_VALID) != 0;

                // Valgrind: 标记buffer内存为可访问
                VALGRIND_MAKE_MEM_DEFINED(BufHdrGetBlock(buf), BLCKSZ);
                break;
            }
            // CAS失败，old_buf_state已被更新为当前值，继续循环
        }
    }
    else
    {
        // ═══════════════════════════════════════════════════
        // 情况B: 已经pin过，无需修改共享状态
        // ═══════════════════════════════════════════════════
        result = true;
    }

    // 步骤5: 增加私有引用计数
    ref->refcount++;
    Assert(ref->refcount > 0);

    // 步骤6: 记录到ResourceOwner（用于事务结束时自动释放）
    ResourceOwnerRememberBuffer(CurrentResourceOwner, b);

    return result;
}

static void
PinBuffer_Locked(BufferDesc *buf)
{
    Buffer      b;
    PrivateRefCountEntry *ref;
    uint32      buf_state;

    // 前置条件检查：不应有预存的私有引用
    Assert(GetPrivateRefCountEntry(BufferDescriptorGetBuffer(buf), false) == NULL);

    // Valgrind: 标记buffer内存为可访问
    VALGRIND_MAKE_MEM_DEFINED(BufHdrGetBlock(buf), BLCKSZ);

    // ═══════════════════════════════════════════════════════════
    // 步骤1: 读取state并增加引用计数
    // 因为持有spinlock，可以直接修改（无需CAS）
    // ═══════════════════════════════════════════════════════════
    buf_state = pg_atomic_read_u32(&buf->state);
    Assert(buf_state & BM_LOCKED);  // 确认持有spinlock
    buf_state += BUF_REFCOUNT_ONE;

    // ═══════════════════════════════════════════════════════════
    // 步骤2: 释放spinlock（写入新状态并清除BM_LOCKED）
    // ═══════════════════════════════════════════════════════════
    UnlockBufHdr(buf, buf_state);

    // ═══════════════════════════════════════════════════════════
    // 步骤3: 创建私有引用计数条目（spinlock已释放）
    // ═══════════════════════════════════════════════════════════
    b = BufferDescriptorGetBuffer(buf);
    ref = NewPrivateRefCountEntry(b);
    ref->refcount++;

    // ═══════════════════════════════════════════════════════════
    // 步骤4: 记录到ResourceOwner
    // ═══════════════════════════════════════════════════════════
    ResourceOwnerRememberBuffer(CurrentResourceOwner, b);
}

static void ReservePrivateRefCountEntry(void)
{
    // 如果已有预留，直接返回
    if (ReservedRefCountEntry != NULL)
        return;

    // 首先在数组中查找空闲槽位
    for (i = 0; i < REFCOUNT_ARRAY_ENTRIES; i++)
    {
        if (PrivateRefCountArray[i].buffer == InvalidBuffer)
        {
            ReservedRefCountEntry = &PrivateRefCountArray[i];
            return;
        }
    }

    // 数组已满，使用时钟算法选择victim移入哈希表
    ReservedRefCountEntry =
        &PrivateRefCountArray[PrivateRefCountClock++ % REFCOUNT_ARRAY_ENTRIES];

    // 将victim移入哈希表
    hashent = hash_search(PrivateRefCountHash,
                          &(ReservedRefCountEntry->buffer),
                          HASH_ENTER, &found);
    hashent->refcount = ReservedRefCountEntry->refcount;

    // 清空数组槽位
    ReservedRefCountEntry->buffer = InvalidBuffer;
    ReservedRefCountEntry->refcount = 0;

    PrivateRefCountOverflowed++;
}

static PrivateRefCountEntry *
GetPrivateRefCountEntry(Buffer buffer, bool do_move)
{
    // 1. 先在数组中查找（O(8)）
    for (i = 0; i < REFCOUNT_ARRAY_ENTRIES; i++)
    {
        if (PrivateRefCountArray[i].buffer == buffer)
            return &PrivateRefCountArray[i];
    }

    // 2. 如果没有溢出，直接返回NULL
    if (PrivateRefCountOverflowed == 0)
        return NULL;

    // 3. 在哈希表中查找
    res = hash_search(PrivateRefCountHash, &buffer, HASH_FIND, NULL);

    if (res == NULL)
        return NULL;

    // 4. 如果do_move=true，将条目移回数组以加速后续访问
    if (do_move)
    {
        ReservePrivateRefCountEntry();
        free = ReservedRefCountEntry;
        free->buffer = buffer;
        free->refcount = res->refcount;

        // 从哈希表删除
        hash_search(PrivateRefCountHash, &buffer, HASH_REMOVE, &found);
        PrivateRefCountOverflowed--;

        return free;
    }

    return res;
}

static BufferDesc *
BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
            BlockNumber blockNum, BufferAccessStrategy strategy, bool *foundPtr)
{
    // 创建BufferTag
    INIT_BUFFERTAG(newTag, smgr->smgr_rnode.node, forkNum, blockNum);
    newHash = BufTableHashCode(&newTag);
    newPartitionLock = BufMappingPartitionLock(newHash);

    // ═══════════════════════════════════════════════════════════════
    // 场景1: 在哈希表中查找buffer
    // ═══════════════════════════════════════════════════════════════
    LWLockAcquire(newPartitionLock, LW_SHARED);
    buf_id = BufTableLookup(&newTag, newHash);

    if (buf_id >= 0)
    {
        // Buffer命中！
        buf = GetBufferDescriptor(buf_id);

        // 调用 PinBuffer (不持有spinlock)
        valid = PinBuffer(buf, strategy);        // ← Pin操作

        LWLockRelease(newPartitionLock);
        *foundPtr = true;

        // 处理BM_VALID检查...
        return buf;
    }

    LWLockRelease(newPartitionLock);

    // ═══════════════════════════════════════════════════════════════
    // 场景2: Buffer未命中，需要分配新buffer
    // ═══════════════════════════════════════════════════════════════
    for (;;)
    {
        // 预留私有引用计数空间
        ReservePrivateRefCountEntry();

        // 获取victim buffer（带spinlock返回）
        buf = StrategyGetBuffer(strategy, &buf_state);

        // 调用 PinBuffer_Locked (持有spinlock)
        PinBuffer_Locked(buf);                   // ← Pin操作

        // 处理脏页刷新、重新验证等...

        // 如果成功，返回buffer
        // 否则UnpinBuffer并继续循环
    }
}

static void
UnpinBuffer(BufferDesc *buf, bool fixOwner)
{
    PrivateRefCountEntry *ref;
    Buffer b = BufferDescriptorGetBuffer(buf);

    // 获取私有引用计数条目
    ref = GetPrivateRefCountEntry(b, false);
    Assert(ref != NULL);

    // 从ResourceOwner移除
    if (fixOwner)
        ResourceOwnerForgetBuffer(CurrentResourceOwner, b);

    // 减少私有引用计数
    Assert(ref->refcount > 0);
    ref->refcount--;

    if (ref->refcount == 0)
    {
        // ═══════════════════════════════════════════════════════════
        // 最后一个私有引用，需要更新共享引用计数
        // ═══════════════════════════════════════════════════════════

        // Valgrind: 标记buffer为不可访问
        VALGRIND_MAKE_MEM_NOACCESS(BufHdrGetBlock(buf), BLCKSZ);

        // 确认没有持有content lock
        Assert(!LWLockHeldByMe(BufferDescriptorGetContentLock(buf)));

        // CAS循环减少共享引用计数
        old_buf_state = pg_atomic_read_u32(&buf->state);
        for (;;)
        {
            if (old_buf_state & BM_LOCKED)
                old_buf_state = WaitBufHdrUnlocked(buf);

            buf_state = old_buf_state;
            buf_state -= BUF_REFCOUNT_ONE;  // 减少引用计数

            if (pg_atomic_compare_exchange_u32(&buf->state,
                                               &old_buf_state, buf_state))
            {
                // 如果有等待者且引用计数为0，唤醒它们
                if ((buf_state & BM_PIN_COUNT_WAITER) &&
                    BUF_STATE_GET_REFCOUNT(buf_state) == 0)
                    /* ... wake up waiters ... */
                break;
            }
        }

        // 释放私有引用计数条目
        ForgetPrivateRefCountEntry(ref);
    }
}

// 在ReadBuffer_common开始时
ResourceOwnerEnlargeBuffers(CurrentResourceOwner);  // bufmgr.c:820

// 在PinBuffer/PinBuffer_Locked结束时
ResourceOwnerRememberBuffer(CurrentResourceOwner, b);  // bufmgr.c:1769, 1828

// 在UnpinBuffer时
ResourceOwnerForgetBuffer(CurrentResourceOwner, b);  // bufmgr.c:1850