[ceph.git] / ceph / src / crimson / os / seastore / onode_manager / staged-fltree / stages / node_stage.cc

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

#include "node_stage.h"

#include "crimson/os/seastore/onode_manager/staged-fltree/node_extent_mutable.h"
#include "node_stage_layout.h"

namespace crimson::os::seastore::onode {

#define NODE_T node_extent_t<FieldType, NODE_TYPE>
#define NODE_INST(FT, NT) node_extent_t<FT, NT>

template <typename FieldType, node_type_t NODE_TYPE>
const char* NODE_T::p_left_bound() const
{
  if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
    // N3 internal node doesn't have the right part
    return nullptr;
  } else {
    auto ret = p_start() +
               fields().get_item_end_offset(keys(), node_size);
    if constexpr (NODE_TYPE == node_type_t::INTERNAL) {
      if (is_level_tail()) {
        ret -= sizeof(laddr_t);
      }
    }
    return ret;
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
node_offset_t NODE_T::size_to_nxt_at(index_t index) const
{
  assert(index < keys());
  if constexpr (FIELD_TYPE == field_type_t::N0 ||
                FIELD_TYPE == field_type_t::N1) {
    return FieldType::estimate_insert_one();
  } else if constexpr (FIELD_TYPE == field_type_t::N2) {
    auto p_end = p_start() +
                 p_fields->get_item_end_offset(index, node_size);
    return FieldType::estimate_insert_one() + ns_oid_view_t(p_end).size();
  } else {
    ceph_abort("N3 node is not nested");
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
container_range_t NODE_T::get_nxt_container(index_t index) const
{
  if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
    ceph_abort("N3 internal node doesn't have the right part");
  } else {
    auto item_start_offset = p_fields->get_item_start_offset(
        index, node_size);
    auto item_end_offset = p_fields->get_item_end_offset(
        index, node_size);
    assert(item_start_offset < item_end_offset);
    auto item_p_start = p_start() + item_start_offset;
    auto item_p_end = p_start() + item_end_offset;
    if constexpr (FIELD_TYPE == field_type_t::N2) {
      // range for sub_items_t<NODE_TYPE>
      item_p_end = ns_oid_view_t(item_p_end).p_start();
      assert(item_p_start < item_p_end);
    } else {
      // range for item_iterator_t<NODE_TYPE>
    }
    return {{item_p_start, item_p_end}, node_size};
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
void NODE_T::bootstrap_extent(
    NodeExtentMutable& mut,
    field_type_t field_type, node_type_t node_type,
    bool is_level_tail, level_t level)
{
  node_header_t::bootstrap_extent(
      mut, field_type, node_type, is_level_tail, level);
  mut.copy_in_relative(
      sizeof(node_header_t), typename FieldType::num_keys_t(0u));
}

template <typename FieldType, node_type_t NODE_TYPE>
void NODE_T::update_is_level_tail(
    NodeExtentMutable& mut, const node_extent_t& extent, bool value)
{
  assert(mut.get_length() == extent.node_size);
  assert(mut.get_read() == extent.p_start());
  node_header_t::update_is_level_tail(mut, extent.p_fields->header, value);
}

template <typename FieldType, node_type_t NODE_TYPE>
template <IsFullKey Key>
memory_range_t NODE_T::insert_prefix_at(
    NodeExtentMutable& mut, const node_extent_t& node, const Key& key,
    index_t index, node_offset_t size, const char* p_left_bound)
{
  assert(mut.get_length() == node.node_size);
  assert(mut.get_read() == node.p_start());
  if constexpr (FIELD_TYPE == field_type_t::N0 ||
                FIELD_TYPE == field_type_t::N1) {
    assert(index <= node.keys());
    assert(p_left_bound == node.p_left_bound());
    assert(size > FieldType::estimate_insert_one());
    auto size_right = size - FieldType::estimate_insert_one();
    const char* p_insert = node.p_start() +
                           node.fields().get_item_end_offset(index, mut.get_length());
    const char* p_insert_front = p_insert - size_right;
    FieldType::insert_at(mut, key, node.fields(), index, size_right);
    mut.shift_absolute(p_left_bound,
                       p_insert - p_left_bound,
                       -(int)size_right);
    return {p_insert_front, p_insert};
  } else if constexpr (FIELD_TYPE == field_type_t::N2) {
    ceph_abort("not implemented");
  } else {
    ceph_abort("impossible");
  }
}
#define IPA_TEMPLATE(FT, NT, Key)                                         \
  template memory_range_t NODE_INST(FT, NT)::insert_prefix_at<Key>(       \
      NodeExtentMutable&, const node_extent_t&, const Key&, \
      index_t, node_offset_t, const char*)
IPA_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, key_view_t);
IPA_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, key_view_t);
IPA_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, key_view_t);
IPA_TEMPLATE(node_fields_0_t, node_type_t::LEAF, key_view_t);
IPA_TEMPLATE(node_fields_1_t, node_type_t::LEAF, key_view_t);
IPA_TEMPLATE(node_fields_2_t, node_type_t::LEAF, key_view_t);
IPA_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, key_hobj_t);
IPA_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, key_hobj_t);
IPA_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, key_hobj_t);
IPA_TEMPLATE(node_fields_0_t, node_type_t::LEAF, key_hobj_t);
IPA_TEMPLATE(node_fields_1_t, node_type_t::LEAF, key_hobj_t);
IPA_TEMPLATE(node_fields_2_t, node_type_t::LEAF, key_hobj_t);

