Author 晓兵

weixin: ssbandjl

公众号: 云原生云

简介

---

分布式异步对象存储 (DAOS) 是一种开源软件定义的对象存储，专为大规模分布式非易失性内存 (NVM) 而设计。DAOS 利用存储级内存 (SCM) 和 NVM express (NVMe) 等下一代 NVM 技术，同时呈现键值存储接口并提供事务性非阻塞 I/O、高级数据保护和自我修复等功能 商品硬件、端到端数据完整性、细粒度数据控制和弹性存储，以优化性能和成本。

CaRT 是用于大数据和 百万兆级 HPC 的开源 RPC 传输层。它支持传统的 P2P RPC 发送和集体 RPC，后者在一组目标服务器上调用 RPC，并具有可扩展的基于树的消息传播。

项目中，使用rdma, verbs接口, 主要提供RPC和BULK(大块)数据传输， swim协议（集群心跳检测）， rank组员管理， IV接口等功能

主流程

---

** **

cart主流程:

\1. 日志初始化/注册协议和操作码 crt_init_opt

\2. 创建上下文（fabric->domain->endpoint->cq…） crt_context_create

\3. 创建请求 crt_req_create / crt_corpc_req_create RPC请求集合 传入目的端点 -> ep_rank

\4. 发送请求（请求跟踪和取消跟踪） crt_req_send / dss_rpc_send

\5. 查看请求进度和回调（progress和trigger） crt_progress 超时或有完成事件产生时函数返回

\6. 发送回复 crt_reply_send / crt_reply_get

初始化

---

crt_init_opt

d_log_init

crt_setup_log_fac 设置日志基础设施

d_fault_inject_init

data_init

​ dump_envariables 打印环境变量

​ start_rpcid 随机生成起始rpcid

​ d_tm_add_metric 添加指标等

prov_data_init 设置提供者属性，初始化上下文双向链表

crt_hg_init

​ hg_log_set_func(crt_hg_log) 设置日志方法

crt_grp_init

​ crt_primary_grp_init

​ crt_grp_priv_create

​ grp_priv_init_membs

​ crt_grp_lc_create

​ d_hash_table_create_inplace gp_lookup_cache

​ d_hash_table_create_inplace gp_uri_lookup_cache

crt_plugin_init

​ cpg_timeout_cbs 分配超时回调

​ cpg_event_cbs 时间控制器回调

crt_self_test_init

crt_opc_map_create

​ crt_opc_map_L2_create 初始化前16个entry， L2分配32个entry

crt_internal_rpc_register

​ crt_proto_register_internal(&cpf)

​ crt_proto_reg_L1

​ crt_proto_reg_L2

​ crt_proto_reg_L3

​ crt_proto_register(&cpf)

创建请求

---

dss_srv_handler

crt_context_create

crt_context_register_rpc_task(dmi->dmi_ctx, dss_rpc_hdlr /* process_cb */, dss_iv_resp_hdlr /* iv_resp_cb */, dx);

​ cc_rpc_cb = dss_rpc_hdlr

​ cc_iv_resp_cb = dss_iv_resp_hdlr

crt_context_init

crt_hg_ctx_init

​ crt_hg_class_init

​ HG_Init_opt

​ 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

​ crt_rpc_handler_common <- hg_proc_info->rpc_cb <- hg_core_rpc_cb <- hg_core_rpc_info->rpc_cb <- hg_core_process

​ HG_Registered_disable_response 单程禁用响应

​ HG_Context_create

​ HG_Context_set_data

​ crt_hg_pool_init

crt_hg_get_addr

crt_provider_get_ctx_list

crt_provider_inc_cur_ctx_num

crt_swim_init

发送请求

crt_req_send(req, shard_update_req_cb, remote_arg)

crt_context_req_track

​ crt_set_timeout

​ epi_req_waitq 如果拥塞就进等待队列

​ crt_req_timeout_track

​ epi_req_q 否则进飞行队列

crt_req_send_internal

​ crt_req_ep_lc_lookup

​ crt_req_send_immediately

​ crt_hg_req_create

​ HG_Create

​ HG_Reset 重用

​ crt_hg_req_send

​ HG_Forward(hdl, crt_hg_req_send_cb,…) 回调机制:forward_cb -> request_callback -> hg_cb(由HG_Trigger执行回调)

​ HG_Core_get_rpc_data

​ hg_set_struct

​ HG_Core_forward(…,hg_core_forward_cb,…)

​ forward(hg_core_handle) hg_core_forward_na|hg_core_forward_self

​ hg_core_forward_na

​ hg_core_gen_request_tag

​ NA_Msg_recv_expected hg_core_recv_output_cb

​ NA_Msg_send_unexpected hg_core_send_input_cb

​ na_ofi_msg_send_unexpected

​ na_ofi_msg_send

​ fi_rx_addr

​ fi_tsend -> rxm_ep_tsend

​ rxm_get_conn 新建或复用连接

​ rxm_send_common

​ na_ofi_op_retry

** **

推进/进度

crt_progress

crt_hg_progress

do 每轮total = 256

HG_Progress

​ HG_Core_progress

​ hg_core_progress

​ do while(now < deadline)

​ context->poll_set

​ hg_core_poll_try_wait

​ NA_Poll_try_wait

​ na_poll_try_wait -> na_ofi_poll_try_wait

​ fi_trywait

​ wait_try -> util_wait_fd_try

​ rxm_ep_trywait_eq

​ fi_trywait

​ vrb_trywait

​ vrb_save_wc

​ fi_poll -> util_poll_run

​ fi_cq_read

​ fi_eq_read

​ vrb_save_wc

​ hg_core_poll_wait

​ hg_poll_wait

​ hg_core_progress_na

​ NA_Trigger

​ hg_atomic_queue_pop_mc(na_private_context->completion_queue)

​ completion_data.callback

​ hg_core_recv_output_cb | hg_core_send_input_cb

​ NA_Progress

​ for(;;)

​ ops->progress na_ofi_progress

​ do

​ fi_wait

​ na_ofi_cq_read

​ na_ofi_cq_process_event

​ na_ofi_cq_process_retries

​ fi_tsend

​ fi_trecv

​ na_ofi_op_cancel

​ na_ofi_complete

HG_Trigger

​ hg_core_rpc_info->rpc_cb -> common回调(crt_rpc_handler_common)

​ hg_cb(&hg_core_cb_info) 转发回调

crt_context_timeout_check

crt_exec_progress_cb

** **

回调后落盘

---

更新目标， 落盘

ds_obj_tgt_update_handler

obj_ioc_begin 访问vos前的各种检查

​ obj_ioc_begin_lite 设置 lite IO 上下文，目前仅适用于复合 RPC

​ obj_capa_check

​ obj_ioc_init_oca

dtx_handle_resend

vos_dtx_commit

dtx_begin

obj_local_rw

​ obj_local_rw_internal

​ bio_iod_prep

​ obj_bulk_transfer

​ bio_iod_post

​ dma_rw

​ nvme_rw

​ spdk_blob_io_write 完成回调 rw_completion

​ blob_request_submit_op

​ blob_request_submit_op_single

​ bs_batch_write_dev

​ blob_bdev->bs_dev.write

​ bdev_blob_write

​ spdk_bdev_write_blocks

​ spdk_blob_io_read