// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include <utility>
#include <functional>

#include "crimson/common/log.h"

#include "crimson/os/seastore/object_data_handler.h"

namespace {
  seastar::logger& logger() {
    return crimson::get_logger(ceph_subsys_seastore_odata);
  }
}

SET_SUBSYS(seastore_odata);

namespace crimson::os::seastore {
#define assert_aligned(x) ceph_assert(((x)%ctx.tm.get_block_size()) == 0)

using context_t = ObjectDataHandler::context_t;
using get_iertr = ObjectDataHandler::write_iertr;

/**
 * extent_to_write_t
 *
 * Encapsulates smallest write operations in overwrite.
 * Indicates a zero/existing extent or a data extent based on whether
 * to_write is populate.
 * Should be handled by prepare_ops_list.
 */
struct extent_to_write_t {
  enum class type_t {
    DATA,
    ZERO,
    EXISTING,
  };
  type_t type;

  /// pin of original extent, not nullptr if type == EXISTING
  LBAMappingRef pin;

  laddr_t addr;
  extent_len_t len;

  /// non-nullopt if and only if type == DATA
  std::optional<bufferlist> to_write;

  extent_to_write_t(const extent_to_write_t &) = delete;
  extent_to_write_t(extent_to_write_t &&) = default;

  bool is_data() const {
    return type == type_t::DATA;
  }

  bool is_zero() const {
    return type == type_t::ZERO;
  }

  bool is_existing() const {
    return type == type_t::EXISTING;
  }

  laddr_t get_end_addr() const {
    return addr + len;
  }

  static extent_to_write_t create_data(
      laddr_t addr, bufferlist to_write) {
    return extent_to_write_t(addr, to_write);
  }

  static extent_to_write_t create_zero(
    laddr_t addr, extent_len_t len) {
    return extent_to_write_t(addr, len);
  }

  static extent_to_write_t create_existing(
    LBAMappingRef &&pin, laddr_t addr, extent_len_t len) {
    assert(pin);
    return extent_to_write_t(std::move(pin), addr, len);
  }

private:
  extent_to_write_t(laddr_t addr, bufferlist to_write)
    : type(type_t::DATA), addr(addr), len(to_write.length()),
      to_write(to_write) {}

  extent_to_write_t(laddr_t addr, extent_len_t len)
    : type(type_t::ZERO), addr(addr), len(len) {}

  extent_to_write_t(LBAMappingRef &&pin, laddr_t addr, extent_len_t len)
    : type(type_t::EXISTING), pin(std::move(pin)), addr(addr), len(len) {}
};
using extent_to_write_list_t = std::list<extent_to_write_t>;

// Encapsulates extents to be written out using do_remappings.
struct extent_to_remap_t {
  enum class type_t {
    REMAP,
    OVERWRITE
  };
  type_t type;
  /// pin of original extent
  LBAMappingRef pin;
  /// offset of remapped extent or overwrite part of overwrite extent.
  /// overwrite part of overwrite extent might correspond to mutiple 
  /// fresh write extent.
  extent_len_t new_offset;
  /// length of remapped extent or overwrite part of overwrite extent
  extent_len_t new_len;

  extent_to_remap_t(const extent_to_remap_t &) = delete;
  extent_to_remap_t(extent_to_remap_t &&) = default;

  bool is_remap() const {
    return type == type_t::REMAP;
  }

  bool is_overwrite() const {
    assert((new_offset != 0) && (pin->get_length() != new_offset + new_len));
    return type == type_t::OVERWRITE;
  }

  using remap_entry = TransactionManager::remap_entry;
  remap_entry create_remap_entry() {
    assert(is_remap());
    return remap_entry(
      new_offset,
      new_len);
  }

  remap_entry create_left_remap_entry() {
    assert(is_overwrite());
    return remap_entry(
      0,
      new_offset);
  }

  remap_entry create_right_remap_entry() {
    assert(is_overwrite());
    return remap_entry(
      new_offset + new_len,
      pin->get_length() - new_offset - new_len);
  }

  static extent_to_remap_t create_remap(
    LBAMappingRef &&pin, extent_len_t new_offset, extent_len_t new_len) {
    return extent_to_remap_t(type_t::REMAP,
      std::move(pin), new_offset, new_len);
  }

  static extent_to_remap_t create_overwrite(
    LBAMappingRef &&pin, extent_len_t new_offset, extent_len_t new_len) {
    return extent_to_remap_t(type_t::OVERWRITE,
      std::move(pin), new_offset, new_len);
  }

private:
  extent_to_remap_t(type_t type,
    LBAMappingRef &&pin, extent_len_t new_offset, extent_len_t new_len)
    : type(type),
      pin(std::move(pin)), new_offset(new_offset), new_len(new_len) {}
};
using extent_to_remap_list_t = std::list<extent_to_remap_t>;

// Encapsulates extents to be written out using do_insertions.
struct extent_to_insert_t {
  enum class type_t {
    DATA,
    ZERO
  };
  type_t type;
  /// laddr of new extent
  laddr_t addr;
  /// length of new extent
  extent_len_t len;
  /// non-nullopt if type == DATA
  std::optional<bufferlist> bl;

  extent_to_insert_t(const extent_to_insert_t &) = default;
  extent_to_insert_t(extent_to_insert_t &&) = default;

  bool is_data() const {
    return type == type_t::DATA;
  }

  bool is_zero() const {
    return type == type_t::ZERO;
  }

  static extent_to_insert_t create_data(
    laddr_t addr, extent_len_t len, std::optional<bufferlist> bl) {
    return extent_to_insert_t(addr, len, bl);
  }

  static extent_to_insert_t create_zero(
    laddr_t addr, extent_len_t len) {
    return extent_to_insert_t(addr, len);
  }

private:
  extent_to_insert_t(laddr_t addr, extent_len_t len,
    std::optional<bufferlist> bl)
    :type(type_t::DATA), addr(addr), len(len), bl(bl) {}