template <typename FieldType, node_type_t NODE_TYPE>
void NODE_T::update_size_at(
    NodeExtentMutable& mut, const node_extent_t& node, index_t index, int change)
{
  assert(mut.get_length() == node.node_size);
  assert(mut.get_read() == node.p_start());
  assert(index < node.keys());
  FieldType::update_size_at(mut, node.fields(), index, change);
}

template <typename FieldType, node_type_t NODE_TYPE>
node_offset_t NODE_T::trim_until(
    NodeExtentMutable& mut, const node_extent_t& node, index_t index)
{
  assert(mut.get_length() == node.node_size);
  assert(mut.get_read() == node.p_start());
  assert(!node.is_level_tail());
  auto keys = node.keys();
  assert(index <= keys);
  if (index == keys) {
    return 0;
  }
  if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
    ceph_abort("not implemented");
  } else {
    mut.copy_in_absolute(
        (void*)&node.p_fields->num_keys, num_keys_t(index));
  }
  // no need to calculate trim size for node
  return 0;
}

template <typename FieldType, node_type_t NODE_TYPE>
node_offset_t NODE_T::trim_at(
    NodeExtentMutable& mut, const node_extent_t& node,
    index_t index, node_offset_t trimmed)
{
  assert(mut.get_length() == node.node_size);
  assert(mut.get_read() == node.p_start());
  assert(!node.is_level_tail());
  assert(index < node.keys());
  if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
    ceph_abort("not implemented");
  } else {
    extent_len_t node_size = mut.get_length();
    node_offset_t offset = node.p_fields->get_item_start_offset(
        index, node_size);
    size_t new_offset = offset + trimmed;
    assert(new_offset < node.p_fields->get_item_end_offset(index, node_size));
    mut.copy_in_absolute(const_cast<void*>(node.p_fields->p_offset(index)),
                         node_offset_t(new_offset));
    mut.copy_in_absolute(
        (void*)&node.p_fields->num_keys, num_keys_t(index + 1));
  }
  // no need to calculate trim size for node
  return 0;
}

template <typename FieldType, node_type_t NODE_TYPE>
node_offset_t NODE_T::erase_at(
    NodeExtentMutable& mut, const node_extent_t& node,
    index_t index, const char* p_left_bound)
{
  assert(mut.get_length() == node.node_size);
  assert(mut.get_read() == node.p_start());
  if constexpr (FIELD_TYPE == field_type_t::N0 ||
                FIELD_TYPE == field_type_t::N1) {
    assert(node.keys() > 0);
    assert(index < node.keys());
    assert(p_left_bound == node.p_left_bound());
    return FieldType::erase_at(mut, node.fields(), index, p_left_bound);
  } else {
    ceph_abort("not implemented");
  }
}

#define NODE_TEMPLATE(FT, NT) template class NODE_INST(FT, NT)
NODE_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL);
NODE_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL);
NODE_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL);
NODE_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL);
NODE_TEMPLATE(node_fields_0_t, node_type_t::LEAF);
NODE_TEMPLATE(node_fields_1_t, node_type_t::LEAF);
NODE_TEMPLATE(node_fields_2_t, node_type_t::LEAF);
NODE_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF);

