openGauss数据库源码解析系列文章——openGauss简介（下）

/* smgr/smgr.cpp, 存储管理 */
...
/* 文件管理函数列表，包含磁盘初始化、开关、同步等操作函数 */
static const f_smgr g_smgrsw[] = {
    /* 磁盘*/
    {mdinit,
        NULL,
        mdclose,
        mdcreate,
        mdexists,
        mdunlink,
        mdextend,
        mdprefetch,
        mdread,
        mdwrite,
        mdwriteback,
        mdnblocks,
        mdtruncate,
        mdimmedsync,
        mdpreckpt,
        mdsync,
        mdpostckpt,
        mdasyncread,
        mdasyncwrite}};
/*
 * 存储管理初始化
 * 当服务器后端启动时调用
 */
void smgrinit(void)
{
    int i;
/* 初始化所有存储相关管理器 */
    for (i = 0; i < SMGRSW_LENGTH; i++) {
        if (g_smgrsw[i].smgr_init) {
            (*(g_smgrsw[i].smgr_init))();
        }
    }
 
    /* 登记存储管理终止程序 */
    if (!IS_THREAD_POOL_SESSION) {
        on_proc_exit(smgrshutdown, 0);
    }
}
/*
 * 当后端服务关闭时，执行存储管理关闭代码
 */
static void smgrshutdown(int code, Datum arg)
{
    int i;
/* 关闭所有存储关联服务 */
    for (i = 0; i < SMGRSW_LENGTH; i++) {
        if (g_smgrsw[i].smgr_shutdown) {
            (*(g_smgrsw[i].smgr_shutdown))();
        }
    }
}