  extent_to_insert_t(laddr_t addr, extent_len_t len)
    :type(type_t::ZERO), addr(addr), len(len) {}
};
using extent_to_insert_list_t = std::list<extent_to_insert_t>;

// Encapsulates extents to be retired in do_removals.
using extent_to_remove_list_t = std::list<LBAMappingRef>;

struct overwrite_ops_t {
  extent_to_remap_list_t to_remap;
  extent_to_insert_list_t to_insert;
  extent_to_remove_list_t to_remove;
};

// prepare to_remap, to_retire, to_insert list
overwrite_ops_t prepare_ops_list(
  lba_pin_list_t &pins_to_remove,
  extent_to_write_list_t &to_write) {
  assert(pins_to_remove.size() != 0);
  overwrite_ops_t ops;
  ops.to_remove.swap(pins_to_remove);
  if (to_write.empty()) {
    logger().debug("empty to_write");
    return ops;
  }
  long unsigned int visitted = 0;
  auto& front = to_write.front();
  auto& back = to_write.back();

  // prepare overwrite, happens in one original extent.
  if (ops.to_remove.size() == 1 &&
      front.is_existing() && back.is_existing()) {
      visitted += 2;
      assert(to_write.size() > 2);
      assert(front.addr == front.pin->get_key());
      assert(back.addr > back.pin->get_key());
      ops.to_remap.push_back(extent_to_remap_t::create_overwrite(
	std::move(front.pin),
	front.len,
	back.addr - front.addr - front.len));
      ops.to_remove.pop_front();
  } else {
    // prepare to_remap, happens in one or multiple extents
    if (front.is_existing()) {
      visitted++;
      assert(to_write.size() > 1);
      assert(front.addr == front.pin->get_key());
      ops.to_remap.push_back(extent_to_remap_t::create_remap(
	std::move(front.pin),
	0,
	front.len));
      ops.to_remove.pop_front();
    }
    if (back.is_existing()) {
      visitted++;
      assert(to_write.size() > 1);
      assert(back.addr + back.len ==
	back.pin->get_key() + back.pin->get_length());
      ops.to_remap.push_back(extent_to_remap_t::create_remap(
	std::move(back.pin),
	back.addr - back.pin->get_key(),
	back.len));
      ops.to_remove.pop_back();
    }
  }

  // prepare to_insert
  for (auto &region : to_write) {
    if (region.is_data()) {
      visitted++;
      assert(region.to_write.has_value());
      ops.to_insert.push_back(extent_to_insert_t::create_data(
	region.addr, region.len, region.to_write));
    } else if (region.is_zero()) {
      visitted++;
      assert(!(region.to_write.has_value()));
      ops.to_insert.push_back(extent_to_insert_t::create_zero(
	region.addr, region.len));
    }
  }

  logger().debug(
    "to_remap list size: {}"
    " to_insert list size: {}"
    " to_remove list size: {}",
    ops.to_remap.size(), ops.to_insert.size(), ops.to_remove.size());
  assert(visitted == to_write.size());
  return ops;
}

/**
 * append_extent_to_write
 *
 * Appends passed extent_to_write_t maintaining invariant that the
 * list may not contain consecutive zero elements by checking and
 * combining them.
 */
void append_extent_to_write(
  extent_to_write_list_t &to_write, extent_to_write_t &&to_append)
{
  assert(to_write.empty() ||
         to_write.back().get_end_addr() == to_append.addr);
  if (to_write.empty() ||
      to_write.back().is_data() ||
      to_append.is_data() ||
      to_write.back().type != to_append.type) {
    to_write.push_back(std::move(to_append));
  } else {
    to_write.back().len += to_append.len;
  }
}

/**
 * splice_extent_to_write
 *
 * splices passed extent_to_write_list_t maintaining invariant that the
 * list may not contain consecutive zero elements by checking and
 * combining them.
 */
void splice_extent_to_write(
  extent_to_write_list_t &to_write, extent_to_write_list_t &&to_splice)
{
  if (!to_splice.empty()) {
    append_extent_to_write(to_write, std::move(to_splice.front()));
    to_splice.pop_front();
    to_write.splice(to_write.end(), std::move(to_splice));
  }
}

/// Creates remap extents in to_remap
ObjectDataHandler::write_ret do_remappings(
  context_t ctx,
  extent_to_remap_list_t &to_remap)
{
  return trans_intr::do_for_each(
    to_remap,
    [ctx](auto &region) {
      if (region.is_remap()) {
        return ctx.tm.remap_pin<ObjectDataBlock, 1>(
          ctx.t,
          std::move(region.pin),
          std::array{
            region.create_remap_entry()
          }
        ).si_then([&region](auto pins) {
          ceph_assert(pins.size() == 1);
          ceph_assert(region.new_len == pins[0]->get_length());
          return ObjectDataHandler::write_iertr::now();
        });
      } else if (region.is_overwrite()) {
        return ctx.tm.remap_pin<ObjectDataBlock, 2>(
          ctx.t,
          std::move(region.pin),
          std::array{
            region.create_left_remap_entry(),
            region.create_right_remap_entry()
          }
        ).si_then([&region](auto pins) {
          ceph_assert(pins.size() == 2);
          ceph_assert(region.pin->get_key() == pins[0]->get_key());
          ceph_assert(region.pin->get_key() + pins[0]->get_length() +
            region.new_len == pins[1]->get_key());
          return ObjectDataHandler::write_iertr::now();
        });
      } else {
        ceph_abort("impossible");
        return ObjectDataHandler::write_iertr::now();
      }
  });
}

ObjectDataHandler::write_ret do_removals(
  context_t ctx,
  lba_pin_list_t &to_remove)
{
  return trans_intr::do_for_each(
    to_remove,
    [ctx](auto &pin) {
      LOG_PREFIX(object_data_handler.cc::do_removals);
      DEBUGT("decreasing ref: {}",
	     ctx.t,
	     pin->get_key());
      return ctx.tm.dec_ref(
	ctx.t,
	pin->get_key()
      ).si_then(
	[](auto){},
	ObjectDataHandler::write_iertr::pass_further{},
	crimson::ct_error::assert_all{
	  "object_data_handler::do_removals invalid error"
	}
      );
    });
}

/// Creates zero/data extents in to_insert
ObjectDataHandler::write_ret do_insertions(
  context_t ctx,
  extent_to_insert_list_t &to_insert)
{
  return trans_intr::do_for_each(
    to_insert,
    [ctx](auto &region) {
      LOG_PREFIX(object_data_handler.cc::do_insertions);
      if (region.is_data()) {
	assert_aligned(region.addr);
	assert_aligned(region.len);
	ceph_assert(region.len == region.bl->length());
	DEBUGT("allocating extent: {}~{}",
	       ctx.t,
	       region.addr,
	       region.len);
	return ctx.tm.alloc_extent<ObjectDataBlock>(
	  ctx.t,
	  region.addr,
	  region.len
	).si_then([&region](auto extent) {
	  if (extent->get_laddr() != region.addr) {
	    logger().debug(
	      "object_data_handler::do_insertions alloc got addr {},"
	      " should have been {}",
	      extent->get_laddr(),
	      region.addr);
	  }
	  ceph_assert(extent->get_laddr() == region.addr);
	  ceph_assert(extent->get_length() == region.len);
	  auto iter = region.bl->cbegin();
	  iter.copy(region.len, extent->get_bptr().c_str());
	  return ObjectDataHandler::write_iertr::now();
	});
      } else if (region.is_zero()) {
	DEBUGT("reserving: {}~{}",
	       ctx.t,
	       region.addr,
	       region.len);
	return ctx.tm.reserve_region(
	  ctx.t,
	  region.addr,
	  region.len
	).si_then([FNAME, ctx, &region](auto pin) {
	  ceph_assert(pin->get_length() == region.len);
	  if (pin->get_key() != region.addr) {
	    ERRORT(
	      "inconsistent laddr: pin: {} region {}",
	      ctx.t,
	      pin->get_key(),
	      region.addr);
	  }
	  ceph_assert(pin->get_key() == region.addr);
	  return ObjectDataHandler::write_iertr::now();
	});
      } else {
	ceph_abort("impossible");
	return ObjectDataHandler::write_iertr::now();
      }
    });
}

enum class overwrite_operation_t {
  UNKNOWN,
  OVERWRITE_ZERO,           // fill unaligned data with zero
  MERGE_EXISTING,           // if present, merge data with the clean/pending extent
  SPLIT_EXISTING,           // split the existing extent, and fill unaligned data
};

std::ostream& operator<<(
  std::ostream &out,
  const overwrite_operation_t &operation)
{
  switch (operation) {
  case overwrite_operation_t::UNKNOWN:
    return out << "UNKNOWN";
  case overwrite_operation_t::OVERWRITE_ZERO:
    return out << "OVERWRITE_ZERO";
  case overwrite_operation_t::MERGE_EXISTING:
    return out << "MERGE_EXISTING";
  case overwrite_operation_t::SPLIT_EXISTING:
    return out << "SPLIT_EXISTING";
  default:
    return out << "!IMPOSSIBLE_OPERATION";
  }
}

/**
 * overwrite_plan_t
 *
 * |<--------------------------pins_size---------------------------------------------->|
 * pin_begin(aligned)                                                   pin_end(aligned)
 *                 |<------aligned_data_size-------------------------->| (aligned-bl)
 *                 aligned_data_begin                   aligned_data_end
 *                                    |<-data_size->| (bl)
 *                                    data_begin  end
 *             left(l)                                            right(r)
 * |<l_extent_size>|<l_alignment_size>|             |<r_alignment_size>|<r_extent_size>|
 * |<-----------left_size------------>|             |<-----------right_size----------->|
 *
 * |<-----(existing left extent/pin)----->|    |<-----(existing right extent/pin)----->|
 * left_paddr                                  right_paddr
 */
struct overwrite_plan_t {
  // addresses
  laddr_t pin_begin;
  laddr_t pin_end;
  paddr_t left_paddr;
  paddr_t right_paddr;
  laddr_t data_begin;
  laddr_t data_end;
  laddr_t aligned_data_begin;
  laddr_t aligned_data_end;