#define APPEND_T node_extent_t<FieldType, NODE_TYPE>::Appender<KT>

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
APPEND_T::Appender(NodeExtentMutable* p_mut, const node_extent_t& node, bool open)
    : p_mut{p_mut}, p_start{p_mut->get_write()}
{
  assert(p_start == node.p_start());
  assert(node.keys());
  assert(node.node_size == p_mut->get_length());
  extent_len_t node_size = node.node_size;
  if (open) {
    // seek as open_nxt()
    if constexpr (FIELD_TYPE == field_type_t::N0 ||
                  FIELD_TYPE == field_type_t::N1) {
      p_append_left = p_start + node.fields().get_key_start_offset(
          node.keys() - 1, node_size);
      p_append_left += sizeof(typename FieldType::key_t);
      p_append_right = p_start +
                       node.fields().get_item_end_offset(node.keys() - 1,
                                                         node_size);
    } else if constexpr (FIELD_TYPE == field_type_t::N2) {
      ceph_abort("not implemented");
    } else {
      ceph_abort("impossible path");
    }
    num_keys = node.keys() - 1;
  } else {
    if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
      std::ignore = node_size;
      ceph_abort("not implemented");
    } else {
      p_append_left = p_start + node.fields().get_key_start_offset(
          node.keys(), node_size);
      p_append_right = p_start +
                       node.fields().get_item_end_offset(node.keys(),
                                                         node_size);
    }
    num_keys = node.keys();
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
void APPEND_T::append(const node_extent_t& src, index_t from, index_t items)
{
  assert(from <= src.keys());
  if (p_src == nullptr) {
    p_src = &src;
  } else {
    assert(p_src == &src);
  }
  assert(p_src->node_size == p_mut->get_length());
  extent_len_t node_size = src.node_size;
  if (items == 0) {
    return;
  }
  assert(from < src.keys());
  assert(from + items <= src.keys());
  num_keys += items;
  if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
    std::ignore = node_size;
    ceph_abort("not implemented");
  } else {
    // append left part forwards
    node_offset_t offset_left_start = src.fields().get_key_start_offset(
        from, node_size);
    node_offset_t offset_left_end = src.fields().get_key_start_offset(
        from + items, node_size);
    node_offset_t left_size = offset_left_end - offset_left_start;
    if (num_keys == 0) {
      // no need to adjust offset
      assert(from == 0);
      assert(p_start + offset_left_start == p_append_left);
      p_mut->copy_in_absolute(p_append_left,
          src.p_start() + offset_left_start, left_size);
    } else {
      node_offset_t step_size = FieldType::estimate_insert_one();
      extent_len_t offset_base = src.fields().get_item_end_offset(
          from, node_size);
      int offset_change = p_append_right - p_start - offset_base;
      auto p_offset_dst = p_append_left;
      if constexpr (FIELD_TYPE != field_type_t::N2) {
        // copy keys
        p_mut->copy_in_absolute(p_append_left,
            src.p_start() + offset_left_start, left_size);
        // point to offset for update
        p_offset_dst += sizeof(typename FieldType::key_t);
      }
      for (auto i = from; i < from + items; ++i) {
        int new_offset = src.fields().get_item_start_offset(i, node_size) +
                         offset_change;
        assert(new_offset > 0);
        assert(new_offset < (int)node_size);
        p_mut->copy_in_absolute(p_offset_dst, node_offset_t(new_offset));
        p_offset_dst += step_size;
      }
      assert(p_append_left + left_size + sizeof(typename FieldType::key_t) ==
             p_offset_dst);
    }
    p_append_left += left_size;

    // append right part backwards
    auto offset_right_start = src.fields().get_item_end_offset(
        from + items, node_size);
    auto offset_right_end = src.fields().get_item_end_offset(
        from, node_size);
    int right_size = offset_right_end - offset_right_start;
    assert(right_size > 0);
    assert(right_size < (int)node_size);
    p_append_right -= right_size;
    p_mut->copy_in_absolute(p_append_right,
        src.p_start() + offset_right_start, node_offset_t(right_size));
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
void APPEND_T::append(
    const full_key_t<KT>& key, const value_input_t& value, const value_t*& p_value)
{
  if constexpr (FIELD_TYPE == field_type_t::N3) {
    ceph_abort("not implemented");
  } else {
    ceph_abort("should not happen");
  }
}

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
std::tuple<NodeExtentMutable*, char*>
APPEND_T::open_nxt(const key_get_type& partial_key)
{
  if constexpr (FIELD_TYPE == field_type_t::N0 ||
                FIELD_TYPE == field_type_t::N1) {
    FieldType::append_key(*p_mut, partial_key, p_append_left);
  } else if constexpr (FIELD_TYPE == field_type_t::N2) {
    FieldType::append_key(*p_mut, partial_key, p_append_right);
  } else {
    ceph_abort("impossible path");
  }
  return {p_mut, p_append_right};
}

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
std::tuple<NodeExtentMutable*, char*>
APPEND_T::open_nxt(const full_key_t<KT>& key)
{
  if constexpr (FIELD_TYPE == field_type_t::N0 ||
                FIELD_TYPE == field_type_t::N1) {
    FieldType::append_key(*p_mut, key, p_append_left);
  } else if constexpr (FIELD_TYPE == field_type_t::N2) {
    FieldType::append_key(*p_mut, key, p_append_right);
  } else {
    ceph_abort("impossible path");
  }
  return {p_mut, p_append_right};
}

template <typename FieldType, node_type_t NODE_TYPE>
template <KeyT KT>
char* APPEND_T::wrap()
{
  assert(p_append_left <= p_append_right);
  assert(p_src);
  if constexpr (NODE_TYPE == node_type_t::INTERNAL) {
    if (p_src->is_level_tail()) {
      laddr_t tail_value = p_src->get_end_p_laddr()->value;
      p_append_right -= sizeof(laddr_t);
      assert(p_append_left <= p_append_right);
      p_mut->copy_in_absolute(p_append_right, tail_value);
    }
  }
  p_mut->copy_in_absolute(p_start + offsetof(FieldType, num_keys), num_keys);
  return p_append_left;
}

#define APPEND_TEMPLATE(FT, NT, KT) template class node_extent_t<FT, NT>::Appender<KT>
APPEND_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, KeyT::VIEW);
APPEND_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_0_t, node_type_t::LEAF, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_1_t, node_type_t::LEAF, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_2_t, node_type_t::LEAF, KeyT::VIEW);
APPEND_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF, KeyT::VIEW);
APPEND_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, KeyT::HOBJ);
APPEND_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, KeyT::HOBJ);
APPEND_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, KeyT::HOBJ);
APPEND_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL, KeyT::HOBJ);
APPEND_TEMPLATE(node_fields_0_t, node_type_t::LEAF, KeyT::HOBJ);
APPEND_TEMPLATE(node_fields_1_t, node_type_t::LEAF, KeyT::HOBJ);
APPEND_TEMPLATE(node_fields_2_t, node_type_t::LEAF, KeyT::HOBJ);
APPEND_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF, KeyT::HOBJ);

}
Commit	Line	Data
f67539c2 TL	1	// -- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:nil --
	2	// vim: ts=8 sw=2 smarttab
	3
	4	#include "node_stage.h"
	5
	6	#include "crimson/os/seastore/onode_manager/staged-fltree/node_extent_mutable.h"
	7	#include "node_stage_layout.h"
	8
	9	namespace crimson::os::seastore::onode {
	10
	11	#define NODE_T node_extent_t<FieldType, NODE_TYPE>
	12	#define NODE_INST(FT, NT) node_extent_t<FT, NT>
	13
	14	template <typename FieldType, node_type_t NODE_TYPE>
20effc67 TL	15	const char* NODE_T::p_left_bound() const
20effc67 TL	16	{
f67539c2 TL	17	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
	18	// N3 internal node doesn't have the right part
	19	return nullptr;
	20	} else {
20effc67 TL	21	auto ret = p_start() +
20effc67 TL	22	fields().get_item_end_offset(keys(), node_size);
f67539c2 TL	23	if constexpr (NODE_TYPE == node_type_t::INTERNAL) {
	24	if (is_level_tail()) {
	25	ret -= sizeof(laddr_t);
	26	}
	27	}
	28	return ret;
	29	}
	30	}
	31
	32	template <typename FieldType, node_type_t NODE_TYPE>
