DAOS引擎是如何收到客户端RPC并处理的?自动生成RPC请求参数及结构体

引擎启动:
server_init -> daos_debug_init -> dss_engine_metrics_init -> drpc_init -> register_dbtree_classes -> dss_topo_init -> abt_init -> dss_module_init interface初始化 -> crt_init_opt 网络初始化-> dss_module_init_all -> vos,rdb,rsvc,security,mgmt,dtx,pool,cont,obj,rebuild 模块初始化 -> dss_srv_init 服务初始化

...
dss_module_init_all(&dss_mod_facs) -> 初始化所有模块, vos(vos_mod_init), ...
dss_srv_init() 初始化主服务(服务初始化)
  dss_xstreams_init() -> xs初始化(协程池)
    dss_start_xs_id(tags, xs_id) -> 按tag和xs_id启动xs, 启动系统服务, 主IO服务, 负载等多个XS
      dss_start_one_xstream(obj->cpuset, tag, xs_id) -> 服务端启动单个XS
        dss_sched_init(dx); -> 创建XS调度器
        daos_abt_thread_create(dx->dx_sp, dss_free_stack_cb, dx->dx_pools[DSS_POOL_NET_POLL], dss_srv_handler, dx, attr, &dx->dx_progress) -> 启动轮询(驱动ULT)线程
  ...
  drpc_listener_init 启动drpc监听
  ...


...  
dss_srv_handler
  if (dx->dx_comm)
    rc = crt_context_create(&dmi->dmi_ctx) -> 创建私有的传输上下文
    rc = crt_context_register_rpc_task(dmi->dmi_ctx, dss_rpc_hdlr, dss_iv_resp_hdlr, dx); -> 为RPC任务注册两个公共回调(rpc控制器和iv回复)
      cc_rpc_cb = dss_rpc_hdlr -> 公共RPC回调(被 progress 后由 crt_rpc_handler_common 触发执行)
      cc_iv_resp_cb = dss_iv_resp_hdlr  -> iv RPC回调
    crt_context_idx(dmi->dmi_ctx, &dmi->dmi_ctx_id) -> 获取cart上下文索引
    tse_sched_init(&dx->dx_sched_dsc, NULL, dmi->dmi_ctx) -> 为支持服务端调用客户端的API, 初始化调度器
    for (;;) -> 死循环
      if (dx->dx_comm)
        crt_progress(dmi->dmi_ctx, dx->dx_timeout) -> 轮询/驱动网络回调
        ABT_thread_yield() -> 每次循环让出cpu一次

...
rpc公共回调, 调用栈1
dss_rpc_hdlr
  sched_req_enqueue 调度请求入队
    should_enqueue_req?(dx->dx_main_xs) 主线程才入队(如VOS)
      req_get(对端 req_put) -> req_enqueue 请求入队 -> d_list_add_tail(&req->sr_link, &sri->sri_req_list)
      ...