  // operations
  overwrite_operation_t left_operation;
  overwrite_operation_t right_operation;

  // helper member
  extent_len_t block_size;

public:
  extent_len_t get_left_size() const {
    return data_begin - pin_begin;
  }

  extent_len_t get_left_extent_size() const {
    return aligned_data_begin - pin_begin;
  }

  extent_len_t get_left_alignment_size() const {
    return data_begin - aligned_data_begin;
  }

  extent_len_t get_right_size() const {
    return pin_end - data_end;
  }

  extent_len_t get_right_extent_size() const {
    return pin_end - aligned_data_end;
  }

  extent_len_t get_right_alignment_size() const {
    return aligned_data_end - data_end;
  }

  extent_len_t get_aligned_data_size() const {
    return aligned_data_end - aligned_data_begin;
  }

  extent_len_t get_pins_size() const {
    return pin_end - pin_begin;
  }

  friend std::ostream& operator<<(
    std::ostream& out,
    const overwrite_plan_t& overwrite_plan) {
    return out << "overwrite_plan_t("
	       << "pin_begin=" << overwrite_plan.pin_begin
	       << ", pin_end=" << overwrite_plan.pin_end
	       << ", left_paddr=" << overwrite_plan.left_paddr
	       << ", right_paddr=" << overwrite_plan.right_paddr
	       << ", data_begin=" << overwrite_plan.data_begin
	       << ", data_end=" << overwrite_plan.data_end
	       << ", aligned_data_begin=" << overwrite_plan.aligned_data_begin
	       << ", aligned_data_end=" << overwrite_plan.aligned_data_end
	       << ", left_operation=" << overwrite_plan.left_operation
	       << ", right_operation=" << overwrite_plan.right_operation
	       << ", block_size=" << overwrite_plan.block_size
	       << ")";
  }

  overwrite_plan_t(laddr_t offset,
		   extent_len_t len,
		   const lba_pin_list_t& pins,
		   extent_len_t block_size,
		   Transaction& t) :
      pin_begin(pins.front()->get_key()),
      pin_end(pins.back()->get_key() + pins.back()->get_length()),
      left_paddr(pins.front()->get_val()),
      right_paddr(pins.back()->get_val()),
      data_begin(offset),
      data_end(offset + len),
      aligned_data_begin(p2align((uint64_t)data_begin, (uint64_t)block_size)),
      aligned_data_end(p2roundup((uint64_t)data_end, (uint64_t)block_size)),
      left_operation(overwrite_operation_t::UNKNOWN),
      right_operation(overwrite_operation_t::UNKNOWN),
      block_size(block_size) {
    validate();
    evaluate_operations(t);
    assert(left_operation != overwrite_operation_t::UNKNOWN);
    assert(right_operation != overwrite_operation_t::UNKNOWN);
  }

private:
  // refer to overwrite_plan_t description
  void validate() const {
    ceph_assert(pin_begin % block_size == 0);
    ceph_assert(pin_end % block_size == 0);
    ceph_assert(aligned_data_begin % block_size == 0);
    ceph_assert(aligned_data_end % block_size == 0);

    ceph_assert(pin_begin <= aligned_data_begin);
    ceph_assert(aligned_data_begin <= data_begin);
    ceph_assert(data_begin <= data_end);
    ceph_assert(data_end <= aligned_data_end);
    ceph_assert(aligned_data_end <= pin_end);
  }