20effc67 TL	33	node_offset_t NODE_T::size_to_nxt_at(index_t index) const
20effc67 TL	34	{
f67539c2 TL	35	assert(index < keys());
	36	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
	37	FIELD_TYPE == field_type_t::N1) {
	38	return FieldType::estimate_insert_one();
	39	} else if constexpr (FIELD_TYPE == field_type_t::N2) {
20effc67 TL	40	auto p_end = p_start() +
20effc67 TL	41	p_fields->get_item_end_offset(index, node_size);
f67539c2 TL	42	return FieldType::estimate_insert_one() + ns_oid_view_t(p_end).size();
	43	} else {
	44	ceph_abort("N3 node is not nested");
	45	}
	46	}
	47
	48	template <typename FieldType, node_type_t NODE_TYPE>
20effc67 TL	49	container_range_t NODE_T::get_nxt_container(index_t index) const
20effc67 TL	50	{
f67539c2 TL	51	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
	52	ceph_abort("N3 internal node doesn't have the right part");
	53	} else {
20effc67 TL	54	auto item_start_offset = p_fields->get_item_start_offset(
	55	index, node_size);
	56	auto item_end_offset = p_fields->get_item_end_offset(
	57	index, node_size);
f67539c2 TL	58	assert(item_start_offset < item_end_offset);
	59	auto item_p_start = p_start() + item_start_offset;
	60	auto item_p_end = p_start() + item_end_offset;
	61	if constexpr (FIELD_TYPE == field_type_t::N2) {
	62	// range for sub_items_t<NODE_TYPE>
	63	item_p_end = ns_oid_view_t(item_p_end).p_start();
	64	assert(item_p_start < item_p_end);
	65	} else {
	66	// range for item_iterator_t<NODE_TYPE>
	67	}
20effc67	68	return {{item_p_start, item_p_end}, node_size};
f67539c2 TL	69	}
	70	}
	71
	72	template <typename FieldType, node_type_t NODE_TYPE>
	73	void NODE_T::bootstrap_extent(
	74	NodeExtentMutable& mut,
	75	field_type_t field_type, node_type_t node_type,
20effc67 TL	76	bool is_level_tail, level_t level)
20effc67 TL	77	{
f67539c2 TL	78	node_header_t::bootstrap_extent(
	79	mut, field_type, node_type, is_level_tail, level);
	80	mut.copy_in_relative(
	81	sizeof(node_header_t), typename FieldType::num_keys_t(0u));
	82	}
	83
	84	template <typename FieldType, node_type_t NODE_TYPE>
	85	void NODE_T::update_is_level_tail(
20effc67 TL	86	NodeExtentMutable& mut, const node_extent_t& extent, bool value)
	87	{
	88	assert(mut.get_length() == extent.node_size);
	89	assert(mut.get_read() == extent.p_start());
f67539c2 TL	90	node_header_t::update_is_level_tail(mut, extent.p_fields->header, value);
	91	}
	92
	93	template <typename FieldType, node_type_t NODE_TYPE>
