DAOS分布式存储_用户态文件系统dfuse_IO全路径(任务调度_RPC_RDMA_BULK_SPDK_NVME_EC_SGL等)

title: "DAOS用户态文件系统IO路径(dfuse io全路径)"
date: 2023-09-03T14:42:07+08:00
draft: true
categories: ['stor', '存储']
tags: ['linux', 'stor']

#创建池, 查池
dmg pool create sxb -z 4g; dmg pool list --verbose
​
#创建容器, 查容器
daos container create sxb --type POSIX sxb; daos container query sxb sxb --verbose; daos cont get-prop sxb sxb
​
#挂载容器到/tmp/sxb分区
mkdir -p /tmp/sxb; dfuse --mountpoint=/tmp/sxb --pool=sxb --cont=sxb; df -h
​
#写文件
cd /tmp/sxb
for i in {0..5};do
  echo "$i, `date`"
  dd if=/dev/zero of=$i bs=1M count=100 oflag=direct
  sleep 3
done

客户端写数据：xb/write.c -> write(fd, direct_write_buf, BUF_SIZE)
write -> dfuse_cb_write 回调写 src/client/dfuse/fuse3
  fuse_req_userdata
  fuse_req_ctx
  fuse_buf_size(bufv)
  ibuf = FUSE_BUFVEC_INIT(len) 分配本地缓冲区
  DFUSE_TRA_DEBUG 调试
  dfuse_mcache_evict -> 清除此处的元数据缓存，以便查找不会返回过时的大小/时间信息
  fuse_buf_copy(&ibuf, bufv, 0)
  dfuse_cache_evict
  d_slab_acquire 以高效的方式分配数据
  fuse_buf_copy libfuse
  daos_event_init 线程事件初始化
    evx->evx_status = DAOS_EVS_READY
    D_INIT_LIST_HEAD(&evx->evx_child) 初始化链表
    daos_eq_putref 从事件队列继承传输上下文
  ev->de_complete_cb = dfuse_cb_write_complete -> 设置回调
  d_iov_set(&ev->de_iov, ibuf.buf[0].mem, len)  # 设置io向量， 将第二参数的地址和长度赋值给第一个参数
  ev->de_sgl.sg_iovs = &ev->de_iov   sgl分散聚集列表
  readahead ie_truncated 预读和截断
  dfs_write (文件系统， 对象，sgl列表，文件(对象)偏移，事件) 将数据写到文件对象
    事件为空
      daos_event_launch
      daos_event_complete
    daos_event_errno_rc(ev) 将错误码转正
      daos_ev2evx(ev)
    daos_array_write 写数组对象: daos_array_write(obj->oh, DAOS_TX_NONE, &iod, sgl, ev)
      dc_task_create(dc_array_write, NULL, ev, &task) -> 创建任务
      args = dc_task_get_args(task)
      dc_task_schedule(task, true)  task与args做转换: dc_task_get_args 调度任务
    return daos_der2errno(rc)
  sem_post(&fs_handle->dpi_sem)   解锁信号量(+1,如果大于0,其他线程将被唤醒执行),唤醒线程（线程同步）: dfuse_progress_thread sem_wait(&fs_handle->dpi_sem)