  /*
   * When trying to modify a portion of an object data block, follow
   * the read-full-extent-then-merge-new-data strategy, if the write
   * amplification caused by it is not greater than
   * seastore_obj_data_write_amplification; otherwise, split the
   * original extent into at most three parts: origin-left, part-to-be-modified
   * and origin-right.
   */
  void evaluate_operations(Transaction& t) {
    auto actual_write_size = get_pins_size();
    auto aligned_data_size = get_aligned_data_size();
    auto left_ext_size = get_left_extent_size();
    auto right_ext_size = get_right_extent_size();

    auto can_merge = [](Transaction& t, paddr_t paddr) {
      CachedExtentRef ext;
      if (paddr.is_relative() || paddr.is_delayed()) {
	  return true;
      } else if (t.get_extent(paddr, &ext) ==
	Transaction::get_extent_ret::PRESENT) {
	// FIXME: there is no need to lookup the cache if the pin can 
	// be associated with the extent state
	if (ext->is_mutable()) {
	  return true;
	}
      }
      return false;
    };
    if (left_paddr.is_zero()) {
      actual_write_size -= left_ext_size;
      left_ext_size = 0;
      left_operation = overwrite_operation_t::OVERWRITE_ZERO;
    } else if (can_merge(t, left_paddr)) {
      aligned_data_size += left_ext_size;
      left_ext_size = 0;
      left_operation = overwrite_operation_t::MERGE_EXISTING;
    }

    if (right_paddr.is_zero()) {
      actual_write_size -= right_ext_size;
      right_ext_size = 0;
      right_operation = overwrite_operation_t::OVERWRITE_ZERO;
    } else if (can_merge(t, right_paddr)) {
      aligned_data_size += right_ext_size;
      right_ext_size = 0;
      right_operation = overwrite_operation_t::MERGE_EXISTING;
    }

    while (left_operation == overwrite_operation_t::UNKNOWN ||
           right_operation == overwrite_operation_t::UNKNOWN) {
      if (((double)actual_write_size / (double)aligned_data_size) <=
          crimson::common::get_conf<double>("seastore_obj_data_write_amplification")) {
        break;
      }
      if (left_ext_size == 0 && right_ext_size == 0) {
        break;
      }
      if (left_ext_size >= right_ext_size) {
        // split left
        assert(left_operation == overwrite_operation_t::UNKNOWN);
        actual_write_size -= left_ext_size;
        left_ext_size = 0;
        left_operation = overwrite_operation_t::SPLIT_EXISTING;
      } else { // left_ext_size < right_ext_size
        // split right
        assert(right_operation == overwrite_operation_t::UNKNOWN);
        actual_write_size -= right_ext_size;
        right_ext_size = 0;
        right_operation = overwrite_operation_t::SPLIT_EXISTING;
      }
    }

    if (left_operation == overwrite_operation_t::UNKNOWN) {
      // no split left, so merge with left
      left_operation = overwrite_operation_t::MERGE_EXISTING;
    }

    if (right_operation == overwrite_operation_t::UNKNOWN) {
      // no split right, so merge with right
      right_operation = overwrite_operation_t::MERGE_EXISTING;
    }
  }
};

} // namespace crimson::os::seastore

#if FMT_VERSION >= 90000
template<> struct fmt::formatter<crimson::os::seastore::overwrite_plan_t> : fmt::ostream_formatter {};
#endif