/* buffer/bufmgr.cpp, 基础行存储管理 */
...
/* 查找或创建一个缓冲区 */
Buffer ReadBufferExtended(
    Relation reln, ForkNumber fork_num, BlockNumber block_num, ReadBufferMode mode, BufferAccessStrategy strategy)
{
    bool hit = false;
    Buffer buf;
 
    if (block_num == P_NEW) {
        STORAGE_SPACE_OPERATION(reln, BLCKSZ);
    }
 
    /* 以smgr(存储管理器)级别 打开一个缓冲区 */
    RelationOpenSmgr(reln);
 
    /* 拒绝读取非局部临时关系的请求，因为可能会获得监控不到的错误数据 */
    if (RELATION_IS_OTHER_TEMP(reln) && fork_num <= INIT_FORKNUM)
        ereport(ERROR,
            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot access temporary tables of other sessions")));
 
    /* 读取缓冲区，更新pgstat数量反馈cache命中与否情况 */
    pgstat_count_buffer_read(reln);
    pgstatCountBlocksFetched4SessionLevel();
    buf = ReadBuffer_common(reln->rd_smgr, reln->rd_rel->relpersistence, fork_num, block_num, mode, strategy, &hit);
    if (hit) {
        pgstat_count_buffer_hit(reln);
    }
    return buf;
}
 
/* 释放一个缓冲区 */
void ReleaseBuffer(Buffer buffer)
{
    BufferDesc* buf_desc = NULL;
    PrivateRefCountEntry* ref = NULL;
    /* 错误释放处理 */
    if (!BufferIsValid(buffer)) {
        ereport(ERROR, (errcode(ERRCODE_INVALID_BUFFER), (errmsg("bad buffer ID: %d", buffer))));
    }
 
    ResourceOwnerForgetBuffer(t_thrd.utils_cxt.CurrentResourceOwner, buffer);
 
    if (BufferIsLocal(buffer)) {
        Assert(u_sess->storage_cxt.LocalRefCount[-buffer - 1] > 0);
        u_sess->storage_cxt.LocalRefCount[-buffer - 1]--;
        return;
    }
    /* 释放当前缓冲区 */
    buf_desc = GetBufferDescriptor(buffer - 1);
 
    PrivateRefCountEntry *free_entry = NULL;
    ref = GetPrivateRefCountEntryFast(buffer, free_entry);
    if (ref == NULL) {
        ref = GetPrivateRefCountEntrySlow(buffer, false, false, free_entry);}
    Assert(ref != NULL);
    Assert(ref->refcount > 0);
 
    if (ref->refcount > 1) {
        ref->refcount--;
    } else {
        UnpinBuffer(buf_desc, false);
    }
}
/* 标记写脏缓冲区 */
void MarkBufferDirty(Buffer buffer)
{
    BufferDesc* buf_desc = NULL;
    uint32 buf_state;
    uint32 old_buf_state;
 
    if (!BufferIsValid(buffer)) {
        ereport(ERROR, (errcode(ERRCODE_INVALID_BUFFER), (errmsg("bad buffer ID: %d", buffer))));}
 
    if (BufferIsLocal(buffer)) {
        MarkLocalBufferDirty(buffer);
        return;
    }
 
    buf_desc = GetBufferDescriptor(buffer - 1);
 
    Assert(BufferIsPinned(buffer));
    /* unfortunately we can't check if the lock is held exclusively */
    Assert(LWLockHeldByMe(buf_desc->content_lock));
 
    old_buf_state = LockBufHdr(buf_desc);
 
    buf_state = old_buf_state | (BM_DIRTY | BM_JUST_DIRTIED);
 
    /* 将未入队的脏页入队 */
    if (g_instance.attr.attr_storage.enableIncrementalCheckpoint) {
        for (;;) {
            buf_state = old_buf_state | (BM_DIRTY | BM_JUST_DIRTIED);
            if (!XLogRecPtrIsInvalid(pg_atomic_read_u64(&buf_desc->rec_lsn))) {
                break;
            }
 
            if (!is_dirty_page_queue_full(buf_desc) && push_pending_flush_queue(buffer)) {
                break;
            }
 
            UnlockBufHdr(buf_desc, old_buf_state);
            pg_usleep(TEN_MICROSECOND);
            old_buf_state = LockBufHdr(buf_desc);
        }
    }
 
    UnlockBufHdr(buf_desc, buf_state);
 
    /* 如果缓冲区不是“脏”状态，则更新相关计数 */
    if (!(old_buf_state & BM_DIRTY)) {
        t_thrd.vacuum_cxt.VacuumPageDirty++;
        u_sess->instr_cxt.pg_buffer_usage->shared_blks_dirtied++;
 
        pgstatCountSharedBlocksDirtied4SessionLevel();
 
        if (t_thrd.vacuum_cxt.VacuumCostActive) {
            t_thrd.vacuum_cxt.VacuumCostBalance += u_sess->attr.attr_storage.VacuumCostPageDirty;
        }
    }
}

/*  cstore_am.cpp */
...
/*  扫描 APIs */
    void InitScan(CStoreScanState *state, Snapshot snapshot = NULL);
    void InitReScan();
    void InitPartReScan(Relation rel);
    bool IsEndScan() const;
 
    /*  晚读 APIs */
    bool IsLateRead(int id) const;
    void ResetLateRead();
 
    /*  更新列存储扫描计时标记*/
    void SetTiming(CStoreScanState *state);
 
    /*  列存储扫描*/
    void ScanByTids(_in_ CStoreIndexScanState *state, _in_ VectorBatch *idxOut, _out_ VectorBatch *vbout);
    void CStoreScanWithCU(_in_ CStoreScanState *state, BatchCUData *tmpCUData, _in_ bool isVerify = false);
 
    /*  加载数据压缩单元描述信息  */
    bool LoadCUDesc(_in_ int col, __inout LoadCUDescCtl *loadInfoPtr, _in_ bool prefetch_control, _in_ Snapshot snapShot = NULL);
 
    /*  从描述表中获取数据压缩单元描述*/
    bool GetCUDesc(_in_ int col, _in_ uint32 cuid, _out_ CUDesc *cuDescPtr, _in_ Snapshot snapShot = NULL);
 
    /*  获取元组删除信息*/
    void GetCUDeleteMaskIfNeed(_in_ uint32 cuid, _in_ Snapshot snapShot);
 
    bool GetCURowCount(_in_ int col, __inout LoadCUDescCtl *loadCUDescInfoPtr, _in_ Snapshot snapShot);
    /* 获取实时行号。*/
    int64 GetLivedRowNumbers(int64 *deadrows);
 
    /*  获得数据压缩单元*/
    CU *GetCUData(_in_ CUDesc *cuDescPtr, _in_ int colIdx, _in_ int valSize, _out_ int &slotId);
 
    CU *GetUnCompressCUData(Relation rel, int col, uint32 cuid, _out_ int &slotId, ForkNumber forkNum = MAIN_FORKNUM,
                            bool enterCache = true) const;
 
    /*  缓冲向量填充 APIs */
    int FillVecBatch(_out_ VectorBatch *vecBatchOut);
 
    /*  填充列向量*/
    template <bool hasDeadRow, int attlen>
    int FillVector(_in_ int colIdx, _in_ CUDesc *cu_desc_ptr, _out_ ScalarVector *vec);
 
    template <int attlen>
    void FillVectorByTids(_in_ int colIdx, _in_ ScalarVector *tids, _out_ ScalarVector *vec);
 
    template <int attlen>
    void FillVectorLateRead(_in_ int seq, _in_ ScalarVector *tids, _in_ CUDesc *cuDescPtr, _out_ ScalarVector *vec);
 
    void FillVectorByIndex(_in_ int colIdx, _in_ ScalarVector *tids, _in_ ScalarVector *srcVec, _out_ ScalarVector *destVec);
 
    /*  填充系统列*/
    int FillSysColVector(_in_ int colIdx, _in_ CUDesc *cu_desc_ptr, _out_ ScalarVector *vec);
 
    template <int sysColOid>
    void FillSysVecByTid(_in_ ScalarVector *tids, _out_ ScalarVector *destVec);
 
    template <bool hasDeadRow>
    int FillTidForLateRead(_in_ CUDesc *cuDescPtr, _out_ ScalarVector *vec);
 
    void FillScanBatchLateIfNeed(__inout VectorBatch *vecBatch);
 
    /* 设置数据压缩单元范围以支持索引扫描 */
    void SetScanRange();
 
    /*  判断行是否可用*/
    bool IsDeadRow(uint32 cuid, uint32 row) const;
 
    void CUListPrefetch();
    void CUPrefetch(CUDesc *cudesc, int col, AioDispatchCUDesc_t **dList, int &count, File *vfdList);
 
    /* 扫描函数 */
    typedef void (CStore::*ScanFuncPtr)(_in_ CStoreScanState *state, _out_ VectorBatch *vecBatchOut);
    void RunScan(_in_ CStoreScanState *state, _out_ VectorBatch *vecBatchOut);
    int GetLateReadCtid() const;
    void IncLoadCuDescCursor();

/* system/mot_engine.cpp */
...
/* 创建内存引擎实例 */
MOTEngine* MOTEngine::CreateInstance(
    const char* configFilePath /* = nullptr */, int argc /* = 0 */, char* argv[] /* = nullptr */)
{
    if (m_engine == nullptr) {
        if (CreateInstanceNoInit(configFilePath, argc, argv) != nullptr) {
            bool result = m_engine->LoadConfig();
            if (!result) {
                MOT_REPORT_ERROR(MOT_ERROR_INTERNAL, "System Startup", "Failed to load Engine configuration");
            } else {
                result = m_engine->Initialize();
                if (!result) {
                    MOT_REPORT_ERROR(MOT_ERROR_INTERNAL, "System Startup", "Engine initialization failed");
                }
            }
 
            if (!result) {
                DestroyInstance();
                MOT_ASSERT(m_engine == nullptr);
            }
        }
    }
    return m_engine;
}
/* 内存引擎初始化 */
bool MOTEngine::Initialize()
{
    bool result = false;
/* 初始化应用服务，开始后台任务 */
    do {  // instead of goto
        m_initStack.push(INIT_CORE_SERVICES_PHASE);
        result = InitializeCoreServices();
        CHECK_INIT_STATUS(result, "Failed to Initialize core services");
 
        m_initStack.push(INIT_APP_SERVICES_PHASE);
        result = InitializeAppServices();
        CHECK_INIT_STATUS(result, "Failed to Initialize applicative services");
 
        m_initStack.push(START_BG_TASKS_PHASE);
        result = StartBackgroundTasks();
        CHECK_INIT_STATUS(result, "Failed to start background tasks");
    } while (0);
 
    if (result) {
        MOT_LOG_INFO("Startup: MOT Engine initialization finished successfully");
        m_initialized = true;
    } else {
        MOT_LOG_PANIC("Startup: MOT Engine initialization failed!");
        /* 调用方应在失败后调用DestroyInstance() */
    }
 
    return result;
}
/* 销毁内存引擎实例 */
void MOTEngine::Destroy()
{
    MOT_LOG_INFO("Shutdown: Shutting down MOT Engine");
    while (!m_initStack.empty()) {
        switch (m_initStack.top()) {
            case START_BG_TASKS_PHASE:
                StopBackgroundTasks();
                break;
 
            case INIT_APP_SERVICES_PHASE:
                DestroyAppServices();
                break;
 
            case INIT_CORE_SERVICES_PHASE:
                DestroyCoreServices();
                break;
 
            case LOAD_CFG_PHASE:
                break;
 
            case INIT_CFG_PHASE:
                DestroyConfiguration();
                break;
 
            default:
                break;
        }
        m_initStack.pop();
    }
    ClearErrorStack();
    MOT_LOG_INFO("Shutdown: MOT Engine shutdown finished");
}