openGauss数据库源码解析系列文章——公共组件源码解析（上）

/*  pg_database本身也是一张表，DatabaseRelationId表示pg_database在系统表pg_class中的OID为1262(pg_class系统表中保存的是表的定义信息) */
#define DatabaseRelationId    1262
/*  pg_database本身也是一个结构类型，DatabaseRelation_Rowtype_Id表示pg_database在系统表pg_type中的OID为1248  */
#define DatabaseRelation_Rowtype_Id  1248
/* pg_database本身也是一个结构类型，DatabaseRelation_Rowtype_Id表示pg_database在系统表pg_type中的OID为1248  */
/*  BKI_SHARED_RELATION表示pg_database是实例级别的系统表 */
CATALOG(pg_database,1262) BKI_SHARED_RELATION BKI_ROWTYPE_OID(1248) BKI_SCHEMA_MACRO
{
    NameData    datname;              * 数据库名称 */
    Oid         datdba;               * 数据库拥有者 */
    int4        encoding;             * 字符集编码 */
    NameData    datcollate;           * LC_COLLATE 设置值 */
    NameData    datctype;             * LC_CTYPE 设置值 */
    bool        datistemplate;        * 是否允许作为模板数据库*/
    bool        datallowconn;         * 是否允许链接 */
    int4        datconnlimit;         * 最大连接数*/
    Oid         datlastsysoid;        * 系统OID最大值*/
    ShortTransactionId datfrozenxid;  /* 冻结事务ID，所有小于这个值的事务ID已经冷冻。为了兼容原来的版本，使用32位事务ID*/
    Oid         dattablespace;     /* 数据库的缺省表空间 */
    NameData    datcompatibility;/* 数据库兼容模式,比如除0是报错还是当做正常处理*/
#ifdef CATALOG_VARLEN          /* 下面字段是变长字段 */
    aclitem        datacl[1];        /* 访问权限*/
#endif
    TransactionId datfrozenxid64; /* 冷冻事务ID，64-bit（比特）事务ID */
} FormData_pg_database;

#define CATALOG(name,oid)  typedef struct CppConcat(FormData_,name)

#define Natts_pg_database                14
#define Anum_pg_database_datname        1
#define Anum_pg_database_datdba            2
#define Anum_pg_database_encoding        3
#define Anum_pg_database_datcollate        4
#define Anum_pg_database_datctype        5
#define Anum_pg_database_datistemplate    6
#define Anum_pg_database_datallowconn    7
#define Anum_pg_database_datconnlimit    8
#define Anum_pg_database_datlastsysoid    9
#define Anum_pg_database_datfrozenxid    10
#define Anum_pg_database_dattablespace    11
#define Anum_pg_database_compatibility    12
#define Anum_pg_database_datacl            13
#define Anum_pg_database_datfrozenxid64    14

DATA(insert OID = 1 (  template1 PGUID ENCODING "LC_COLLATE" "LC_CTYPE" t t -1 0 0 1663 "DB_COMPATIBILITY" _null_ 3));
SHDESCR("default template for new databases");
#define TemplateDbOid            1
#define DEFAULT_DATABASE  "postgres"  /*  表示创建默认数据库的名称为postgres */

  {DatabaseRelationId, /* DATABASEOID */
        DatabaseOidIndexId,
        1,
        {ObjectIdAttributeNumber, 0, 0, 0},
        4},

struct cachedesc {
    Oid reloid;   /* 缓存的表的OID */
    Oid indoid;   /* 缓存数据的索引OID */
    int nkeys;    /*  缓存搜索的key的个数 */
    int key[4];   /* key属性的编号*/
    int nbuckets; /* 缓存hash桶的个数*/
};

static void setup_schema(void)
{
    PG_CMD_DECL;
    char** line;
    char** lines;
    int nRet = 0;
    char* buf_features = NULL;
    fputs(_("creating information schema ... "), stdout);
    (void)fflush(stdout);
    lines = readfile(info_schema_file);
    /*
     * 使用-j 避免在information_schema.sql反斜杠处理
     */
    nRet = snprintf_s(
        cmd, sizeof(cmd), sizeof(cmd) - 1, "\"%s\" %s -j template1 >%s", backend_exec, backend_options, DEVNULL);
    securec_check_ss_c(nRet, "\0", "\0");
    PG_CMD_OPEN;
    for (line = lines; *line != NULL; line++) {
        PG_CMD_PUTS(*line);
        FREE_AND_RESET(*line);
    }
    FREE_AND_RESET(lines);
    PG_CMD_CLOSE;
    nRet = snprintf_s(
        cmd, sizeof(cmd), sizeof(cmd) - 1, "\"%s\" %s template1 >%s", backend_exec, backend_options, DEVNULL);
    securec_check_ss_c(nRet, "\0", "\0");
    PG_CMD_OPEN;
    PG_CMD_PRINTF1("UPDATE information_schema.sql_implementation_info "
                   "  SET character_value = '%s' "
                   "  WHERE implementation_info_name = 'DBMS VERSION';\n",
        infoversion);


    buf_features = escape_quotes(features_file);
    PG_CMD_PRINTF1("COPY information_schema.sql_features "
                   "  (feature_id, feature_name, sub_feature_id, "
                   "  sub_feature_name, is_supported, comments) "
                   " FROM E'%s';\n",
        buf_features);
    FREE_AND_RESET(buf_features);
    PG_CMD_CLOSE;
    check_ok();
}