1e59de90	94	template <IsFullKey Key>
f67539c2	95	memory_range_t NODE_T::insert_prefix_at(
1e59de90	96	NodeExtentMutable& mut, const node_extent_t& node, const Key& key,
20effc67 TL	97	index_t index, node_offset_t size, const char* p_left_bound)
	98	{
	99	assert(mut.get_length() == node.node_size);
	100	assert(mut.get_read() == node.p_start());
f67539c2 TL	101	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
	102	FIELD_TYPE == field_type_t::N1) {
	103	assert(index <= node.keys());
	104	assert(p_left_bound == node.p_left_bound());
	105	assert(size > FieldType::estimate_insert_one());
	106	auto size_right = size - FieldType::estimate_insert_one();
20effc67 TL	107	const char* p_insert = node.p_start() +
20effc67 TL	108	node.fields().get_item_end_offset(index, mut.get_length());
f67539c2	109	const char* p_insert_front = p_insert - size_right;
1e59de90	110	FieldType::insert_at(mut, key, node.fields(), index, size_right);
f67539c2 TL	111	mut.shift_absolute(p_left_bound,
	112	p_insert - p_left_bound,
	113	-(int)size_right);
	114	return {p_insert_front, p_insert};
	115	} else if constexpr (FIELD_TYPE == field_type_t::N2) {
	116	ceph_abort("not implemented");
	117	} else {
	118	ceph_abort("impossible");
	119	}
	120	}
1e59de90 TL	121	#define IPA_TEMPLATE(FT, NT, Key) \
	122	template memory_range_t NODE_INST(FT, NT)::insert_prefix_at<Key>( \
	123	NodeExtentMutable&, const node_extent_t&, const Key&, \
f67539c2	124	index_t, node_offset_t, const char*)
1e59de90 TL	125	IPA_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, key_view_t);
	126	IPA_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, key_view_t);
	127	IPA_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, key_view_t);
	128	IPA_TEMPLATE(node_fields_0_t, node_type_t::LEAF, key_view_t);
	129	IPA_TEMPLATE(node_fields_1_t, node_type_t::LEAF, key_view_t);
	130	IPA_TEMPLATE(node_fields_2_t, node_type_t::LEAF, key_view_t);
	131	IPA_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, key_hobj_t);
	132	IPA_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, key_hobj_t);
	133	IPA_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, key_hobj_t);
	134	IPA_TEMPLATE(node_fields_0_t, node_type_t::LEAF, key_hobj_t);
	135	IPA_TEMPLATE(node_fields_1_t, node_type_t::LEAF, key_hobj_t);
	136	IPA_TEMPLATE(node_fields_2_t, node_type_t::LEAF, key_hobj_t);