调度器初始化后就开始处理队列(process_all)
dss_sched_init
  sched_run
    sched_start_cycle(data, pools)
      process_all
        policy_ops[sched_policy].process_io(dx) 处理io
          policy_fifo_process 先进先出
            process_req_list
              req_kickoff
                req_kickoff_internal(dx, &req->sr_attr, req->sr_func,req->sr_arg)  sr_func -> crt_handle_rpc 不入队列，直接处理回调
                  sched_create_thread(dx, func func -> crt_handle_rpc
                    ABT_thread_create(abt_pool, func, arg, t_attr, thread)
                      crt_handle_rpc

process_all -> policy_ops[sched_policy].process_io(dx) -> .process_io = policy_fifo_process -> 
  process_req_list 
    d_list_for_each_entry_safe(req, tmp, list, sr_link) -> 遍历队列另一端, 并处理请求
      process_req -> req_kickoff(dx, req) 开始处理请求 -> crt_handle_rpc


调用栈2
dss_rpc_hdlr(crt_context_t *ctx, void *hdlr_arg, void (*real_rpc_hdlr)(void *), void *arg)
  opc_get_mod_id(rpc->cr_opc) 获取模块id 偏移+掩码 daos_modeul_id
  SCHED_REQ_ANONYM 匿名 sra=调度请求属性
  struct dss_module	*module = dss_module_get(mod_id);
  module->sm_mod_ops->dms_get_req_attr 获取属性
  sched_req_enqueue 入队  real_rpc_hdlr = crt_handle_rpc req->sr_func
    should_enqueue_req SCHED_REQ_ANONYM 匿名不入队列
    req_enqueue
      d_list_add_tail(&req->sr_link, &sri->sri_req_list)



crt_rpc_handler_common
    HG_Get_info
    HG_Context_get_data
    crt_hg_unpack_header
    crt_opc_lookup
    crt_hg_header_copy
    crt_rpc_priv_init
      crp_completed = 0
    crt_rpc_common_hdlr 不是集合rpc
        crt_grp_priv_get_primary_rank
        crt_rpc_cb_customized 自定义回调, 并且非心跳rpc, crt_opc_is_swim, 那么就执行自定义回调
        rc = crt_ctx->cc_rpc_cb((crt_context_t)crt_ctx, &rpc_priv->crp_pub, crt_handle_rpc,crt_ctx->cc_rpc_cb_arg); // 接收端处理, cart的rpc控制器
        cc_rpc_cb = dss_rpc_hdlr -> sched_create_thread(dx, func -> ABT_thread_create -> crt_handle_rpc
        ... 入队，出队
        crt_handle_rpc
            rpc_priv->crp_opc_info->coi_rpc_cb(rpc_pub) 执行回调,如 obj_req_create DAOS_OBJ_RPC_TGT_UPDATE 对应的 void ds_obj_tgt_update_handler(crt_rpc_t *rpc)
                struct obj_rw_in		*orw = crt_req_get(rpc); -> 获取客户端参数
                struct obj_rw_out		*orwo = crt_reply_get(rpc);
    crt_corpc_common_hdlr -> 集合RPC请求

DAOS上下文的 RPC 回调，当上下文接收到任何 RPC 时将调用该回调。 在此回调中，处理程序可以在此上下文中专门为此 RPC 执行某些操作，例如创建另一个 ULT 来处理它，请参阅 DAOS。
typedef int (*crt_rpc_task_t) (crt_context_t *ctx, void *rpc_hdlr_arg, void (*rpc_hdlr)(void *), void *arg);

crt_context_create -> 创建上下文的时候已经注册好了rpc公共回调(crt_rpc_handler_common), 等待被 progress 触发执行
  crt_context_init
  crt_hg_ctx_init
      crt_hg_class_init
          HG_Init_opt
            HG_Init_opt
              NA_Initialize_opt 网络抽象类初始化 class: ofi Protocal: verbs;ofi_rxm Hostname: mlx5_bond_1/ip:50177
                na_private_class->na_class.ops = na_class_table[plugin_index] 抽象网络表 na_ofi_class_ops_g
                na_class.ops->initialize 初始化
                na_class_ops NA_PLUGIN_OPS(ofi) 插件实现
                fi_getinfo
                ofi_check
          crt_hg_get_addr
          crt_hg_reg_rpcid
              crt_hg_reg CRT_HG_RPCID | CRT_HG_ONEWAY_RPCID 单程 -> 注册公共回调
                  crt_proc_in_common
                  crt_proc_out_common
                    rpc_priv = container_of(data, struct crt_rpc_priv, crp_pub.cr_output) -> reply 回复数据
                  crt_rpc_handler_common <- hg_proc_info->rpc_cb <- hg_core_rpc_cb <- hg_core_rpc_info->rpc_cb <- hg_core_process


dss_xstreams_init -> 初始化DAOS服务端xstream(初始化协程调度器及执行ULT调度)
  dss_start_xs_id 0,1..., 计算偏移, 如: xs_id = dss_sys_xs_nr + dss_tgt_nr + i;
    hwloc_bitmap_first 计算位图中的第一个索引（最低有效位）
    dss_start_one_xstream(obj->cpuset, xs_id) 用计算的cpu集启动, 内部绑核
      dss_xstream_alloc(cpus)
        dx->dx_cpuset = hwloc_bitmap_dup(cpus)
      dss_sched_init
        ABT_sched_def		sched_def
        .init	= sched_init,
        .run	= sched_run, -> run 函数被argobot调用
        sched_info_init(dx)
          info->si_cur_ts = daos_getmtime_coarse() 毫秒
          d_hash_table_create si_pool_hash 创建hash表 大小为2的4次方=16
          prealloc_requests(info, SCHED_PREALLOC_INIT_CNT) 预分配请求 8192
            D_ALLOC_PTR(req)
            D_INIT_LIST_HEAD(&req->sr_link)
            d_list_add_tail(&req->sr_link, &info->si_idle_list) 插入链表 
        sched_create_pools(dx); 创建3个池 从预定义类型创建新池。 ABT_pool_create_basic() 创建一个新池，由池类型 kind、访问类型 access 和自动标志 automatic 给出，并通过 newpool 返回其句柄。 kind 指定 newpool 的实现。 有关预定义池的详细信息，请参阅#ABT_pool_kind。 access 提示创建的池的使用。 Argobots 可以为具有更受限制的访问类型的池选择优化的实现（#ABT_POOL_ACCESS_PRIV 是最严格的访问类型）。 有关详细信息，请参阅#ABT_pool_access。 如果 automatic 为 ABT_FALSE，则 newpool 不会自动释放，因此 newpool 在使用后必须由 ABT_pool_free() 释放，除非 newpool 与主执行流的主调度程序相关联。
          ABT_pool_create_basic(ABT_POOL_FIFO, ABT_POOL_ACCESS_MPSC 创建的池可能只被一个执行流弹出
          ABT_sched_config_create(&config, event_freq, 512, dx_ptr, dx, ABT_sched_config_var_end)
          ABT_sched_create(&sched_def, DSS_POOL_CNT, dx->dx_pools, config, &dx->dx_sched) -> 启动调度器(poll) -> sched_run
          使用调度程序定义创建新的调度程序。ABT_sched_create() 创建一个新的调度器，由定义 def 和调度器配置 config 定义，并通过 newsched 返回它的句柄。def 必须定义所有非可选函数。 有关详细信息，请参阅 ABT_sched_def。newsched 与 pools 数组关联，它有 num_pools 个 ABT_pool 句柄。 如果池的第 i 个元素是 ABT_POOL_NULL，则新创建具有默认池配置的默认 FIFO 池并用作第 i 个池。
      ABT_xstream_create_with_rank 创建具有特定等级的新执行流。 ABT_xstream_create_with_rank() 使用调度程序 sched 创建一个新的执行流，并通过 newxstream 返回其句柄。 如果 sched 为 ABT_SCHED_NULL，则使用具有基本 FIFO 队列和默认调度程序配置的默认调度程序。
      ABT_thread_attr_create
      ABT_thread_attr_set_stacksize 在 ULT 属性中设置堆栈大小。 ABT_thread_attr_set_stacksize() 设置 ULT 属性 attr 中的堆栈大小 stacksize（以字节为单位）。 如果堆栈内存已由 ABT_thread_attr_set_stack() 设置，此例程将更新堆栈大小，同时将堆栈内存保留在 attr 中。
      ABT_thread_create dss_srv_handler


dss_srv_handler 服务控制器，设置cpu亲和性，初始化TLS，crt， nvme，gc, nvme, 启动网络poll
  dss_xstream_set_affinity 设置亲和性
  hwloc_set_cpubind 绑核
  hwloc_set_membind 将当前进程或线程的默认内存绑定策略设置为更喜欢由 set 指定的 NUMA 节点。 这是最便携的形式，因为它允许 hwloc 使用基于进程的操作系统功能或基于线程的操作系统功能，具体取决于可用的功能。 如果指定了 ::HWLOC_MEMBIND_BYNODESET，则集合被视为节点集。 否则它是一个cpuset。
  dss_tls_init 初始化本地存储, 为特定线程分配 dss_thread_local_storage 并将指针存储在特定于线程的值中，该值可以随时使用 dss_tls_get() 获取。
    dss_thread_local_storage_init obj_tls_init
      dtls->dtls_values[i] = dmk->dmk_init(xs_id, tgt_id) 
      dmk->dmk_init(xs_id, tgt_id) -> dss_srv_tls_init | vos_tls_init 
    pthread_setspecific(dss_tls_key, dtls)
  dss_get_module_info
    dss_module_key_get
  crt_context_create
  crt_context_register_rpc_task dss_rpc_hdlr dss_iv_resp_hdlr 注册两个公共回调
  crt_context_idx
  tse_sched_init 使用可选的完成回调和指向用户数据的指针初始化调度程序。 调用者负责完成或取消调度程序。
  bio_xsctxt_alloc 初始化spdk环境和nvme上下文, 为主 XS 初始化 SPDK env 和 per-xstream NVMe 上下文
  dss_nvme_poll_ult
  for (;;) 死循环
    crt_progress

自动生成结构体:
CRT_RPC_DECLARE(obj_rw,		DAOS_ISEQ_OBJ_RW, DAOS_OSEQ_OBJ_RW)
#define CRT_RPC_DECLARE(rpc_name, fields_in, fields_out)
rpc输入输出参数, 用于编码/解码rpc数据:
struct obj_rw_in { 
    struct dtx_id orw_dti; 
    daos_unit_oid_t orw_oid; 
    uuid_t orw_pool_uuid; 
    uuid_t orw_co_hdl; 
    uuid_t orw_co_uuid; 
    uint64_t orw_epoch; 
    uint64_t orw_epoch_first; 
    uint64_t orw_api_flags; 
    uint64_t orw_dkey_hash; 
    uint32_t orw_map_ver; 
    uint32_t orw_nr; 
    uint32_t orw_start_shard; 
    uint32_t orw_flags; 
    daos_key_t orw_dkey; 
    struct dcs_csum_info *orw_dkey_csum;
    struct obj_iod_array orw_iod_array;
    struct { 
        uint64_t ca_count;
        struct dtx_id *ca_arrays;
    } orw_dti_cos;
    struct { 
        uint64_t ca_count;
        d_sg_list_t *ca_arrays;
    } orw_sgls;
    struct { 
        uint64_t ca_count;
        crt_bulk_t *ca_arrays;
    } orw_bulks;
    struct { 
        uint64_t ca_count;
        struct daos_shard_tgt *ca_arrays;
    } orw_shard_tgts;
    uint32_t orw_tgt_idx;
    uint32_t orw_tgt_max;
};

struct obj_rw_out { 
    int32_t orw_ret;
    uint32_t orw_map_version;
    uint64_t orw_epoch;
    struct { 
        uint64_t ca_count;
        daos_size_t *ca_arrays;
    } orw_iod_sizes;
    struct { 
        uint64_t ca_count;
        daos_size_t *ca_arrays;
    } orw_data_sizes;
    struct { 
        uint64_t ca_count;
        d_sg_list_t *ca_arrays;
    } orw_sgls;
    struct { 
        uint64_t ca_count;
        uint32_t *ca_arrays;
    } orw_nrs;
    struct { 
        uint64_t ca_count;
        struct dcs_iod_csums *ca_arrays;
    } orw_iod_csums;
    struct { 
        uint64_t ca_count;
        struct daos_recx_ep_list *ca_arrays;
    } orw_rels;
    struct { 
        uint64_t ca_count;
        daos_iom_t *ca_arrays;
    } orw_maps;
};

_Pragma("pack(push, 1)") struct obj_rw_in_packed { 
    struct dtx_id orw_dti;
    daos_unit_oid_t orw_oid;
    uuid_t orw_pool_uuid;
    uuid_t orw_co_hdl;
    uuid_t orw_co_uuid;
    uint64_t orw_epoch;
    uint64_t orw_epoch_first;
    uint64_t orw_api_flags;
    uint64_t orw_dkey_hash;
    uint32_t orw_map_ver;
    uint32_t orw_nr;
    uint32_t orw_start_shard;
    uint32_t orw_flags;
    daos_key_t orw_dkey;
    struct dcs_csum_info *orw_dkey_csum;
    struct obj_iod_array orw_iod_array;
    struct { 
        uint64_t ca_count;
        struct dtx_id *ca_arrays;
    } orw_dti_cos;
    struct { 
        uint64_t ca_count;
        d_sg_list_t *ca_arrays;
    } orw_sgls;
    struct { 
        uint64_t ca_count;
        crt_bulk_t *ca_arrays;
    } orw_bulks;
    struct { 
        uint64_t ca_count;
        struct daos_shard_tgt *ca_arrays;
    } orw_shard_tgts;
    uint32_t orw_tgt_idx;
    uint32_t orw_tgt_max;
};

struct obj_rw_out_packed { 
    int32_t orw_ret;
    uint32_t orw_map_version;
    uint64_t orw_epoch;
    struct { 
        uint64_t ca_count;
        daos_size_t *ca_arrays;
    } orw_iod_sizes;
    struct { 
        uint64_t ca_count;
        daos_size_t *ca_arrays;
    } orw_data_sizes;
    struct { 
        uint64_t ca_count;
        d_sg_list_t *ca_arrays;
    } orw_sgls;
    struct { 
        uint64_t ca_count;
        uint32_t *ca_arrays;
    } orw_nrs;
    struct { 
        uint64_t ca_count;
        struct dcs_iod_csums *ca_arrays;
    } orw_iod_csums;
    struct { 
        uint64_t ca_count;
        struct daos_recx_ep_list *ca_arrays;
    } orw_rels;
    struct { 
        uint64_t ca_count;
        daos_iom_t *ca_arrays;
    } orw_maps;
};
_Pragma("pack(pop)") _Static_assert(__builtin_offsetof(struct obj_rw_out_packed, orw_maps) == __builtin_offsetof(struct obj_rw_out, orw_maps), "obj_rw" " output struct has a hole");
_Static_assert(__builtin_offsetof(struct obj_rw_in_packed, orw_tgt_max) == __builtin_offsetof(struct obj_rw_in, orw_tgt_max), "obj_rw" " input struct has a hole");
extern struct crt_req_format CQF_obj_rw;