int create_v2(IoCtx& io_ctx, const char *imgname, uint64_t bid, uint64_t size,
		int order, uint64_t features, uint64_t stripe_unit,
		uint64_t stripe_count)
  {
  	///(1)创建rbd_id.<image_name>对象
    id_obj = id_obj_name(imgname); ///rbd_id.<image_name>, object id

    int r = io_ctx.create(id_obj, true); ///create rbd_id.<image_name> object

    ///(2)将image id写到rbd_id.<image_name>
    extra = rand() % 0xFFFFFFFF;
    bid_ss << std::hex << bid << std::hex << extra;
    id = bid_ss.str();
    r = cls_client::set_id(&io_ctx, id_obj, id); ///rbd set_id

    ///(3)exec rbd dir_add_image
    r = cls_client::dir_add_image(&io_ctx, RBD_DIRECTORY, imgname, id); ///rbd dir_add_image

    ///(4)exec rbd create
    oss << RBD_DATA_PREFIX << id; ///"rbd_data."
    header_oid = header_name(id); ///rbd_header.<image_id>
    r = cls_client::create_image(&io_ctx, header_oid, size, order, ///rbd create
				 features, oss.str());

	///(5)exec rbd set_stripe_unit_count
    if ((stripe_unit || stripe_count) &&
	(stripe_count != 1 || stripe_unit != (1ull << order))) {
      r = cls_client::set_stripe_unit_count(&io_ctx, header_oid, ///rbd set_stripe_unit_count
					    stripe_unit, stripe_count);
}

///csl_rbd_client.cc
int create_image(librados::IoCtx *ioctx, const std::string &oid,
		 uint64_t size, uint8_t order, uint64_t features,
		 const std::string &object_prefix)
{
  bufferlist bl, bl2;
  ::encode(size, bl);
  ::encode(order, bl);
  ::encode(features, bl);
  ::encode(object_prefix, (bl));

  return ioctx->exec(oid, "rbd", "create", bl, bl2);
}

///cls_rbd.cc
/**
 * Initialize the header with basic metadata.
 * Extra features may initialize more fields in the future.
 * Everything is stored as key/value pairs as omaps in the header object.
 *
 * If features the OSD does not understand are requested, -ENOSYS is
 * returned.
 *
 * Input:
 * @param size number of bytes in the image (uint64_t)
 * @param order bits to shift to determine the size of data objects (uint8_t)
 * @param features what optional things this image will use (uint64_t)
 * @param object_prefix a prefix for all the data objects
 *
 * Output:
 * @return 0 on success, negative error code on failure
 */
int create(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  string object_prefix;
  uint64_t features, size;
  uint8_t order;

  try {
    bufferlist::iterator iter = in->begin();
    ::decode(size, iter);
    ::decode(order, iter);
    ::decode(features, iter);
    ::decode(object_prefix, iter);
  } catch (const buffer::error &err) {
    return -EINVAL;
  }

  CLS_LOG(20, "create object_prefix=%s size=%llu order=%u features=%llu",
	  object_prefix.c_str(), (unsigned long long)size, order,
	  (unsigned long long)features);

  if (features & ~RBD_FEATURES_ALL) {
    return -ENOSYS;
  }

  if (!object_prefix.size()) {
    return -EINVAL;
  }

  bufferlist stored_prefixbl;
  int r = cls_cxx_map_get_val(hctx, "object_prefix", &stored_prefixbl);
  if (r != -ENOENT) {
    CLS_ERR("reading object_prefix returned %d", r);
    return -EEXIST;
  }

  bufferlist sizebl;
  ::encode(size, sizebl);
  r = cls_cxx_map_set_val(hctx, "size", &sizebl);
  if (r < 0)
    return r;

  bufferlist orderbl;
  ::encode(order, orderbl);
  r = cls_cxx_map_set_val(hctx, "order", &orderbl);
  if (r < 0)
    return r;

  bufferlist featuresbl;
  ::encode(features, featuresbl);
  r = cls_cxx_map_set_val(hctx, "features", &featuresbl);
  if (r < 0)
    return r;

  bufferlist object_prefixbl;
  ::encode(object_prefix, object_prefixbl);
  r = cls_cxx_map_set_val(hctx, "object_prefix", &object_prefixbl);
  if (r < 0)
    return r;

  bufferlist snap_seqbl;
  uint64_t snap_seq = 0;
  ::encode(snap_seq, snap_seqbl);
  r = cls_cxx_map_set_val(hctx, "snap_seq", &snap_seqbl);
  if (r < 0)
    return r;

  return 0;
}

struct cls_rbd_snap {
  snapid_t id;
  string name;
  uint64_t image_size;
  uint64_t features;
  uint8_t protection_status;
  cls_rbd_parent parent;
  uint64_t flags;
}

int add_snap(ImageCtx *ictx, const char *snap_name)
{
uint64_t snap_id;
///(1)alloc snap_id
int r = ictx->md_ctx.selfmanaged_snap_create(&snap_id);
if (r < 0) {
  lderr(ictx->cct) << "failed to create snap id: " << cpp_strerror(-r)
		   << dendl;
  return r;
}

///(2)exec rbd snapshot_add
if (ictx->old_format) {
  r = cls_client::old_snapshot_add(&ictx->md_ctx, ictx->header_oid,
				   snap_id, snap_name);
} else {
  r = cls_client::snapshot_add(&ictx->md_ctx, ictx->header_oid,
			   snap_id, snap_name);
}
//...
return 0;
}

///cls_rbd_client.cc
int old_snapshot_add(librados::IoCtx *ioctx, const std::string &oid,
		 snapid_t snap_id, const std::string &snap_name)
{
  bufferlist bl, bl2;
  ::encode(snap_name, bl); ///snap name
  ::encode(snap_id, bl);   ///snap id

  return ioctx->exec(oid, "rbd", "snap_add", bl, bl2);
}

/**
 * Adds a snapshot to an rbd header. Ensures the id and name are unique.
 *
 * Input:
 * @param snap_name name of the snapshot (string)
 * @param snap_id id of the snapshot (uint64_t)
 *
 * Output:
 * @returns 0 on success, negative error code on failure.
 * @returns -ESTALE if the input snap_id is less than the image's snap_seq
 * @returns -EEXIST if the id or name are already used by another snapshot
 */
int snapshot_add(cls_method_context_t hctx, bufferlist *in, bufferlist *out)
{
  bufferlist snap_namebl, snap_idbl;
  cls_rbd_snap snap_meta;

  try {
    bufferlist::iterator iter = in->begin();
    ::decode(snap_meta.name, iter);
    ::decode(snap_meta.id, iter);
  } catch (const buffer::error &err) {
    return -EINVAL;
  }

  ///....
  bufferlist snap_metabl, snap_seqbl;
  ::encode(snap_meta, snap_metabl);
  ::encode(snap_meta.id, snap_seqbl); ///snapshot的 snap id

  string snapshot_key;
  key_from_snap_id(snap_meta.id, &snapshot_key);
  map<string, bufferlist> vals;
  vals["snap_seq"] = snap_seqbl; ///更新snap_seq field
  vals[snapshot_key] = snap_metabl; ///snapshot_$ID = struct cls_rbd_snap
  r = cls_cxx_map_set_vals(hctx, &vals);

}

#define IMAGE_BUF_SIZE 4194304
err = rados_ioctx_create(cluster, poolname, &io);
if (err < 0) {
    fprintf(stderr, "%s: cannot open rados pool %s: %s\n", argv[0], poolname, strerror(-err));
    rados_shutdown(cluster);
    exit(1);
}

err = rbd_open(io, "user1_image2", &image, NULL);
if (err < 0){
    fprintf(stderr, "open image failed: %s\n", strerror(-err));
    goto out;
}

err = rbd_read(image, IMAGE_BUF_SIZE, IMAGE_BUF_SIZE, buf);
if (err < 0) {
    fprintf(stderr, "%s: cannot read image: %s\n",  poolname, strerror(-err));
}else{
    fprintf(stderr, "read image return :%d\n", err);
}

class AioRequest
{   
  void complete(int r)
  {
    if (should_complete(r)) {
      if (m_hide_enoent && r == -ENOENT)
          r = 0;
      m_completion->complete(r);
      delete this;
    }
  }

void C_AioRead::finish(int r)
{
  ldout(m_cct, 10) << "C_AioRead::finish() " << this << " r = " << r << dendl;
  if (r >= 0 || r == -ENOENT) { // this was a sparse_read operation

      m_completion->destriper.add_partial_sparse_result(
    m_cct, m_req->data(), m_req->m_ext_map, m_req->m_object_off,
    m_req->m_buffer_extents);
}

void Striper::StripedReadResult::assemble_result(CephContext *cct, bufferlist& bl, bool zero_tail)
{
  ldout(cct, 10) << "assemble_result(" << this << ") zero_tail=" << zero_tail << dendl;

  // go backwards, so that we can efficiently discard zeros
  map<uint64_t,pair<bufferlist,uint64_t> >::reverse_iterator p = partial.rbegin();
  if (p == partial.rend())
    return;

  uint64_t end = p->first + p->second.second;
  while (p != partial.rend()) {
    // sanity check
    ldout(cct, 20) << "assemble_result(" << this << ") " << p->first << "~" << p->second.second
       << " " << p->second.first.length() << " bytes"
       << dendl;
    assert(p->first == end - p->second.second);
    end = p->first;

    size_t len = p->second.first.length(); ///return data len = 0
    if (len < p->second.second) {
      if (zero_tail || bl.length()) {
        bufferptr bp(p->second.second - len); ///intended len(4M) - data len(0)
        bp.zero();
        bl.push_front(bp); ///zero block
        bl.claim_prepend(p->second.first);
      }