f67539c2 TL	137
	138	template <typename FieldType, node_type_t NODE_TYPE>
	139	void NODE_T::update_size_at(
20effc67 TL	140	NodeExtentMutable& mut, const node_extent_t& node, index_t index, int change)
	141	{
	142	assert(mut.get_length() == node.node_size);
	143	assert(mut.get_read() == node.p_start());
f67539c2 TL	144	assert(index < node.keys());
	145	FieldType::update_size_at(mut, node.fields(), index, change);
	146	}
	147
	148	template <typename FieldType, node_type_t NODE_TYPE>
	149	node_offset_t NODE_T::trim_until(
20effc67 TL	150	NodeExtentMutable& mut, const node_extent_t& node, index_t index)
	151	{
	152	assert(mut.get_length() == node.node_size);
	153	assert(mut.get_read() == node.p_start());
f67539c2 TL	154	assert(!node.is_level_tail());
	155	auto keys = node.keys();
	156	assert(index <= keys);
	157	if (index == keys) {
	158	return 0;
	159	}
	160	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
	161	ceph_abort("not implemented");
	162	} else {
	163	mut.copy_in_absolute(
	164	(void*)&node.p_fields->num_keys, num_keys_t(index));
	165	}
	166	// no need to calculate trim size for node
	167	return 0;
	168	}
	169
	170	template <typename FieldType, node_type_t NODE_TYPE>
	171	node_offset_t NODE_T::trim_at(
	172	NodeExtentMutable& mut, const node_extent_t& node,
20effc67 TL	173	index_t index, node_offset_t trimmed)
	174	{
	175	assert(mut.get_length() == node.node_size);
	176	assert(mut.get_read() == node.p_start());
f67539c2 TL	177	assert(!node.is_level_tail());
	178	assert(index < node.keys());
	179	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
	180	ceph_abort("not implemented");
	181	} else {
20effc67 TL	182	extent_len_t node_size = mut.get_length();
	183	node_offset_t offset = node.p_fields->get_item_start_offset(
	184	index, node_size);
f67539c2	185	size_t new_offset = offset + trimmed;
20effc67	186	assert(new_offset < node.p_fields->get_item_end_offset(index, node_size));
f67539c2 TL	187	mut.copy_in_absolute(const_cast<void*>(node.p_fields->p_offset(index)),
	188	node_offset_t(new_offset));
	189	mut.copy_in_absolute(
	190	(void*)&node.p_fields->num_keys, num_keys_t(index + 1));
	191	}
	192	// no need to calculate trim size for node
	193	return 0;
	194	}
	195
20effc67 TL	196	template <typename FieldType, node_type_t NODE_TYPE>
	197	node_offset_t NODE_T::erase_at(
	198	NodeExtentMutable& mut, const node_extent_t& node,
	199	index_t index, const char* p_left_bound)
	200	{
	201	assert(mut.get_length() == node.node_size);
	202	assert(mut.get_read() == node.p_start());
	203	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
	204	FIELD_TYPE == field_type_t::N1) {
	205	assert(node.keys() > 0);
	206	assert(index < node.keys());
	207	assert(p_left_bound == node.p_left_bound());
	208	return FieldType::erase_at(mut, node.fields(), index, p_left_bound);
	209	} else {
	210	ceph_abort("not implemented");
	211	}
	212	}
	213
f67539c2 TL	214	#define NODE_TEMPLATE(FT, NT) template class NODE_INST(FT, NT)
	215	NODE_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL);
	216	NODE_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL);
	217	NODE_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL);
	218	NODE_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL);
	219	NODE_TEMPLATE(node_fields_0_t, node_type_t::LEAF);
	220	NODE_TEMPLATE(node_fields_1_t, node_type_t::LEAF);
	221	NODE_TEMPLATE(node_fields_2_t, node_type_t::LEAF);
	222	NODE_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF);
	223
	224	#define APPEND_T node_extent_t<FieldType, NODE_TYPE>::Appender<KT>
	225
	226	template <typename FieldType, node_type_t NODE_TYPE>
	227	template <KeyT KT>