create pg_type 1247 bootstrap rowtype_oid 71
 (
 typname = name ,
 typnamespace = oid ,
 typowner = oid ,
 typlen = int2 ,
 typbyval = bool ,
 typtype = char ,
 typcategory = char ,
 typispreferred = bool ,
 typisdefined = bool ,
 typdelim = char ,
 typrelid = oid ,
 typelem = oid ,
 typarray = oid ,
 typinput = regproc ,
 typoutput = regproc ,
 typreceive = regproc ,
 typsend = regproc ,
 typmodin = regproc ,
 typmodout = regproc ,
 typanalyze = regproc ,
 typalign = char ,
 typstorage = char ,
 typnotnull = bool ,
 typbasetype = oid ,
 typtypmod = int4 ,
 typndims = int4 ,
 typcollation = oid ,
 typdefaultbin = pg_node_tree ,
 typdefault = text ,
 typacl = aclitem[]
 )
INSERT OID = 16 ( bool 11 10 1 t b B t t \054 0 0 1000 boolin boolout boolrecv boolsend - - - c p f 0 -1 0 0 _null_ _null_ _null_ )
INSERT OID = 17 ( bytea 11 10 -1 f b U f t \054 0 0 1001 byteain byteaout bytearecv byteasend - - - i x f 0 -1 0 0 _null_ _null_ _null_ )
...........
close pg_type

ThreadId initialize_thread(ThreadArg* thr_argv)
{
    gs_thread_t thread;
    if (0 != gs_thread_create(&thread, InternalThreadFunc, 1, (void*)thr_argv)) {
        gs_thread_release_args_slot(thr_argv);
        return InvalidTid;
    }
    return gs_thread_id(thread);
}

static void* InternalThreadFunc(void* args)
{
    knl_thread_arg* thr_argv = (knl_thread_arg*)args;
    gs_thread_exit((GetThreadEntry(thr_argv->role))(thr_argv));
    return (void*)NULL;
}
GaussdbThreadEntry GetThreadEntry(knl_thread_role role)
{
    Assert(role > MASTER && role < THREAD_ENTRY_BOUND);
    return GaussdbThreadEntryGate[role];
}
static GaussdbThreadEntry GaussdbThreadEntryGate[] = {GaussDbThreadMain<MASTER>,
    GaussDbThreadMain<WORKER>,
    GaussDbThreadMain<THREADPOOL_WORKER>,
    GaussDbThreadMain<THREADPOOL_LISTENER>,
  ......};

    if (threadPoolActivated) {
        bool enableNumaDistribute = (g_instance.shmem_cxt.numaNodeNum > 1);
        g_threadPoolControler->Init(enableNumaDistribute);
}

if (threadPoolActivated &&!(i < MAXLISTEN && t_thrd.postmaster_cxt.listen_sock_type[i] == HA_LISTEN_SOCKET))
result = g_threadPoolControler->DispatchSession(port);
/*  ThreadPoolControler::DispatchSession的代码实现如下，找到一个会话数最少的线程组，创建会话，把会话添加到线程组的监听线程中  */
int ThreadPoolControler::DispatchSession(Port* port)
{
    ThreadPoolGroup* grp = NULL;
    knl_session_context* sc = NULL;
    grp = FindThreadGroupWithLeastSession();
    if (grp == NULL) {
        Assert(false);
        return STATUS_ERROR;
    }
    sc = m_sessCtrl->CreateSession(port);
    if (sc == NULL)
        return STATUS_ERROR;
    grp->GetListener()->AddNewSession(sc);
    return STATUS_OK;
}

  if (IS_THREAD_POOL_WORKER) {
        u_sess->proc_cxt.MyProcPort->sock = PGINVALID_SOCKET;
        t_thrd.threadpool_cxt.worker->NotifyReady();
}
       if (IS_THREAD_POOL_WORKER) {
            t_thrd.threadpool_cxt.worker->WaitMission();
            Assert(u_sess->status != KNL_SESS_FAKE);
        }
    case 'X':
    case EOF:
                RemoveTempNamespace();
                InitThreadLocalWhenSessionExit();
                if (IS_THREAD_POOL_WORKER) {
                    t_thrd.threadpool_cxt.worker->CleanUpSession(false);
                    break;
                }