namespace crimson::os::seastore {

/**
 * operate_left
 *
 * Proceed overwrite_plan.left_operation.
 */
using operate_ret_bare = std::pair<
  std::optional<extent_to_write_t>,
  std::optional<bufferptr>>;
using operate_ret = get_iertr::future<operate_ret_bare>;
operate_ret operate_left(context_t ctx, LBAMappingRef &pin, const overwrite_plan_t &overwrite_plan)
{
  if (overwrite_plan.get_left_size() == 0) {
    return get_iertr::make_ready_future<operate_ret_bare>(
      std::nullopt,
      std::nullopt);
  }

  if (overwrite_plan.left_operation == overwrite_operation_t::OVERWRITE_ZERO) {
    assert(pin->get_val().is_zero());
    auto zero_extent_len = overwrite_plan.get_left_extent_size();
    assert_aligned(zero_extent_len);
    auto zero_prepend_len = overwrite_plan.get_left_alignment_size();
    return get_iertr::make_ready_future<operate_ret_bare>(
      (zero_extent_len == 0
       ? std::nullopt
       : std::make_optional(extent_to_write_t::create_zero(
           overwrite_plan.pin_begin, zero_extent_len))),
      (zero_prepend_len == 0
       ? std::nullopt
       : std::make_optional(bufferptr(
           ceph::buffer::create(zero_prepend_len, 0))))
    );
  } else if (overwrite_plan.left_operation == overwrite_operation_t::MERGE_EXISTING) {
    auto prepend_len = overwrite_plan.get_left_size();
    if (prepend_len == 0) {
      return get_iertr::make_ready_future<operate_ret_bare>(
        std::nullopt,
        std::nullopt);
    } else {
      extent_len_t off = pin->get_intermediate_offset();
      return ctx.tm.read_pin<ObjectDataBlock>(
	ctx.t, pin->duplicate()
      ).si_then([prepend_len, off](auto left_extent) {
        return get_iertr::make_ready_future<operate_ret_bare>(
          std::nullopt,
          std::make_optional(bufferptr(
            left_extent->get_bptr(),
            off,
            prepend_len)));
      });
    }
  } else {
    assert(overwrite_plan.left_operation == overwrite_operation_t::SPLIT_EXISTING);

    auto extent_len = overwrite_plan.get_left_extent_size();
    assert(extent_len);
    std::optional<extent_to_write_t> left_to_write_extent =
      std::make_optional(extent_to_write_t::create_existing(
        pin->duplicate(),
        pin->get_key(),
        extent_len));

    auto prepend_len = overwrite_plan.get_left_alignment_size();
    if (prepend_len == 0) {
      return get_iertr::make_ready_future<operate_ret_bare>(
        std::move(left_to_write_extent),
        std::nullopt);
    } else {
      extent_len_t off = pin->get_intermediate_offset();
      return ctx.tm.read_pin<ObjectDataBlock>(
	ctx.t, pin->duplicate()
      ).si_then([prepend_offset=extent_len + off, prepend_len,
                 left_to_write_extent=std::move(left_to_write_extent)]
                (auto left_extent) mutable {
        return get_iertr::make_ready_future<operate_ret_bare>(
          std::move(left_to_write_extent),
          std::make_optional(bufferptr(
            left_extent->get_bptr(),
            prepend_offset,
            prepend_len)));
      });
    }
  }
};

/**
 * operate_right
 *
 * Proceed overwrite_plan.right_operation.
 */
operate_ret operate_right(context_t ctx, LBAMappingRef &pin, const overwrite_plan_t &overwrite_plan)
{
  if (overwrite_plan.get_right_size() == 0) {
    return get_iertr::make_ready_future<operate_ret_bare>(
      std::nullopt,
      std::nullopt);
  }

  auto right_pin_begin = pin->get_key();
  assert(overwrite_plan.data_end >= right_pin_begin);
  if (overwrite_plan.right_operation == overwrite_operation_t::OVERWRITE_ZERO) {
    assert(pin->get_val().is_zero());
    auto zero_suffix_len = overwrite_plan.get_right_alignment_size();
    auto zero_extent_len = overwrite_plan.get_right_extent_size();
    assert_aligned(zero_extent_len);
    return get_iertr::make_ready_future<operate_ret_bare>(
      (zero_extent_len == 0
       ? std::nullopt
       : std::make_optional(extent_to_write_t::create_zero(
           overwrite_plan.aligned_data_end, zero_extent_len))),
      (zero_suffix_len == 0
       ? std::nullopt
       : std::make_optional(bufferptr(
           ceph::buffer::create(zero_suffix_len, 0))))
    );
  } else if (overwrite_plan.right_operation == overwrite_operation_t::MERGE_EXISTING) {
    auto append_len = overwrite_plan.get_right_size();
    if (append_len == 0) {
      return get_iertr::make_ready_future<operate_ret_bare>(
        std::nullopt,
        std::nullopt);
    } else {
      auto append_offset =
	overwrite_plan.data_end
	- right_pin_begin
	+ pin->get_intermediate_offset();
      return ctx.tm.read_pin<ObjectDataBlock>(
	ctx.t, pin->duplicate()
      ).si_then([append_offset, append_len](auto right_extent) {
        return get_iertr::make_ready_future<operate_ret_bare>(
          std::nullopt,
          std::make_optional(bufferptr(
            right_extent->get_bptr(),
            append_offset,
            append_len)));
      });
    }
  } else {
    assert(overwrite_plan.right_operation == overwrite_operation_t::SPLIT_EXISTING);

    auto extent_len = overwrite_plan.get_right_extent_size();
    assert(extent_len);
    std::optional<extent_to_write_t> right_to_write_extent =
      std::make_optional(extent_to_write_t::create_existing(
        pin->duplicate(),
        overwrite_plan.aligned_data_end,
        extent_len));

    auto append_len = overwrite_plan.get_right_alignment_size();
    if (append_len == 0) {
      return get_iertr::make_ready_future<operate_ret_bare>(
        std::move(right_to_write_extent),
        std::nullopt);
    } else {
      auto append_offset =
	overwrite_plan.data_end
	- right_pin_begin
	+ pin->get_intermediate_offset();
      return ctx.tm.read_pin<ObjectDataBlock>(
	ctx.t, pin->duplicate()
      ).si_then([append_offset, append_len,
                 right_to_write_extent=std::move(right_to_write_extent)]
                (auto right_extent) mutable {
        return get_iertr::make_ready_future<operate_ret_bare>(
          std::move(right_to_write_extent),
          std::make_optional(bufferptr(
            right_extent->get_bptr(),
            append_offset,
            append_len)));
      });
    }
  }
};

template <typename F>
auto with_object_data(
  ObjectDataHandler::context_t ctx,
  F &&f)
{
  return seastar::do_with(
    ctx.onode.get_layout().object_data.get(),
    std::forward<F>(f),
    [ctx](auto &object_data, auto &f) {
      return std::invoke(f, object_data
      ).si_then([ctx, &object_data] {
	if (object_data.must_update()) {
	  ctx.onode.get_mutable_layout(ctx.t).object_data.update(object_data);
	}
	return seastar::now();
      });
    });
}

template <typename F>
auto with_objects_data(
  ObjectDataHandler::context_t ctx,
  F &&f)
{
  ceph_assert(ctx.d_onode);
  return seastar::do_with(
    ctx.onode.get_layout().object_data.get(),
    ctx.d_onode->get_layout().object_data.get(),
    std::forward<F>(f),
    [ctx](auto &object_data, auto &d_object_data, auto &f) {
      return std::invoke(f, object_data, d_object_data
      ).si_then([ctx, &object_data, &d_object_data] {
	if (object_data.must_update()) {
	  ctx.onode.get_mutable_layout(ctx.t).object_data.update(object_data);
	}
	if (d_object_data.must_update()) {
	  ctx.d_onode->get_mutable_layout(
	    ctx.t).object_data.update(d_object_data);
	}
	return seastar::now();
      });
    });
}

ObjectDataHandler::write_ret ObjectDataHandler::prepare_data_reservation(
  context_t ctx,
  object_data_t &object_data,
  extent_len_t size)
{
  LOG_PREFIX(ObjectDataHandler::prepare_data_reservation);
  ceph_assert(size <= max_object_size);
  if (!object_data.is_null()) {
    ceph_assert(object_data.get_reserved_data_len() == max_object_size);
    DEBUGT("reservation present: {}~{}",
           ctx.t,
           object_data.get_reserved_data_base(),
           object_data.get_reserved_data_len());
    return write_iertr::now();
  } else {
    DEBUGT("reserving: {}~{}",
           ctx.t,
           ctx.onode.get_data_hint(),
           max_object_size);
    return ctx.tm.reserve_region(
      ctx.t,
      ctx.onode.get_data_hint(),
      max_object_size
    ).si_then([max_object_size=max_object_size, &object_data](auto pin) {
      ceph_assert(pin->get_length() == max_object_size);
      object_data.update_reserved(
	pin->get_key(),
	pin->get_length());
      return write_iertr::now();
    });
  }
}

ObjectDataHandler::clear_ret ObjectDataHandler::trim_data_reservation(
  context_t ctx, object_data_t &object_data, extent_len_t size)
{
  ceph_assert(!object_data.is_null());
  ceph_assert(size <= object_data.get_reserved_data_len());
  return seastar::do_with(
    lba_pin_list_t(),
    extent_to_write_list_t(),
    [ctx, size, &object_data](auto &pins, auto &to_write) {
      LOG_PREFIX(ObjectDataHandler::trim_data_reservation);
      DEBUGT("object_data: {}~{}",
	     ctx.t,
	     object_data.get_reserved_data_base(),
	     object_data.get_reserved_data_len());
      return ctx.tm.get_pins(
	ctx.t,
	object_data.get_reserved_data_base() + size,
	object_data.get_reserved_data_len() - size
      ).si_then([ctx, size, &pins, &object_data, &to_write](auto _pins) {
	_pins.swap(pins);
	ceph_assert(pins.size());
	if (!size) {
	  // no need to reserve region if we are truncating the object's
	  // size to 0
	  return clear_iertr::now();
	}
	auto &pin = *pins.front();
	ceph_assert(pin.get_key() >= object_data.get_reserved_data_base());
	ceph_assert(
	  pin.get_key() <= object_data.get_reserved_data_base() + size);
	auto pin_offset = pin.get_key() -
	  object_data.get_reserved_data_base();
	if ((pin.get_key() == (object_data.get_reserved_data_base() + size)) ||
	  (pin.get_val().is_zero())) {
	  /* First pin is exactly at the boundary or is a zero pin.  Either way,
	   * remove all pins and add a single zero pin to the end. */
	  to_write.push_back(extent_to_write_t::create_zero(
	    pin.get_key(),
	    object_data.get_reserved_data_len() - pin_offset));
	  return clear_iertr::now();
	} else {
	  /* First pin overlaps the boundary and has data, remap it
	   * if aligned or rewrite it if not aligned to size */
          auto roundup_size = p2roundup(size, ctx.tm.get_block_size());
          auto append_len = roundup_size - size;
          if (append_len == 0) {
            LOG_PREFIX(ObjectDataHandler::trim_data_reservation);
            TRACET("First pin overlaps the boundary and has aligned data"
              "create existing at addr:{}, len:{}",
              ctx.t, pin.get_key(), size - pin_offset);
            to_write.push_back(extent_to_write_t::create_existing(
              pin.duplicate(),
              pin.get_key(),
              size - pin_offset));
	    to_write.push_back(extent_to_write_t::create_zero(
	      object_data.get_reserved_data_base() + roundup_size,
	      object_data.get_reserved_data_len() - roundup_size));
            return clear_iertr::now();
          } else {
            return ctx.tm.read_pin<ObjectDataBlock>(
              ctx.t,
              pin.duplicate()
            ).si_then([ctx, size, pin_offset, append_len, roundup_size,
                      &pin, &object_data, &to_write](auto extent) {
              bufferlist bl;
	      bl.append(
	        bufferptr(
	          extent->get_bptr(),
		  pin.get_intermediate_offset(),
	          size - pin_offset
	      ));
              bl.append_zero(append_len);
              LOG_PREFIX(ObjectDataHandler::trim_data_reservation);
              TRACET("First pin overlaps the boundary and has unaligned data"
                "create data at addr:{}, len:{}",
                ctx.t, pin.get_key(), bl.length());
	      to_write.push_back(extent_to_write_t::create_data(
	        pin.get_key(),
	        bl));
	      to_write.push_back(extent_to_write_t::create_zero(
	        object_data.get_reserved_data_base() + roundup_size,
	        object_data.get_reserved_data_len() - roundup_size));
              return clear_iertr::now();
            });
          }
	}
      }).si_then([ctx, size, &to_write, &object_data, &pins] {
        return seastar::do_with(
          prepare_ops_list(pins, to_write),
          [ctx, size, &object_data](auto &ops) {
            return do_remappings(ctx, ops.to_remap
            ).si_then([ctx, &ops] {
              return do_removals(ctx, ops.to_remove);
            }).si_then([ctx, &ops] {
              return do_insertions(ctx, ops.to_insert);
            }).si_then([size, &object_data] {
	      if (size == 0) {
	        object_data.clear();
	      }
	      return ObjectDataHandler::clear_iertr::now();
            });
        });
      });
    });
}

/**
 * get_to_writes_with_zero_buffer
 *
 * Returns extent_to_write_t's reflecting a zero region extending
 * from offset~len with headptr optionally on the left and tailptr
 * optionally on the right.
 */
extent_to_write_list_t get_to_writes_with_zero_buffer(
  const extent_len_t block_size,
  laddr_t offset, extent_len_t len,
  std::optional<bufferptr> &&headptr, std::optional<bufferptr> &&tailptr)
{
  auto zero_left = p2roundup(offset, (laddr_t)block_size);
  auto zero_right = p2align(offset + len, (laddr_t)block_size);
  auto left = headptr ? (offset - headptr->length()) : offset;
  auto right = tailptr ?
    (offset + len + tailptr->length()) :
    (offset + len);

  assert(
    (headptr && ((zero_left - left) ==
		 p2roundup(headptr->length(), block_size))) ^
    (!headptr && (zero_left == left)));
  assert(
    (tailptr && ((right - zero_right) ==
		 p2roundup(tailptr->length(), block_size))) ^
    (!tailptr && (right == zero_right)));

  assert(right > left);
  assert((left % block_size) == 0);
  assert((right % block_size) == 0);

  // zero region too small for a reserved section,
  // headptr and tailptr in same extent
  if (zero_right <= zero_left) {
    bufferlist bl;
    if (headptr) {
      bl.append(*headptr);
    }
    bl.append_zero(
      right - left - bl.length() - (tailptr ? tailptr->length() : 0));
    if (tailptr) {
      bl.append(*tailptr);
    }
    assert(bl.length() % block_size == 0);
    assert(bl.length() == (right - left));
    extent_to_write_list_t ret;
    ret.push_back(extent_to_write_t::create_data(left, bl));
    return ret;
  } else {
    // reserved section between ends, headptr and tailptr in different extents
    extent_to_write_list_t ret;
    if (headptr) {
      bufferlist headbl;
      headbl.append(*headptr);
      headbl.append_zero(zero_left - left - headbl.length());
      assert(headbl.length() % block_size == 0);
      assert(headbl.length() > 0);
      ret.push_back(extent_to_write_t::create_data(left, headbl));
    }
    // reserved zero region
    ret.push_back(extent_to_write_t::create_zero(zero_left, zero_right - zero_left));
    assert(ret.back().len % block_size == 0);
    assert(ret.back().len > 0);
    if (tailptr) {
      bufferlist tailbl;
      tailbl.append(*tailptr);
      tailbl.append_zero(right - zero_right - tailbl.length());
      assert(tailbl.length() % block_size == 0);
      assert(tailbl.length() > 0);
      ret.push_back(extent_to_write_t::create_data(zero_right, tailbl));
    }
    return ret;
  }
}

/**
 * get_to_writes
 *
 * Returns extent_to_write_t's from bl.
 *
 * TODO: probably add some kind of upper limit on extent size.
 */
extent_to_write_list_t get_to_writes(laddr_t offset, bufferlist &bl)
{
  auto ret = extent_to_write_list_t();
  ret.push_back(extent_to_write_t::create_data(offset, bl));
  return ret;
};

ObjectDataHandler::write_ret ObjectDataHandler::overwrite(
  context_t ctx,
  laddr_t offset,
  extent_len_t len,
  std::optional<bufferlist> &&bl,
  lba_pin_list_t &&_pins)
{
  if (bl.has_value()) {
    assert(bl->length() == len);
  }
  overwrite_plan_t overwrite_plan(offset, len, _pins, ctx.tm.get_block_size(), ctx.t);
  return seastar::do_with(
    std::move(_pins),
    extent_to_write_list_t(),
    [ctx, len, offset, overwrite_plan, bl=std::move(bl)]
    (auto &pins, auto &to_write) mutable
  {
    LOG_PREFIX(ObjectDataHandler::overwrite);
    DEBUGT("overwrite: {}~{}",
           ctx.t,
           offset,
           len);
    ceph_assert(pins.size() >= 1);
    DEBUGT("overwrite: split overwrite_plan {}", ctx.t, overwrite_plan);

    return operate_left(
      ctx,
      pins.front(),
      overwrite_plan
    ).si_then([ctx, len, offset, overwrite_plan, bl=std::move(bl),
               &to_write, &pins](auto p) mutable {
      auto &[left_extent, headptr] = p;
      if (left_extent) {
        ceph_assert(left_extent->addr == overwrite_plan.pin_begin);
        append_extent_to_write(to_write, std::move(*left_extent));
      }
      if (headptr) {
        assert(headptr->length() > 0);
      }
      return operate_right(
        ctx,
        pins.back(),
        overwrite_plan
      ).si_then([ctx, len, offset,
                 pin_begin=overwrite_plan.pin_begin,
                 pin_end=overwrite_plan.pin_end,
                 bl=std::move(bl), headptr=std::move(headptr),
                 &to_write, &pins](auto p) mutable {
        auto &[right_extent, tailptr] = p;
        if (bl.has_value()) {
          auto write_offset = offset;
          bufferlist write_bl;
          if (headptr) {
            write_bl.append(*headptr);
            write_offset -= headptr->length();
            assert_aligned(write_offset);
          }
          write_bl.claim_append(*bl);
          if (tailptr) {
            write_bl.append(*tailptr);
            assert_aligned(write_bl.length());
          }
          splice_extent_to_write(
            to_write,
            get_to_writes(write_offset, write_bl));
        } else {
          splice_extent_to_write(
            to_write,
            get_to_writes_with_zero_buffer(
              ctx.tm.get_block_size(),
              offset,
              len,
              std::move(headptr),
              std::move(tailptr)));
        }
        if (right_extent) {
          ceph_assert(right_extent->get_end_addr() == pin_end);
          append_extent_to_write(to_write, std::move(*right_extent));
        }
        assert(to_write.size());
        assert(pin_begin == to_write.front().addr);
        assert(pin_end == to_write.back().get_end_addr());

        return seastar::do_with(
          prepare_ops_list(pins, to_write),
          [ctx](auto &ops) {
            return do_remappings(ctx, ops.to_remap
            ).si_then([ctx, &ops] {
              return do_removals(ctx, ops.to_remove);
            }).si_then([ctx, &ops] {
              return do_insertions(ctx, ops.to_insert);
            });
        });
      });
    });
  });
}

ObjectDataHandler::zero_ret ObjectDataHandler::zero(
  context_t ctx,
  objaddr_t offset,
  extent_len_t len)
{
  return with_object_data(
    ctx,
    [this, ctx, offset, len](auto &object_data) {
      LOG_PREFIX(ObjectDataHandler::zero);
      DEBUGT("zero to {}~{}, object_data: {}~{}, is_null {}",
             ctx.t,
             offset,
             len,
             object_data.get_reserved_data_base(),
             object_data.get_reserved_data_len(),
             object_data.is_null());
      return prepare_data_reservation(
	ctx,
	object_data,
	p2roundup(offset + len, ctx.tm.get_block_size())
      ).si_then([this, ctx, offset, len, &object_data] {
	auto logical_offset = object_data.get_reserved_data_base() + offset;
	return ctx.tm.get_pins(
	  ctx.t,
	  logical_offset,
	  len
	).si_then([this, ctx, logical_offset, len](auto pins) {
	  return overwrite(
	    ctx, logical_offset, len,
	    std::nullopt, std::move(pins));
	});
      });
    });
}

ObjectDataHandler::write_ret ObjectDataHandler::write(
  context_t ctx,
  objaddr_t offset,
  const bufferlist &bl)
{
  return with_object_data(
    ctx,
    [this, ctx, offset, &bl](auto &object_data) {
      LOG_PREFIX(ObjectDataHandler::write);
      DEBUGT("writing to {}~{}, object_data: {}~{}, is_null {}",
             ctx.t,
             offset,
	     bl.length(),
	     object_data.get_reserved_data_base(),
	     object_data.get_reserved_data_len(),
             object_data.is_null());
      return prepare_data_reservation(
	ctx,
	object_data,
	p2roundup(offset + bl.length(), ctx.tm.get_block_size())
      ).si_then([this, ctx, offset, &object_data, &bl] {
	auto logical_offset = object_data.get_reserved_data_base() + offset;
	return ctx.tm.get_pins(
	  ctx.t,
	  logical_offset,
	  bl.length()
	).si_then([this, ctx,logical_offset, &bl](
		   auto pins) {
	  return overwrite(
	    ctx, logical_offset, bl.length(),
	    bufferlist(bl), std::move(pins));
	});
      });
    });
}

ObjectDataHandler::read_ret ObjectDataHandler::read(
  context_t ctx,
  objaddr_t obj_offset,
  extent_len_t len)
{
  return seastar::do_with(
    bufferlist(),
    [ctx, obj_offset, len](auto &ret) {
      return with_object_data(
	ctx,
	[ctx, obj_offset, len, &ret](const auto &object_data) {
	  LOG_PREFIX(ObjectDataHandler::read);
	  DEBUGT("reading {}~{}",
		 ctx.t,
		 object_data.get_reserved_data_base(),
		 object_data.get_reserved_data_len());
	  /* Assumption: callers ensure that onode size is <= reserved
	   * size and that len is adjusted here prior to call */
	  ceph_assert(!object_data.is_null());
	  ceph_assert((obj_offset + len) <= object_data.get_reserved_data_len());
	  ceph_assert(len > 0);
	  laddr_t loffset =
	    object_data.get_reserved_data_base() + obj_offset;
	  return ctx.tm.get_pins(
	    ctx.t,
	    loffset,
	    len
	  ).si_then([ctx, loffset, len, &ret](auto _pins) {
	    // offset~len falls within reserved region and len > 0
	    ceph_assert(_pins.size() >= 1);
	    ceph_assert((*_pins.begin())->get_key() <= loffset);
	    return seastar::do_with(
	      std::move(_pins),
	      loffset,
	      [ctx, loffset, len, &ret](auto &pins, auto &current) {
		return trans_intr::do_for_each(
		  pins,
		  [ctx, loffset, len, &current, &ret](auto &pin)
		  -> read_iertr::future<> {
		    ceph_assert(current <= (loffset + len));
		    ceph_assert(
		      (loffset + len) > pin->get_key());
		    laddr_t end = std::min(
		      pin->get_key() + pin->get_length(),
		      loffset + len);
		    if (pin->get_val().is_zero()) {
		      ceph_assert(end > current); // See LBAManager::get_mappings
		      ret.append_zero(end - current);
		      current = end;
		      return seastar::now();
		    } else {
		      LOG_PREFIX(ObjectDataHandler::read);
		      auto key = pin->get_key();
		      bool is_indirect = pin->is_indirect();
                      extent_len_t off = pin->get_intermediate_offset();
		      DEBUGT("reading {}~{}, indirect: {}, "
			"intermediate offset: {}, current: {}, end: {}",
			ctx.t,
			key,
			pin->get_length(),
			is_indirect,
			off,
			current,
			end);
		      return ctx.tm.read_pin<ObjectDataBlock>(
			ctx.t,
			std::move(pin)
		      ).si_then([&ret, &current, end, key, off,
				is_indirect](auto extent) {
			ceph_assert(
			  is_indirect
			    ? (key - off + extent->get_length()) >= end
			    : (extent->get_laddr() + extent->get_length()) >= end);
			ceph_assert(end > current);
			ret.append(
			  bufferptr(
			    extent->get_bptr(),
			    off + current - (is_indirect ? key : extent->get_laddr()),
			    end - current));
			current = end;
			return seastar::now();
		      }).handle_error_interruptible(
			read_iertr::pass_further{},
			crimson::ct_error::assert_all{
			  "ObjectDataHandler::read hit invalid error"
			}
		      );
		    }
		  });
	      });
	  });
	}).si_then([&ret] {
	  return std::move(ret);
	});
    });
}

ObjectDataHandler::fiemap_ret ObjectDataHandler::fiemap(
  context_t ctx,
  objaddr_t obj_offset,
  extent_len_t len)
{
  return seastar::do_with(
    std::map<uint64_t, uint64_t>(),
    [ctx, obj_offset, len](auto &ret) {
    return with_object_data(
      ctx,
      [ctx, obj_offset, len, &ret](const auto &object_data) {
      LOG_PREFIX(ObjectDataHandler::fiemap);
      DEBUGT(
	"{}~{}, reservation {}~{}",
        ctx.t,
        obj_offset,
        len,
        object_data.get_reserved_data_base(),
        object_data.get_reserved_data_len());
      /* Assumption: callers ensure that onode size is <= reserved
       * size and that len is adjusted here prior to call */
      ceph_assert(!object_data.is_null());
      ceph_assert((obj_offset + len) <= object_data.get_reserved_data_len());
      ceph_assert(len > 0);
      laddr_t loffset =
        object_data.get_reserved_data_base() + obj_offset;
      return ctx.tm.get_pins(
        ctx.t,
        loffset,
        len
      ).si_then([loffset, len, &object_data, &ret](auto &&pins) {
	ceph_assert(pins.size() >= 1);
        ceph_assert((*pins.begin())->get_key() <= loffset);
	for (auto &&i: pins) {
	  if (!(i->get_val().is_zero())) {
	    auto ret_left = std::max(i->get_key(), loffset);
	    auto ret_right = std::min(
	      i->get_key() + i->get_length(),
	      loffset + len);
	    assert(ret_right > ret_left);
	    ret.emplace(
	      std::make_pair(
		ret_left - object_data.get_reserved_data_base(),
		ret_right - ret_left
	      ));
	  }
	}
      });
    }).si_then([&ret] {
      return std::move(ret);
    });
  });
}

ObjectDataHandler::truncate_ret ObjectDataHandler::truncate(
  context_t ctx,
  objaddr_t offset)
{
  return with_object_data(
    ctx,
    [this, ctx, offset](auto &object_data) {
      LOG_PREFIX(ObjectDataHandler::truncate);
      DEBUGT("truncating {}~{} offset: {}",
	     ctx.t,
	     object_data.get_reserved_data_base(),
	     object_data.get_reserved_data_len(),
	     offset);
      if (offset < object_data.get_reserved_data_len()) {
	return trim_data_reservation(ctx, object_data, offset);
      } else if (offset > object_data.get_reserved_data_len()) {
	return prepare_data_reservation(
	  ctx,
	  object_data,
	  p2roundup(offset, ctx.tm.get_block_size()));
      } else {
	return truncate_iertr::now();
      }
    });
}

ObjectDataHandler::clear_ret ObjectDataHandler::clear(
  context_t ctx)
{
  return with_object_data(
    ctx,
    [this, ctx](auto &object_data) {
      LOG_PREFIX(ObjectDataHandler::clear);
      DEBUGT("clearing: {}~{}",
	     ctx.t,
	     object_data.get_reserved_data_base(),
	     object_data.get_reserved_data_len());
      if (object_data.is_null()) {
	return clear_iertr::now();
      }
      return trim_data_reservation(ctx, object_data, 0);
    });
}

ObjectDataHandler::clone_ret ObjectDataHandler::clone_extents(
  context_t ctx,
  object_data_t &object_data,
  lba_pin_list_t &pins,
  laddr_t data_base)
{
  LOG_PREFIX(ObjectDataHandler::clone_extents);
  TRACET(" object_data: {}~{}, data_base: {}",
    ctx.t,
    object_data.get_reserved_data_base(),
    object_data.get_reserved_data_len(),
    data_base);
  return ctx.tm.dec_ref(
    ctx.t,
    object_data.get_reserved_data_base()
  ).si_then(
    [&pins, &object_data, ctx, data_base](auto) mutable {
      return seastar::do_with(
	(extent_len_t)0,
	[&object_data, ctx, data_base, &pins](auto &last_pos) {
	return trans_intr::do_for_each(
	  pins,
	  [&last_pos, &object_data, ctx, data_base](auto &pin) {
	  auto offset = pin->get_key() - data_base;
	  ceph_assert(offset == last_pos);
	  auto fut = TransactionManager::alloc_extent_iertr
	    ::make_ready_future<LBAMappingRef>();
	  auto addr = object_data.get_reserved_data_base() + offset;
	  if (pin->get_val().is_zero()) {
	    fut = ctx.tm.reserve_region(ctx.t, addr, pin->get_length());
	  } else {
	    fut = ctx.tm.clone_pin(ctx.t, addr, *pin);
	  }
	  return fut.si_then(
	    [&pin, &last_pos, offset](auto) {
	    last_pos = offset + pin->get_length();
	    return seastar::now();
	  }).handle_error_interruptible(
	    crimson::ct_error::input_output_error::pass_further(),
	    crimson::ct_error::assert_all("not possible")
	  );
	}).si_then([&last_pos, &object_data, ctx] {
	  if (last_pos != object_data.get_reserved_data_len()) {
	    return ctx.tm.reserve_region(
	      ctx.t,
	      object_data.get_reserved_data_base() + last_pos,
	      object_data.get_reserved_data_len() - last_pos
	    ).si_then([](auto) {
	      return seastar::now();
	    });
	  }
	  return TransactionManager::reserve_extent_iertr::now();
	});
      });
    },
    ObjectDataHandler::write_iertr::pass_further{},
    crimson::ct_error::assert_all{
      "object_data_handler::clone invalid error"
    }
  );
}

ObjectDataHandler::clone_ret ObjectDataHandler::clone(
  context_t ctx)
{
  // the whole clone procedure can be seperated into the following steps:
  // 	1. let clone onode(d_object_data) take the head onode's
  // 	   object data base;
  // 	2. reserve a new region in lba tree for the head onode;
  // 	3. clone all extents of the clone onode, see transaction_manager.h
  // 	   for the details of clone_pin;
  // 	4. reserve the space between the head onode's size and its reservation
  // 	   length.
  return with_objects_data(
    ctx,
    [ctx, this](auto &object_data, auto &d_object_data) {
    ceph_assert(d_object_data.is_null());
    if (object_data.is_null()) {
      return clone_iertr::now();
    }
    return prepare_data_reservation(
      ctx,
      d_object_data,
      object_data.get_reserved_data_len()
    ).si_then([&object_data, &d_object_data, ctx, this] {
      assert(!object_data.is_null());
      auto base = object_data.get_reserved_data_base();
      auto len = object_data.get_reserved_data_len();
      object_data.clear();
      LOG_PREFIX(ObjectDataHandler::clone);
      DEBUGT("cloned obj reserve_data_base: {}, len {}",
	ctx.t,
	d_object_data.get_reserved_data_base(),
	d_object_data.get_reserved_data_len());
      return prepare_data_reservation(
	ctx,
	object_data,
	d_object_data.get_reserved_data_len()
      ).si_then([&d_object_data, ctx, &object_data, base, len, this] {
	LOG_PREFIX("ObjectDataHandler::clone");
	DEBUGT("head obj reserve_data_base: {}, len {}",
	  ctx.t,
	  object_data.get_reserved_data_base(),
	  object_data.get_reserved_data_len());
	return ctx.tm.get_pins(ctx.t, base, len
	).si_then([ctx, &object_data, &d_object_data, base, this](auto pins) {
	  return seastar::do_with(
	    std::move(pins),
	    [ctx, &object_data, &d_object_data, base, this](auto &pins) {
	    return clone_extents(ctx, object_data, pins, base
	    ).si_then([ctx, &d_object_data, base, &pins, this] {
	      return clone_extents(ctx, d_object_data, pins, base);
	    }).si_then([&pins, ctx] {
	      return do_removals(ctx, pins);
	    });
	  });
	});
      });
    });
  });
}

} // namespace crimson::os::seastore