20effc67 TL	228	APPEND_T::Appender(NodeExtentMutable* p_mut, const node_extent_t& node, bool open)
	229	: p_mut{p_mut}, p_start{p_mut->get_write()}
	230	{
	231	assert(p_start == node.p_start());
	232	assert(node.keys());
	233	assert(node.node_size == p_mut->get_length());
	234	extent_len_t node_size = node.node_size;
	235	if (open) {
	236	// seek as open_nxt()
	237	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
	238	FIELD_TYPE == field_type_t::N1) {
	239	p_append_left = p_start + node.fields().get_key_start_offset(
	240	node.keys() - 1, node_size);
	241	p_append_left += sizeof(typename FieldType::key_t);
	242	p_append_right = p_start +
	243	node.fields().get_item_end_offset(node.keys() - 1,
	244	node_size);
	245	} else if constexpr (FIELD_TYPE == field_type_t::N2) {
	246	ceph_abort("not implemented");
	247	} else {
	248	ceph_abort("impossible path");
	249	}
	250	num_keys = node.keys() - 1;
	251	} else {
	252	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
	253	std::ignore = node_size;
	254	ceph_abort("not implemented");
	255	} else {
	256	p_append_left = p_start + node.fields().get_key_start_offset(
	257	node.keys(), node_size);
	258	p_append_right = p_start +
	259	node.fields().get_item_end_offset(node.keys(),
	260	node_size);
	261	}
	262	num_keys = node.keys();
	263	}
	264	}
	265
	266	template <typename FieldType, node_type_t NODE_TYPE>
	267	template <KeyT KT>
	268	void APPEND_T::append(const node_extent_t& src, index_t from, index_t items)
	269	{
f67539c2 TL	270	assert(from <= src.keys());
	271	if (p_src == nullptr) {
	272	p_src = &src;
	273	} else {
	274	assert(p_src == &src);
	275	}
20effc67 TL	276	assert(p_src->node_size == p_mut->get_length());
20effc67 TL	277	extent_len_t node_size = src.node_size;
f67539c2 TL	278	if (items == 0) {
	279	return;
	280	}
	281	assert(from < src.keys());
	282	assert(from + items <= src.keys());
	283	num_keys += items;
	284	if constexpr (std::is_same_v<FieldType, internal_fields_3_t>) {
20effc67 TL	285	std::ignore = node_size;
20effc67 TL	286	ceph_abort("not implemented");
f67539c2 TL	287	} else {
f67539c2 TL	288	// append left part forwards
20effc67 TL	289	node_offset_t offset_left_start = src.fields().get_key_start_offset(
	290	from, node_size);
	291	node_offset_t offset_left_end = src.fields().get_key_start_offset(
	292	from + items, node_size);
f67539c2 TL	293	node_offset_t left_size = offset_left_end - offset_left_start;
	294	if (num_keys == 0) {
	295	// no need to adjust offset
	296	assert(from == 0);
	297	assert(p_start + offset_left_start == p_append_left);
	298	p_mut->copy_in_absolute(p_append_left,
	299	src.p_start() + offset_left_start, left_size);
	300	} else {
	301	node_offset_t step_size = FieldType::estimate_insert_one();
20effc67 TL	302	extent_len_t offset_base = src.fields().get_item_end_offset(
20effc67 TL	303	from, node_size);
f67539c2 TL	304	int offset_change = p_append_right - p_start - offset_base;
	305	auto p_offset_dst = p_append_left;
	306	if constexpr (FIELD_TYPE != field_type_t::N2) {
	307	// copy keys
	308	p_mut->copy_in_absolute(p_append_left,
	309	src.p_start() + offset_left_start, left_size);
	310	// point to offset for update
	311	p_offset_dst += sizeof(typename FieldType::key_t);
	312	}
	313	for (auto i = from; i < from + items; ++i) {
20effc67 TL	314	int new_offset = src.fields().get_item_start_offset(i, node_size) +
	315	offset_change;
	316	assert(new_offset > 0);
	317	assert(new_offset < (int)node_size);
	318	p_mut->copy_in_absolute(p_offset_dst, node_offset_t(new_offset));
f67539c2 TL	319	p_offset_dst += step_size;
	320	}
	321	assert(p_append_left + left_size + sizeof(typename FieldType::key_t) ==
	322	p_offset_dst);
	323	}
	324	p_append_left += left_size;
	325
	326	// append right part backwards
20effc67 TL	327	auto offset_right_start = src.fields().get_item_end_offset(
	328	from + items, node_size);
	329	auto offset_right_end = src.fields().get_item_end_offset(
	330	from, node_size);
	331	int right_size = offset_right_end - offset_right_start;
	332	assert(right_size > 0);
	333	assert(right_size < (int)node_size);
f67539c2 TL	334	p_append_right -= right_size;
f67539c2 TL	335	p_mut->copy_in_absolute(p_append_right,
20effc67	336	src.p_start() + offset_right_start, node_offset_t(right_size));
f67539c2 TL	337	}
	338	}
	339
	340	template <typename FieldType, node_type_t NODE_TYPE>
	341	template <KeyT KT>
	342	void APPEND_T::append(
20effc67 TL	343	const full_key_t<KT>& key, const value_input_t& value, const value_t*& p_value)
20effc67 TL	344	{
f67539c2 TL	345	if constexpr (FIELD_TYPE == field_type_t::N3) {
	346	ceph_abort("not implemented");
	347	} else {
	348	ceph_abort("should not happen");
	349	}
	350	}
	351
	352	template <typename FieldType, node_type_t NODE_TYPE>
	353	template <KeyT KT>
	354	std::tuple<NodeExtentMutable, char>