​
dc_array_write
  daos_task_get_args task和args可互转
  dc_array_io opc = DAOS_OPC_ARRAY_WRITE 操作码是写数组  读:DAOS_OPC_ARRAY_READ
    array_hdl2ptr
    io_extent_same
    D_INIT_LIST_HEAD(&io_task_list)
    daos_task_create(DAOS_OPC_ARRAY_GET_SIZE 短读任务 DAOS_OPC_ARRAY_READ
    while (u < rg_iod->arr_nr) 遍历每个范围，但同时组合属于同一 dkey 的连续范围。 如果用户给出的范围不增加偏移量，则它们可能不会合并，除非分隔范围也属于同一个 dkey
      compute_dkey 计算分布式key 在给定此范围的数组索引的情况下计算 dkey。 还计算从我们开始的索引开始，dkey 可以保存的记录数写作。 相对于 dkey 的记录索引, 比如10B, dkey_val=1
      struct io_params *prev, *current 如果有多个dkey io, 则通过链表连接起来
      num_ios++
      d_iov_set(dkey, &params->dkey_val, sizeof(uint64_t));
      d_iov_set(&iod->iod_name, &params->akey_val, 1);
      compute_dkey 再次计算dkey
      create_sgl 创建分散聚集列表
      daos_task_create(DAOS_OPC_OBJ_FETCH 读: DAOS_OPC_ARRAY_READ 按索引号 -> dc_obj_fetch_task
      daos_task_create(DAOS_OPC_OBJ_UPDATE 写 或 DAOS_OPC_ARRAY_PUNCH truncate dc_funcs[opc].task_func 客户端方法数组
      daos_task_get_args
      tse_task_register_deps 注册在计划任务之前需要完成的依赖任务。 依赖任务无法进行, 如果一个任务依赖于其他任务，只有依赖的任务完成了，才可以将任务添加到调度器列表中
      tse_task_list_add(io_task, &io_task_list)  d_list_add_tail(&dtp->dtp_task_list, head); 添加任务到链表
    tse_task_register_comp_cb(task, free_io_params_cb, &head, sizeof(head))  为任务注册完成回调
    if (op_type == DAOS_OPC_ARRAY_READ && array->byte_array) 短读
      tse_task_register_deps(task, 1, &stask) 注册依赖任务, 最终是子减父引用
      tse_task_list_add(stask, &io_task_list) 加到io任务列表
    tse_task_list_sched(&io_task_list, false); 批量任务调度执行 -> dc_obj_update_task

dc_obj_update_task(tse_task_t *task) DAOS_OPC_OBJ_UPDATE 写
  obj_req_valid(task, args, DAOS_OBJ_RPC_UPDATE
    obj_auxi = tse_task_stack_push(task, sizeof(*obj_auxi)) -> 将任务压栈
      pushed_ptr = dtp->dtp_buf + sizeof(dtp->dtp_buf) - dtp->dtp_stack_top
    ...
    dc_io_epoch_set(epoch, opc)
    tse_task_stack_pop -> 将任务从栈上弹出来
      poped_ptr = dtp->dtp_buf + sizeof(dtp->dtp_buf) - dtp->dtp_stack_top
  dc_tx_attach(args->th, obj, DAOS_OBJ_RPC_UPDATE, task) 如果事务有效(hdl.cookie == 1), 则走dtx
  return dc_obj_update(task, &epoch, map_ver, args, obj) -> 提交对象更新
    obj_task_init(task, DAOS_OBJ_RPC_UPDATE, map_ver, args->th, &obj_auxi, obj)
      obj_task_init_common(task, opc, map_ver, th, auxi, obj)
        tse_task_stack_push
        shard_task_list_init(obj_auxi)
        obj_auxi->is_ec_obj = obj_is_ec(obj) -> 设置EC对象标志
      tse_task_register_comp_cb(task, obj_comp_cb, NULL, 0) -> 为任务注册对象完成回调, 弹出任务参数, 重试, 错误处理等
      ----------------------
    obj_update_sgls_dup(obj_auxi, args) -> 用户可能提供 iov_len < iov_buf_len 的 sql，这可能会给内部处理带来一些麻烦，例如 crt_bulk_create/daos_iov_left() 总是使用 iov_buf_len。 对于这种情况，我们复制 sql 并使其 iov_buf_len = iov_len
    obj_auxi->dkey_hash = obj_dkey2hash(obj->cob_md.omd_id, args->dkey) -> 比如为1
    if (obj_is_ec(obj)) -> 如果是EC对象(对象类属性上的封装方法为 DAOS_RES_EC ), 则重新组装对象写请求
      obj_rw_req_reassemb(obj, args, NULL, obj_auxi) -> 配置了EC的对象需要重新组装, 对象的读写请求
        struct obj_reasb_req	*reasb_req = &obj_auxi->reasb_req -> EC请求, 重新组装 obj 请求。 用户输入的 iod/sgl 可能需要在发送到服务器之前在客户端重新组装，例如：合并相邻的recx，或者对无序的recx进行排序并生成新的sgl与之匹配； 对于EC obj，将iod/recxs拆分到每个目标，生成新的sgl与之匹配，创建oiod/siod以指定每个shard/tgt的IO req
        if (!obj_auxi->req_reasbed)
          obj_reasb_req_init(&obj_auxi->reasb_req, obj, args->iods, args->nr) -> 创建reasb_req并设置iod的值，从输入iod中重用缓冲区，iod_type / iod_size分配为输入iod，iod_kcsum / iod_nr / iod_recx / iod_csums / iod_eprs数组将设置为0 / NULL
          daos_recx_t -> 记录是任意长度的原子 blob，它总是作为一个整体来获取/更新。 记录的大小可能会随着时间的推移而改变。 记录由以下复合键唯一标识： - 分布键（又名 dkey）表示位于同一存储目标上的一组数组。 dkey 具有任意大小。 - 属性键（又名 akey）区分各个数组。 同样，akey 具有任意大小。 - 数组中的索引区分各个记录。 索引是一个范围从零到无穷大的整数。 一系列索引标识称为范围的连续记录集。 范围内的所有记录必须具有相同的大小。 记录范围是数组内相同大小的连续记录范围。 rx_idx 是该范围的第一个数组索引，rx_nr 是该范围覆盖的记录数
            reasb_req->orr_oca = obj_get_oca(obj)
            size_iod = roundup(sizeof(daos_iod_t) * iod_nr, 8)
            ...    
        obj_ec_req_reasb(obj, args->iods, obj_auxi->dkey_hash, args->sgls, reasb_req, args->nr, obj_auxi->opc == DAOS_OBJ_RPC_UPDATE);
            for (i = 0; i < iod_nr; i++)
                obj_ec_singv_req_reasb
                    ec_recx_array->oer_k = oca->u.ec.e_k
                    ec_recx_array->oer_p = oca->u.ec.e_p
                    if (obj_ec_singv_one_tgt(iod->iod_size, sgl, oca))
                        obj_ec_fail_info_parity_get
                        obj_ec_singv_small_idx
                        obj_ec_set_parity_bitmaps
                        obj_ec_parity_tgt_nr
                    obj_ec_singv_cell_bytes
                    obj_io_desc_init
                    codec = codec_get(reasb_req, obj->cob_md.omd_id) -> isal支持
                        reasb_req->orr_codec = obj_ec_codec_get(daos_obj_id2class(oid))
                            daos_array_find(ecc_array, oc_ec_codec_nr, oc_id, &ecc_sort_ops)
                            daos_array_find(ecc_array, oc_ec_codec_nr, oc_id, &ecc_redun_sort_ops)
                    obj_ec_singv_encode(codec, oca, iod, sgl, ec_recx_array) -> DAOS-7539 EC：自定义 EC 单元大小 (#5832)，用户可以通过池或容器属性指定 EC 单元大小，DAOS_PROP_PO_EC_CELL_SZ，设置池的默认 EC 单元大小，DAOS_PROP_CO_EC_CELL_SZ，设置容器的默认 EC 单元大小，如果是 EC 为池和容器都设置了单元格大小，然后容器的值会覆盖池的值。 此补丁将 EC 单元大小从属性应用到客户端 EC 堆栈、服务器 I/O 处理程序、数据迁移服务和 EC 聚合服务。 - EC单元大小应为4K的倍数且小于1MB，默认单元大小仍为1MB，后续应更改修补
                        obj_ec_pbufs_init(recxs, c_bytes)
                        obj_ec_recx_encode(codec, oca, iod, sgl, recxs) -> 编码满条带
                            for (i = 0; i < recx_nr; i++)
                                obj_ec_stripe_encode(iod, sgl, iov_idx, iov_off, codec, oca, cell_bytes, parity_buf)
                                    ec_encode_data(cell_bytes, k, p, codec->ec_gftbls, data, parity_bufs) -> ec编码, isa-l 中 ec_init_tables() 的用途: https://blog.csdn.net/choumin/article/details/126898021
                    d_sgl_init(r_sgl, sgl->sg_nr + obj_ec_parity_tgt_nr(oca)) -> 重组sgl
                    d_iov_set(&r_sgl->sg_iovs[iov_nr + idx]
                    obj_reasb_req_dump(reasb_req, sgl, oca, 0, iod_idx) -> 打印EC调试信息
                obj_ec_recx_scan
                    obj_ec_recx_cell_nr
                    ec_partial_tgt_recx_nrs
                    ec_all_tgt_recx_nrs
                    obj_ec_recov_tgt_recx_nrs
                    obj_ec_recxs_init
                    obj_io_desc_init
                    obj_ec_riod_init
                    obj_ec_seg_sorter_init
                    obj_ec_pbufs_init
                obj_ec_recx_reasb -> 为EC重组iod/sgl/recx, 输入iod, sgl, recx_array, 输出riod, rsgl, oiod
                    recx_with_full_stripe
                    ec_recov_recx_seg_add
                    ec_data_recx_add
                    ec_data_seg_add
                    ec_parity_recx_add
                    ec_parity_seg_add
                    obj_ec_seg_pack
            obj_ec_encode
                obj_ec_recx_encode -> 对全条带recx_array中的数据进行编码，结果奇偶校验存储在struct obj_ec_recx_array::oer_pbufs中
                obj_ec_stripe_encode -> 编码一个完整的条带，结果奇偶校验缓冲区将被填满
    obj_update_shards_get
    obj_shards_2_fwtgts -> 根据分片查找转发的目标
      req_tgts->ort_shard_tgts = req_tgts->ort_tgts_inline -> 分片目标数组，包含 (ort_grp_nr * ort_grp_size) 个目标。 如果#targets <= OBJ_TGT_INLINE_NR 那么它指向ort_tgts_inline。 在数组中，[0, ort_grp_size - 1] 表示第一组，[ort_grp_size, ort_grp_size * 2 - 1] 表示第二组，依此类推。 如果 (ort_srv_disp == 1)，则在每个组中，第一个目标是领导分片，后面的 (ort_grp_size - 1) 目标是前向非领导分片。 现在只有一种情况 (ort_grp_nr > 1) 用于对象打孔，所有其他情况均为 (ort_grp_nr == 1)
      obj_shard_tgts_query -> 分片目标查询
    obj_csum_update
    -------------------
    obj_req_get_tgts 获取对象对应的目标
      obj_dkey2grpmemb
        obj_dkey2grpidx
          pool_map_ver = pool_map_get_version(pool->dp_map)
          grp_size = obj_get_grp_size(obj)
          grp_idx = d_hash_jump(hash, obj->cob_shards_nr / grp_size) how hash generate? obj with pool
      obj_shards_2_fwtgts
        obj_shard_tgts_query 分片目标查询
          obj_shard_open
            dc_obj_shard_open
              pool_map_find_target 二分查找
                comp_sorter_find_target(sorter, id)
                  daos_array_find
                    array_bin_search
          obj_shard2tgtid
            *tgt_id = obj->cob_shards->do_shards[shard].do_target_id -> dc_obj_layout 客户端对象布局
          obj_shard_close(obj_shard)
        obj_auxi->flags |= ORF_CONTAIN_LEADER -> 要求转发给容器leader
        obj_grp_leader_get
          pl_select_leader obj_get_shard
            array_bin_search 二分查找 daos_obj_classes
    tse_task_register_comp_cb(task, obj_comp_cb, NULL, 0)
    obj_csum_update(obj, args, obj_auxi)
    obj_rw_bulk_prep(obj, args->iods, args->sgls, args->nr, true, obj_auxi->req_tgts.ort_srv_disp, task, obj_auxi) -> 准备读写大块数据
      daos_sgls_packed_size -> 内联提取需要将 sqls 缓冲区打包到 RPC 中，因此使用它来检查是否需要批量传输
      obj_bulk_prep
        crt_bulk_create
        crt_bulk_bind -> 将批量句柄绑定到本地上下文，将本地上下文的源地址与批量句柄关联起来。 它可用于将批量句柄从一台服务器转发/共享到另一台服务器，在这种情况下，批量句柄的原始地址可以即时序列化/反序列化。 示例用法：客户端向服务器 A 发送嵌入批量句柄的 RPC 请求，服务器 A 将客户端批量句柄转发到另一台服务器 B。对于该用法，客户端应调用此 API 将批量句柄与其本地上下文绑定 因此，当服务器B收到服务器A转发的反序列化的批量句柄时，服务器B就可以知道客户端的原始地址来进行批量传输。 用户应注意，绑定批量句柄会增加序列化的额外开销，因此建议谨慎使用。 在源上绑定批量句柄时，应使用 crt_bulk_bind_transfer()，因为源地址信息嵌入在句柄中
    obj_req_fanout(obj, obj_auxi, dkey_hash, map_ver, epoch, shard_rw_prep, dc_obj_shard_rw, task) -> 扇出 shard_io_cb = io_cb = dc_obj_shard_rw

dc_obj_shard_rw 客户端对象分片读写(读写对象分片)
  dc_cont2uuid(shard->do_co, &cont_hdl_uuid, &cont_uuid) -> 设置容器uuid
  obj_shard_ptr2pool(shard) 根据分片获取池
  tgt_ep.ep_grp = pool->dp_sys->sy_group
  tgt_ep.ep_tag = shard->do_target_idx
  tgt_ep.ep_rank = shard->do_target_rank -> 设置cart端点(目标组,tag,rank)
  obj_req_create opc = DAOS_OBJ_RPC_UPDATE -> ds_obj_rw_handler
    crt_req_create(crt_ctx, tgt_ep, opcode, req)
  uuid_copy
  daos_dti_copy 拷贝dtx_id
  orw... > 填充容器读写结构体
  跳过ec逻辑
  tse_task_register_comp_cb(task, dc_rw_cb -> 注册容器读写任务回调
  daos_rpc_send
    crt_req_send(rpc, daos_rpc_cb, task) -> engine收到后处理 -> ds_obj_rw_handler -> tse_task_complete 发送完成回调流程:hg -> crt_hg_req_send_cb -> crp_complete_cb -> -> daos_rpc_cb -> dc_rw_cb

ds_obj_rw_handler(crt_rpc_t *rpc) -> 服务端对象读写处理器(DAOS_OBJ_RPC_UPDATE)
  obj_ioc_begin 访问VOS前的各种检查
  obj_rpc_is_fetch
  process_epoch
  obj_rpc_is_fetch
  rc = dtx_begin 返回超时?
    dtx_handle_init
      dtx_shares_init(dth) 初始化以下链表, 提交,中断,活动,检查
      dtx_epoch_bound
      vos_dtx_rsrvd_init(dth)
  obj_local_rw 本地读写

更新目标， 落盘
ds_obj_tgt_update_handler
    obj_ioc_begin 访问vos前的各种检查
        obj_ioc_begin_lite -> 设置lite IO上下文，到目前为止仅适用于复合RPC，1.还没有关联对象，2.权限检查（不确定它是读/写）
          obj_ioc_init -> 查找并返回容器句柄，如果是重建句柄，永远不会关联特定容器，则容器结构将返回给ioc::ioc_coc
            ds_cont_csummer_init(coc) -> 如果尚未加载，则按需加载 csummer 进行重建
          dss_rpc_cntr_enter(DSS_RC_OBJ) -> 增加 RPC 类型的活动计数器和总计数器
          tls = obj_tls_get()
          ioc->ioc_start_time = daos_get_ntime() -> IO开始时间
        obj_inflight_io_check
          如果传入 I/O 处于集成期间(integration)，则需要等待 vos 丢弃完成，否则可能会丢弃这些新的正在进行的 I/O 更新
          重建过程中的所有I/O，都需要等待重建栅栏生成（参见rebuild_prepare_one()），这将为重建创建一个边界，因此不应重建boundary(epoch)之后的数据，否则可能会被写入 重复，这可能会导致 VOS 失败
        obj_capa_check -> DAOS-7235 obj：当 cont_rf 损坏时关闭读/写权限，(#5822) 对于活动打开的容器句柄，DAOS 将在内部 - 1. 当 cont_rf 损坏时关闭读/写权限 2. 当 cont_rf 损坏时打开读/写权限 通过以下方式恢复 - 重新集成故障设备，或通过清除 UNCLEAN 容器状态 - “daos cont set-prop --properties=status:healthy ...” 将测试用例添加到 co_rf_simple()。 现在DAOS_PROP_CO_STATUS仅用于存储cont_create的pm_ver，以及当用户清除UNCLEAN状态时。 并且不要将 UNCLEAN 状态设置为 RDB 以避免在检查活动容器句柄时出现歧义
        obj_ioc_init_oca
    process_epoch -> 处理传入操作的纪元状态。 一旦该函数返回，纪元状态将包含选定的纪元。 另外，如果返回值为PE_OK_LOCAL，则该纪元可以毫无不确定性地用于本地RDG操作
    dtx_leader_begin
      dtx_handle_init
      vos_dtx_attach
    dtx_leader_exec_ops(dlh, obj_tgt_update, NULL, 0, &exec_arg) -> 在所有目标上执行对象更新操作
    dtx_leader_end(dlh, ioc.ioc_coh, rc)
      local_rc = func(dlh, func_arg, -1, NULL) -> obj_tgt_update -> 本地仅执行一次对象读写 -> obj_local_rw(exec_arg->rpc, exec_arg->ioc, &dlh->dlh_handle)
--------------------------
XXX：对于非单独DTX，领导者和非领导者将并行地进行各自的局部修改。 如果非领导者的速度太快，以至于非领导者可能已经开始处理下一个 RPC，但领导者还没有真正开始当前的修改，例如在批量数据传输阶段被阻止。 在这种情况下，有可能当non-leader处理下一个请求时，它命中了本地刚刚准备好的DTX，那么non-leader就会向leader检查这样的DTX状态。 但此时，Leader 上的 DTX 条目不存在，这会误导非 Leader 错过中止此类 DTX。 为了避免这种糟糕的情况，领导者需要在将当前请求分派给非领导者之前在 DRAM 中构建其 DTX 条目。 另一方面，即使对于单独的DTX，由于服务器端负载过重，PRC也可能被延迟处理。 如果是大数据传输的更新RPC，那么客户端有可能认为更新RPC超时，在原来的RPC批量数据传输期间重新发送RPC，这会导致CPU消耗，然后服务器上的重发逻辑将找不到相关的 DTX 条目，因为原始 RPC 的 DTX 尚未“准备好”。 在这种情况下，更新请求将在服务器上双重执行。 应该避免这种情况。 因此，在批量数据传输之前预分配 DTX 条目是必要的
obj_local_rw
  obj_get_iods_offs
  obj_local_rw_internal
    csum_verify_keys
    obj_singv_ec_rw_filter
    vos_update_begin
      vos_check_akeys
      vos_ioc_create
      vos_space_hold
      dkey_update_begin
    bio_iod_prep CRT_BULK_RW
      iod_map_iovs(biod, arg)
    bio_iod_copy(biod, orw->orw_sgls.ca_arrays, iods_nr)
    vos_dedup_verify
    obj_verify_bio_csum
    bio_iod_post_async(biod, rc)
    obj_bulk_transfer
    bio_iod_post
      写入时将数据从缓冲区转移到介质
      dma_rw
        nvme_rw
          xs_ctxt = biod->bd_ctxt->bic_xs_ctxt
          ...
          drain_inflight_ios(xs_ctxt, bxb)
            spdk_thread_poll(ctxt->bxc_thread, 0, 0)
          if (biod->bd_type == BIO_IOD_TYPE_UPDATE)
            spdk_blob_io_write(blob, channel, payload,
              page2io_unit(biod->bd_ctxt, pg_idx, BIO_DMA_PAGE_SZ),
              page2io_unit(biod->bd_ctxt, rw_cnt, BIO_DMA_PAGE_SZ),
              rw_completion, biod); -> 完成回调 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

obj_rw_complete(crt_rpc_t *rpc, struct obj_io_context *ioc,
    daos_handle_t ioh, int status, struct dtx_handle *dth)  -> 对象读写完成, 更新延迟计数器, 发送回复, 释放资源等
obj_update_latency
obj_rw_reply(rpc, rc, epoch.oe_value, &ioc)
  crt_reply_send(rpc)
obj_ioc_end(&ioc, rc)
  dss_rpc_cntr_exit
  obj_update_sensors
obj_ioc_fini

rw_completion(void *cb_arg, int err)
  biod = cb_arg -> 拿到BIO描述
  spdk_thread_send_msg bio_media_error -> 如果有错误,则通知错误处理线程
  iod_dma_completion -> 完成DMA
    biod->bd_completion(biod->bd_comp_arg, err) -> wal_completion | data_completion
    ...