20effc67 TL	355	APPEND_T::open_nxt(const key_get_type& partial_key)
20effc67 TL	356	{
f67539c2 TL	357	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
	358	FIELD_TYPE == field_type_t::N1) {
	359	FieldType::append_key(*p_mut, partial_key, p_append_left);
	360	} else if constexpr (FIELD_TYPE == field_type_t::N2) {
	361	FieldType::append_key(*p_mut, partial_key, p_append_right);
	362	} else {
	363	ceph_abort("impossible path");
	364	}
	365	return {p_mut, p_append_right};
	366	}
	367
	368	template <typename FieldType, node_type_t NODE_TYPE>
	369	template <KeyT KT>
	370	std::tuple<NodeExtentMutable, char>
20effc67 TL	371	APPEND_T::open_nxt(const full_key_t<KT>& key)
20effc67 TL	372	{
f67539c2 TL	373	if constexpr (FIELD_TYPE == field_type_t::N0 \|\|
f67539c2 TL	374	FIELD_TYPE == field_type_t::N1) {
1e59de90	375	FieldType::append_key(*p_mut, key, p_append_left);
f67539c2	376	} else if constexpr (FIELD_TYPE == field_type_t::N2) {
1e59de90	377	FieldType::append_key(*p_mut, key, p_append_right);
f67539c2 TL	378	} else {
	379	ceph_abort("impossible path");
	380	}
	381	return {p_mut, p_append_right};
	382	}
	383
	384	template <typename FieldType, node_type_t NODE_TYPE>
	385	template <KeyT KT>
20effc67 TL	386	char* APPEND_T::wrap()
20effc67 TL	387	{
f67539c2 TL	388	assert(p_append_left <= p_append_right);
	389	assert(p_src);
	390	if constexpr (NODE_TYPE == node_type_t::INTERNAL) {
	391	if (p_src->is_level_tail()) {
	392	laddr_t tail_value = p_src->get_end_p_laddr()->value;
	393	p_append_right -= sizeof(laddr_t);
	394	assert(p_append_left <= p_append_right);
	395	p_mut->copy_in_absolute(p_append_right, tail_value);
	396	}
	397	}
	398	p_mut->copy_in_absolute(p_start + offsetof(FieldType, num_keys), num_keys);
	399	return p_append_left;
	400	}
	401
	402	#define APPEND_TEMPLATE(FT, NT, KT) template class node_extent_t<FT, NT>::Appender<KT>
	403	APPEND_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, KeyT::VIEW);
	404	APPEND_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, KeyT::VIEW);
	405	APPEND_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, KeyT::VIEW);
	406	APPEND_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL, KeyT::VIEW);
	407	APPEND_TEMPLATE(node_fields_0_t, node_type_t::LEAF, KeyT::VIEW);
	408	APPEND_TEMPLATE(node_fields_1_t, node_type_t::LEAF, KeyT::VIEW);
	409	APPEND_TEMPLATE(node_fields_2_t, node_type_t::LEAF, KeyT::VIEW);
	410	APPEND_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF, KeyT::VIEW);
	411	APPEND_TEMPLATE(node_fields_0_t, node_type_t::INTERNAL, KeyT::HOBJ);
	412	APPEND_TEMPLATE(node_fields_1_t, node_type_t::INTERNAL, KeyT::HOBJ);
	413	APPEND_TEMPLATE(node_fields_2_t, node_type_t::INTERNAL, KeyT::HOBJ);
	414	APPEND_TEMPLATE(internal_fields_3_t, node_type_t::INTERNAL, KeyT::HOBJ);
	415	APPEND_TEMPLATE(node_fields_0_t, node_type_t::LEAF, KeyT::HOBJ);
	416	APPEND_TEMPLATE(node_fields_1_t, node_type_t::LEAF, KeyT::HOBJ);
	417	APPEND_TEMPLATE(node_fields_2_t, node_type_t::LEAF, KeyT::HOBJ);
	418	APPEND_TEMPLATE(leaf_fields_3_t, node_type_t::LEAF, KeyT::HOBJ);
	419
	420	}