tuple.h

#ifndef _NDB_TUPLE_H_
#define _NDB_TUPLE_H_

#include <atomic>
#include <vector>
#include <string>
#include <utility>
#include <limits>
#include <unordered_map>
#include <ostream>

#include "amd64.h"
#include "core.h"
#include "counter.h"
#include "macros.h"
#include "varkey.h"
#include "util.h"
#include "rcu.h"
#include "thread.h"
#include "spinlock.h"
#include "small_unordered_map.h"
#include "prefetch.h"
#include "pthread.h"

// debugging tool
//#define TUPLE_LOCK_OWNERSHIP_CHECKING

template <template <typename> class Protocol, typename Traits>
  class transaction; // forward decl

// XXX: hack
extern std::string (*g_proto_version_str)(uint64_t v);

/**
 * A dbtuple is the type of value which we stick
 * into underlying (non-transactional) data structures- it
 * also contains the memory of the value
 */
struct dbtuple {
  friend std::ostream &
  operator<<(std::ostream &o, const dbtuple &tuple);

public:
  // trying to save space by putting constraints
  // on node maximums
  typedef uint32_t version_t;
  typedef uint16_t node_size_type;
  typedef uint64_t tid_t;
  typedef uint8_t * record_type;
  typedef const uint8_t * const_record_type;
  typedef size_t size_type;
  typedef std::string string_type;

  static const tid_t MIN_TID = 0;
  static const tid_t MAX_TID = (tid_t) -1;

  // lock ownership helpers- works by recording all tuple
  // locks obtained in each transaction, and then when the txn
  // finishes, calling AssertAllTupleLocksReleased(), which makes
  // sure the current thread is no longer the owner of any locks
  // acquired during the txn
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
  static void
  TupleLockRegionBegin()
  {
    if (!tl_locked_nodes)
      tl_locked_nodes = new std::vector<const dbtuple *>;
    tl_locked_nodes->clear();
  }

  // is used to signal the end of a tuple lock region
  static void
  AssertAllTupleLocksReleased()
  {
    ALWAYS_ASSERT(tl_locked_nodes);
    for (auto p : *tl_locked_nodes)
      ALWAYS_ASSERT(!p->is_lock_owner());
    tl_locked_nodes->clear();
  }

private:

  static void
  AddTupleToLockRegion(const dbtuple *tuple)
  {
    ALWAYS_ASSERT(tuple->is_locked());
    ALWAYS_ASSERT(tuple->is_lock_owner());
    if (tl_locked_nodes)
      tl_locked_nodes->emplace_back(tuple);
  }

  static __thread std::vector<const dbtuple *> *tl_locked_nodes;
#endif

private:
  static const version_t HDR_LOCKED_MASK = 0x1;

  static const version_t HDR_DELETING_SHIFT = 1;
  static const version_t HDR_DELETING_MASK = 0x1 << HDR_DELETING_SHIFT;

  static const version_t HDR_WRITE_INTENT_SHIFT = 2;
  static const version_t HDR_WRITE_INTENT_MASK = 0x1 << HDR_WRITE_INTENT_SHIFT;

  static const version_t HDR_MODIFYING_SHIFT = 3;
  static const version_t HDR_MODIFYING_MASK = 0x1 << HDR_MODIFYING_SHIFT;

  static const version_t HDR_LATEST_SHIFT = 4;
  static const version_t HDR_LATEST_MASK = 0x1 << HDR_LATEST_SHIFT;

  static const version_t HDR_VERSION_SHIFT = 5;
  static const version_t HDR_VERSION_MASK = ((version_t)-1) << HDR_VERSION_SHIFT;

public:

#ifdef TUPLE_MAGIC
  class magic_failed_exception: public std::exception {};
  static const uint8_t TUPLE_MAGIC = 0x29U;
  uint8_t magic;
  inline ALWAYS_INLINE void CheckMagic() const
  {
    if (unlikely(magic != TUPLE_MAGIC)) {
      print(1);
      // so we can catch it later and print out useful debugging output
      throw magic_failed_exception();
    }
  }
#else
  inline ALWAYS_INLINE void CheckMagic() const {}
#endif

  // NB(stephentu): ABA problem happens after some multiple of
  // 2^(NBits(version_t)-6) concurrent modifications- somewhat low probability
  // event, so we let it happen
  //
  // <-- low bits
  // [ locked | deleting | write_intent | modifying | latest | version ]
  // [  0..1  |   1..2   |    2..3      |   3..4    |  4..5  |  5..32  ]
  volatile version_t hdr;

#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
  pthread_t lock_owner;
#endif

  // uninterpreted TID
  tid_t version;

  // small sizes on purpose
  node_size_type size; // actual size of record
                       // (only meaningful is the deleting bit is not set)
  node_size_type alloc_size; // max size record allowed. is the space
                             // available for the record buf

  dbtuple *next; // be very careful about traversing this pointer,
                 // GC is capable of reaping it at certain (well defined)
                 // points, and will not bother to set it to null

#ifdef TUPLE_CHECK_KEY
  // for debugging
  std::string key;
  void *tree;
#endif

#ifdef CHECK_INVARIANTS
  // for debugging
  std::atomic<uint64_t> opaque;
#endif

  // must be last field
  uint8_t value_start[0];

  void print(std::ostream &o, unsigned len) const;

private:
  // private ctor/dtor b/c we do some special memory stuff
  // ctors start node off as latest node

  static inline ALWAYS_INLINE node_size_type
  CheckBounds(size_type s)
  {
    INVARIANT(s <= std::numeric_limits<node_size_type>::max());
    return s;
  }

  dbtuple(const dbtuple &) = delete;
  dbtuple(dbtuple &&) = delete;
  dbtuple &operator=(const dbtuple &) = delete;

  // creates a (new) record with a tentative value at MAX_TID
  dbtuple(size_type size, size_type alloc_size, bool acquire_lock)
    :
#ifdef TUPLE_MAGIC
      magic(TUPLE_MAGIC),
#endif
      hdr(HDR_LATEST_MASK |
          (acquire_lock ? (HDR_LOCKED_MASK | HDR_WRITE_INTENT_MASK) : 0) |
          (!size ? HDR_DELETING_MASK : 0))
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
      , lock_owner(0) // not portable
#endif
      , version(MAX_TID)
      , size(CheckBounds(size))
      , alloc_size(CheckBounds(alloc_size))
      , next(nullptr)
#ifdef TUPLE_CHECK_KEY
      , key()
      , tree(nullptr)
#endif
#ifdef CHECK_INVARIANTS
      , opaque(0)
#endif
  {
    INVARIANT(((char *)this) + sizeof(*this) == (char *) &value_start[0]);
    INVARIANT(is_latest());
    INVARIANT(size || is_deleting());
    ++g_evt_dbtuple_creates;
    g_evt_dbtuple_bytes_allocated += alloc_size + sizeof(dbtuple);
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
    if (acquire_lock) {
      lock_owner = pthread_self();
      AddTupleToLockRegion(this);
      INVARIANT(is_lock_owner());
    }
#endif
    COMPILER_MEMORY_FENCE; // for acquire_lock
  }

  // creates a record at version derived from base
  // (inheriting its value).
  dbtuple(tid_t version,
          struct dbtuple *base,
          size_type alloc_size,
          bool set_latest)
    :
#ifdef TUPLE_MAGIC
      magic(TUPLE_MAGIC),
#endif
      hdr(set_latest ? HDR_LATEST_MASK : 0)
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
      , lock_owner(0) // not portable
#endif
      , version(version)
      , size(base->size)
      , alloc_size(CheckBounds(alloc_size))
      , next(base->next)
#ifdef TUPLE_CHECK_KEY
      , key()
      , tree(nullptr)
#endif
#ifdef CHECK_INVARIANTS
      , opaque(0)
#endif
  {
    INVARIANT(size <= alloc_size);
    INVARIANT(set_latest == is_latest());
    if (base->is_deleting())
      mark_deleting();
    NDB_MEMCPY(&value_start[0], base->get_value_start(), size);
    ++g_evt_dbtuple_creates;
    g_evt_dbtuple_bytes_allocated += alloc_size + sizeof(dbtuple);
  }

  // creates a spill record, copying in the *old* value if necessary, but
  // setting the size to the *new* value
  dbtuple(tid_t version,
          const_record_type r,
          size_type old_size,
          size_type new_size,
          size_type alloc_size,
          struct dbtuple *next,
          bool set_latest,
          bool needs_old_value)
    :
#ifdef TUPLE_MAGIC
      magic(TUPLE_MAGIC),
#endif
      hdr((set_latest ? HDR_LATEST_MASK : 0) | (!new_size ? HDR_DELETING_MASK : 0))
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
      , lock_owner(0) // not portable
#endif
      , version(version)
      , size(CheckBounds(new_size))
      , alloc_size(CheckBounds(alloc_size))
      , next(next)
#ifdef TUPLE_CHECK_KEY
      , key()
      , tree(nullptr)
#endif
#ifdef CHECK_INVARIANTS
      , opaque(0)
#endif
  {
    INVARIANT(!needs_old_value || old_size <= alloc_size);
    INVARIANT(new_size <= alloc_size);
    INVARIANT(set_latest == is_latest());
    INVARIANT(new_size || is_deleting());
    if (needs_old_value)
      NDB_MEMCPY(&value_start[0], r, old_size);
    ++g_evt_dbtuple_creates;
    g_evt_dbtuple_bytes_allocated += alloc_size + sizeof(dbtuple);
  }

  friend class rcu;
  ~dbtuple();

  static event_avg_counter g_evt_avg_dbtuple_stable_version_spins;
  static event_avg_counter g_evt_avg_dbtuple_lock_acquire_spins;
  static event_avg_counter g_evt_avg_dbtuple_read_retries;

public:

  enum ReadStatus {
    READ_FAILED,
    READ_EMPTY,
    READ_RECORD,
  };

  inline void
  prefetch() const
  {
#ifdef TUPLE_PREFETCH
    prefetch_bytes(this, sizeof(*this) + alloc_size);
#endif
  }

  // gcs *this* instance, ignoring the chain
  void gc_this();

  inline bool
  is_locked() const
  {
    return IsLocked(hdr);
  }

  static inline bool
  IsLocked(version_t v)
  {
    return v & HDR_LOCKED_MASK;
  }

#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
  inline bool
  is_lock_owner() const
  {
    return pthread_equal(pthread_self(), lock_owner);
  }
#else
  inline bool
  is_lock_owner() const
  {
    return true;
  }
#endif

  inline version_t
  lock(bool write_intent)
  {
    // XXX: implement SPINLOCK_BACKOFF
    CheckMagic();
#ifdef ENABLE_EVENT_COUNTERS
    unsigned nspins = 0;
#endif
    version_t v = hdr;
    const version_t lockmask = write_intent ?
      (HDR_LOCKED_MASK | HDR_WRITE_INTENT_MASK) :
      (HDR_LOCKED_MASK);
    while (IsLocked(v) ||
           !__sync_bool_compare_and_swap(&hdr, v, v | lockmask)) {
      nop_pause();
      v = hdr;
#ifdef ENABLE_EVENT_COUNTERS
      ++nspins;
#endif
    }
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
    lock_owner = pthread_self();
    AddTupleToLockRegion(this);
    INVARIANT(is_lock_owner());
#endif
    COMPILER_MEMORY_FENCE;
    INVARIANT(IsLocked(hdr));
    INVARIANT(!write_intent || IsWriteIntent(hdr));
    INVARIANT(!IsModifying(hdr));
#ifdef ENABLE_EVENT_COUNTERS
    g_evt_avg_dbtuple_lock_acquire_spins.offer(nspins);
#endif
    return hdr;
  }

  inline void
  unlock()
  {
    CheckMagic();
    version_t v = hdr;
    bool newv = false;
    INVARIANT(IsLocked(v));
    INVARIANT(is_lock_owner());
    if (IsModifying(v) || IsWriteIntent(v)) {
      newv = true;
      const version_t n = Version(v);
      v &= ~HDR_VERSION_MASK;
      v |= (((n + 1) << HDR_VERSION_SHIFT) & HDR_VERSION_MASK);
    }
    // clear locked + modifying bits
    v &= ~(HDR_LOCKED_MASK | HDR_MODIFYING_MASK | HDR_WRITE_INTENT_MASK);
    if (newv) {
      INVARIANT(!reader_check_version(v));
      INVARIANT(!writer_check_version(v));
    }
    INVARIANT(!IsLocked(v));
    INVARIANT(!IsModifying(v));
    INVARIANT(!IsWriteIntent(v));
#ifdef TUPLE_LOCK_OWNERSHIP_CHECKING
    // XXX: not portable, but whatever
    lock_owner = 0;
    INVARIANT(!is_lock_owner());
#endif
    COMPILER_MEMORY_FENCE;
    hdr = v;
  }

  inline bool
  is_deleting() const
  {
    return IsDeleting(hdr);
  }

  static inline bool
  IsDeleting(version_t v)
  {
    return v & HDR_DELETING_MASK;
  }

  inline void
  mark_deleting()
  {
    CheckMagic();
    // the lock on the latest version guards non-latest versions
    INVARIANT(!is_latest() || is_locked());
    INVARIANT(!is_latest() || is_lock_owner());
    INVARIANT(!is_deleting());
    hdr |= HDR_DELETING_MASK;
  }

  inline void
  clear_deleting()
  {
    CheckMagic();
    INVARIANT(is_locked());
    INVARIANT(is_lock_owner());
    INVARIANT(is_deleting());
    hdr &= ~HDR_DELETING_MASK;
  }

  inline bool
  is_modifying() const
  {
    return IsModifying(hdr);
  }

  inline void
  mark_modifying()
  {
    CheckMagic();
    version_t v = hdr;
    INVARIANT(IsLocked(v));
    INVARIANT(is_lock_owner());
    //INVARIANT(!IsModifying(v)); // mark_modifying() must be re-entrant
    v |= HDR_MODIFYING_MASK;
    COMPILER_MEMORY_FENCE; // XXX: is this fence necessary?
    hdr = v;
    COMPILER_MEMORY_FENCE;
  }

  static inline bool
  IsModifying(version_t v)
  {
    return v & HDR_MODIFYING_MASK;
  }

  inline bool
  is_write_intent() const
  {
    return IsWriteIntent(hdr);
  }

  static inline bool
  IsWriteIntent(version_t v)
  {
    return v & HDR_WRITE_INTENT_MASK;
  }

  inline bool
  is_latest() const
  {
    return IsLatest(hdr);
  }

  static inline bool
  IsLatest(version_t v)
  {
    return v & HDR_LATEST_MASK;
  }

  inline void
  clear_latest()
  {
    CheckMagic();
    INVARIANT(is_locked());
    INVARIANT(is_lock_owner());
    INVARIANT(is_latest());
    hdr &= ~HDR_LATEST_MASK;
  }

  static inline version_t
  Version(version_t v)
  {
    return (v & HDR_VERSION_MASK) >> HDR_VERSION_SHIFT;
  }

  inline version_t
  reader_stable_version(bool allow_write_intent) const
  {
    version_t v = hdr;
#ifdef ENABLE_EVENT_COUNTERS
    unsigned long nspins = 0;
#endif
    while (IsModifying(v) ||
           (!allow_write_intent && IsWriteIntent(v))) {
      nop_pause();
      v = hdr;
#ifdef ENABLE_EVENT_COUNTERS
      ++nspins;
#endif
    }
    COMPILER_MEMORY_FENCE;
#ifdef ENABLE_EVENT_COUNTERS
    g_evt_avg_dbtuple_stable_version_spins.offer(nspins);
#endif
    return v;
  }

  /**
   * returns true if succeeded, false otherwise
   */
  inline bool
  try_writer_stable_version(version_t &v, unsigned int spins) const
  {
    v = hdr;
    while (IsWriteIntent(v) && spins--) {
      INVARIANT(IsLocked(v));
      nop_pause();
      v = hdr;
    }
    const bool ret = !IsWriteIntent(v);
    COMPILER_MEMORY_FENCE;
    INVARIANT(ret || IsLocked(v));
    INVARIANT(!ret || !IsModifying(v));
    return ret;
  }

  inline version_t
  unstable_version() const
  {
    return hdr;
  }

  inline bool
  reader_check_version(version_t version) const
  {
    COMPILER_MEMORY_FENCE;
    // are the versions the same, modulo the
    // {locked, write_intent, latest} bits?
    const version_t MODULO_BITS =
      (HDR_LOCKED_MASK | HDR_WRITE_INTENT_MASK | HDR_LATEST_MASK);
    return (hdr & ~MODULO_BITS) == (version & ~MODULO_BITS);
  }

  inline bool
  writer_check_version(version_t version) const
  {
    COMPILER_MEMORY_FENCE;
    return hdr == version;
  }

  inline ALWAYS_INLINE struct dbtuple *
  get_next()
  {
    return next;
  }

  inline const struct dbtuple *
  get_next() const
  {
    return next;
  }

  inline ALWAYS_INLINE void
  set_next(struct dbtuple *next)
  {
    CheckMagic();
    this->next = next;
  }

  inline void
  clear_next()
  {
    CheckMagic();
    this->next = nullptr;
  }

  inline ALWAYS_INLINE uint8_t *
  get_value_start()
  {
    CheckMagic();
    return &value_start[0];
  }

  inline ALWAYS_INLINE const uint8_t *
  get_value_start() const
  {
    return &value_start[0];
  }

  // worst name ever...
  inline bool
  is_not_behind(tid_t t) const
  {
    return version <= t;
  }

private:

#ifdef ENABLE_EVENT_COUNTERS
  struct scoped_recorder {
    scoped_recorder(unsigned long &n) : n(&n) {}
    ~scoped_recorder()
    {
      g_evt_avg_dbtuple_read_retries.offer(*n);
    }
  private:
    unsigned long *n;
  };
#endif

  // written to be non-recursive
  template <typename Reader, typename StringAllocator>
  static ReadStatus
  record_at_chain(
      const dbtuple *starting, tid_t t, tid_t &start_t,
      Reader &reader, StringAllocator &sa, bool allow_write_intent)
  {
#ifdef ENABLE_EVENT_COUNTERS
    unsigned long nretries = 0;
    scoped_recorder rec(nretries);
#endif
    const dbtuple *current = starting;
  loop:
    INVARIANT(current->version != MAX_TID);
    const version_t v = current->reader_stable_version(allow_write_intent);
    const struct dbtuple *p;
    const bool found = current->is_not_behind(t);
    if (found) {
      start_t = current->version;
      const size_t read_sz = IsDeleting(v) ? 0 : current->size;
      if (unlikely(read_sz && !reader(current->get_value_start(), read_sz, sa)))
        goto retry;
      if (unlikely(!current->reader_check_version(v)))
        goto retry;
      return read_sz ? READ_RECORD : READ_EMPTY;
    } else {
      p = current->get_next();
    }
    if (unlikely(!current->reader_check_version(v)))
      goto retry;
    if (p) {
      current = p;
      goto loop;
    }
    // see note in record_at()
    start_t = MIN_TID;
    return READ_EMPTY;
  retry:
#ifdef ENABLE_EVENT_COUNTERS
    ++nretries;
#endif
    goto loop;
  }

  // we force one level of inlining, but don't force record_at_chain()
  // to be inlined
  template <typename Reader, typename StringAllocator>
  inline ALWAYS_INLINE ReadStatus
  record_at(
      tid_t t, tid_t &start_t,
      Reader &reader, StringAllocator &sa, bool allow_write_intent) const
  {
#ifdef ENABLE_EVENT_COUNTERS
    unsigned long nretries = 0;
    scoped_recorder rec(nretries);
#endif
    if (unlikely(version == MAX_TID)) {
      // XXX(stephentu): HACK! we use MAX_TID to indicate a tentative
      // "insert"- the actual latest value is empty.
      //
      // since our system is screwed anyways if we ever reach MAX_TID, this
      // is OK for now, but a real solution should exist at some point
      start_t = MIN_TID;
      return READ_EMPTY;
    }
  loop:
    const version_t v = reader_stable_version(allow_write_intent);
    const struct dbtuple *p;
    const bool found = is_not_behind(t);
    if (found) {
      //if (unlikely(!IsLatest(v)))
      //  return READ_FAILED;
      start_t = version;
      const size_t read_sz = IsDeleting(v) ? 0 : size;
      if (unlikely(read_sz && !reader(get_value_start(), read_sz, sa)))
        goto retry;
      if (unlikely(!reader_check_version(v)))
        goto retry;
      return read_sz ? READ_RECORD : READ_EMPTY;
    } else {
      p = get_next();
    }
    if (unlikely(!reader_check_version(v)))
      goto retry;
    if (p)
      return record_at_chain(p, t, start_t, reader, sa, allow_write_intent);
    // NB(stephentu): if we reach the end of a chain then we assume that
    // the record exists as a deleted record.
    //
    // This is safe because we have been very careful to not garbage collect
    // elements along the chain until it is guaranteed that the record
    // is superceded by later record in any consistent read. Therefore,
    // if we reach the end of the chain, then it *must* be the case that
    // the record does not actually exist.
    //
    // Note that MIN_TID is the *wrong* tid to use here given wrap-around- we
    // really should be setting this value to the tid which represents the
    // oldest TID possible in the system. But we currently don't implement
    // wrap around
    start_t = MIN_TID;
    return READ_EMPTY;
  retry:
#ifdef ENABLE_EVENT_COUNTERS
    ++nretries;
#endif
    goto loop;
  }

  static event_counter g_evt_dbtuple_creates;
  static event_counter g_evt_dbtuple_logical_deletes;
  static event_counter g_evt_dbtuple_physical_deletes;
  static event_counter g_evt_dbtuple_bytes_allocated;
  static event_counter g_evt_dbtuple_bytes_freed;
  static event_counter g_evt_dbtuple_spills;
  static event_counter g_evt_dbtuple_inplace_buf_insufficient;
  static event_counter g_evt_dbtuple_inplace_buf_insufficient_on_spill;
  static event_avg_counter g_evt_avg_record_spill_len;

public:

  /**
   * Read the record at tid t. Returns true if such a record exists, false
   * otherwise (ie the record was GC-ed, or other reasons). On a successful
   * read, the value @ start_t will be stored in r
   *
   * NB(stephentu): calling stable_read() while holding the lock
   * is an error- this will cause deadlock
   */
  template <typename Reader, typename StringAllocator>
  inline ALWAYS_INLINE ReadStatus
  stable_read(
      tid_t t, tid_t &start_t,
      Reader &reader, StringAllocator &sa,
      bool allow_write_intent) const
  {
    return record_at(t, start_t, reader, sa, allow_write_intent);
  }

  inline bool
  is_latest_version(tid_t t) const
  {
    return is_latest() && is_not_behind(t);
  }

  bool
  stable_is_latest_version(tid_t t) const
  {
    version_t v = 0;
    if (!try_writer_stable_version(v, 16))
      return false;
    // now v is a stable version
    INVARIANT(!IsWriteIntent(v));
    INVARIANT(!IsModifying(v));
    const bool ret = IsLatest(v) && is_not_behind(t);
    // only check_version() if the answer would be true- otherwise,
    // no point in doing a version check
    if (ret && writer_check_version(v))
      return true;
    else
      // no point in retrying, since we know it will fail (since we had a
      // version change)
      return false;
  }

  inline bool
  latest_value_is_nil() const
  {
    return is_latest() && size == 0;
  }

  inline bool
  stable_latest_value_is_nil() const
  {
    version_t v = 0;
    if (!try_writer_stable_version(v, 16))
      return false;
    INVARIANT(!IsWriteIntent(v));
    INVARIANT(!IsModifying(v));
    const bool ret = IsLatest(v) && size == 0;
    if (ret && writer_check_version(v))
      return true;
    else
      return false;
  }

  struct write_record_ret {
    write_record_ret() : head_(), rest_(), forced_spill_() {}
    write_record_ret(dbtuple *head, dbtuple* rest, bool forced_spill)
      : head_(head), rest_(rest), forced_spill_(forced_spill)
    {
      INVARIANT(head);
      INVARIANT(head != rest);
      INVARIANT(!forced_spill || rest);
    }
    dbtuple *head_;
    dbtuple *rest_;
    bool forced_spill_;
  };

  // XXX: kind of hacky, but we do this to avoid virtual
  // functions / passing multiple function pointers around
  enum TupleWriterMode {
    TUPLE_WRITER_NEEDS_OLD_VALUE, // all three args ignored
    TUPLE_WRITER_COMPUTE_NEEDED,
    TUPLE_WRITER_COMPUTE_DELTA_NEEDED, // last two args ignored
    TUPLE_WRITER_DO_WRITE,
    TUPLE_WRITER_DO_DELTA_WRITE,
  };
  typedef size_t (*tuple_writer_t)(TupleWriterMode, const void *, uint8_t *, size_t);

  /**
   * Always writes the record in the latest (newest) version slot,
   * not asserting whether or not inserting r @ t would violate the
   * sorted order invariant
   *
   * ret.first  = latest tuple after the write (guaranteed to not be nullptr)
   * ret.second = old version of tuple, iff no overwrite (can be nullptr)
   *
   * Note: if this != ret.first, then we need a tree replacement
   */
  template <typename Transaction>
  write_record_ret
  write_record_at(const Transaction *txn, tid_t t,
                  const void *v, tuple_writer_t writer)
  {
#ifndef DISABLE_OVERWRITE_IN_PLACE
    CheckMagic();
    INVARIANT(is_locked());
    INVARIANT(is_lock_owner());
    INVARIANT(is_latest());
    INVARIANT(is_write_intent());

    const size_t new_sz =
      v ? writer(TUPLE_WRITER_COMPUTE_NEEDED, v, get_value_start(), size) : 0;
    INVARIANT(!v || new_sz);
    INVARIANT(is_deleting() || size);
    const size_t old_sz = is_deleting() ? 0 : size;

    if (!new_sz)
      ++g_evt_dbtuple_logical_deletes;

    // try to overwrite this record
    if (likely(txn->can_overwrite_record_tid(version, t) && old_sz)) {
      INVARIANT(!is_deleting());
      // see if we have enough space
      if (likely(new_sz <= alloc_size)) {
        // directly update in place
        mark_modifying();
        if (v)
          writer(TUPLE_WRITER_DO_WRITE, v, get_value_start(), old_sz);
        version = t;
        size = new_sz;
        if (!new_sz)
          mark_deleting();
        return write_record_ret(this, nullptr, false);
      }

      //std::cerr
      //  << "existing: " << g_proto_version_str(version) << std::endl
      //  << "new     : " << g_proto_version_str(t)       << std::endl
      //  << "alloc_size : " << alloc_size                << std::endl
      //  << "new_sz     : " << new_sz                    << std::endl;

      // keep this tuple in the chain (it's wasteful, but not incorrect)
      // so that cleanup is easier
      //
      // XXX(stephentu): alloc_spill() should acquire the lock on
      // the returned tuple in the ctor, as an optimization

      const bool needs_old_value =
        writer(TUPLE_WRITER_NEEDS_OLD_VALUE, nullptr, nullptr, 0);
      INVARIANT(new_sz);
      INVARIANT(v);
      dbtuple * const rep =
        alloc_spill(t, get_value_start(), old_sz, new_sz,
                    this, true, needs_old_value);
      writer(TUPLE_WRITER_DO_WRITE, v, rep->get_value_start(), old_sz);
      INVARIANT(rep->is_latest());
      INVARIANT(rep->size == new_sz);
      clear_latest();
      ++g_evt_dbtuple_inplace_buf_insufficient;

      // [did not spill because of epochs, need to replace this with rep]
      return write_record_ret(rep, this, false);
    }

    //std::cerr
    //  << "existing: " << g_proto_version_str(version) << std::endl
    //  << "new     : " << g_proto_version_str(t)       << std::endl
    //  << "alloc_size : " << alloc_size                << std::endl
    //  << "new_sz     : " << new_sz                    << std::endl;

    // need to spill
    ++g_evt_dbtuple_spills;
    g_evt_avg_record_spill_len.offer(size);

    if (new_sz <= alloc_size && old_sz) {
      INVARIANT(!is_deleting());
      dbtuple * const spill = alloc(version, this, false);
      INVARIANT(!spill->is_latest());
      mark_modifying();
      set_next(spill);
      if (v)
        writer(TUPLE_WRITER_DO_WRITE, v, get_value_start(), size);
      version = t;
      size = new_sz;
      if (!new_sz)
        mark_deleting();
      return write_record_ret(this, spill, true);
    }

    const bool needs_old_value =
      writer(TUPLE_WRITER_NEEDS_OLD_VALUE, nullptr, nullptr, 0);
    dbtuple * const rep =
      alloc_spill(t, get_value_start(), old_sz, new_sz,
                  this, true, needs_old_value);
    if (v)
      writer(TUPLE_WRITER_DO_WRITE, v, rep->get_value_start(), size);
    INVARIANT(rep->is_latest());
    INVARIANT(rep->size == new_sz);
    INVARIANT(new_sz || rep->is_deleting()); // set by alloc_spill()
    clear_latest();
    ++g_evt_dbtuple_inplace_buf_insufficient_on_spill;
    return write_record_ret(rep, this, true);
#else
    CheckMagic();
    INVARIANT(is_locked());
    INVARIANT(is_lock_owner());
    INVARIANT(is_latest());
    INVARIANT(is_write_intent());

    const size_t new_sz =
      v ? writer(TUPLE_WRITER_COMPUTE_NEEDED, v, get_value_start(), size) : 0;
    INVARIANT(!v || new_sz);
    INVARIANT(is_deleting() || size);
    const size_t old_sz = is_deleting() ? 0 : size;

    if (!new_sz)
      ++g_evt_dbtuple_logical_deletes;

    const bool needs_old_value =
      writer(TUPLE_WRITER_NEEDS_OLD_VALUE, nullptr, nullptr, 0);
    dbtuple * const rep =
      alloc_spill(t, get_value_start(), old_sz, new_sz,
                  this, true, needs_old_value);
    if (v)
      writer(TUPLE_WRITER_DO_WRITE, v, rep->get_value_start(), size);
    INVARIANT(rep->is_latest());
    INVARIANT(rep->size == new_sz);
    INVARIANT(new_sz || rep->is_deleting()); // set by alloc_spill()
    clear_latest();
    ++g_evt_dbtuple_inplace_buf_insufficient_on_spill;
    return write_record_ret(rep, this, true);
#endif
  }

  // NB: we round up allocation sizes because jemalloc will do this
  // internally anyways, so we might as well grab more usable space (really
  // just internal vs external fragmentation)

  static inline dbtuple *
  alloc_first(size_type sz, bool acquire_lock)
  {
    INVARIANT(sz <= std::numeric_limits<node_size_type>::max());
    const size_t max_alloc_sz =
      std::numeric_limits<node_size_type>::max() + sizeof(dbtuple);
    const size_t alloc_sz =
      std::min(
          util::round_up<size_t, allocator::LgAllocAlignment>(sizeof(dbtuple) + sz),
          max_alloc_sz);
    char *p = reinterpret_cast<char *>(rcu::s_instance.alloc(alloc_sz));
    INVARIANT(p);
    INVARIANT((alloc_sz - sizeof(dbtuple)) >= sz);
    return new (p) dbtuple(
        sz, alloc_sz - sizeof(dbtuple), acquire_lock);
  }

  static inline dbtuple *
  alloc(tid_t version, struct dbtuple *base, bool set_latest)
  {
    const size_t max_alloc_sz =
      std::numeric_limits<node_size_type>::max() + sizeof(dbtuple);
    const size_t alloc_sz =
      std::min(
          util::round_up<size_t, allocator::LgAllocAlignment>(sizeof(dbtuple) + base->size),
          max_alloc_sz);
    char *p = reinterpret_cast<char *>(rcu::s_instance.alloc(alloc_sz));
    INVARIANT(p);
    return new (p) dbtuple(
        version, base, alloc_sz - sizeof(dbtuple), set_latest);
  }

  static inline dbtuple *
  alloc_spill(tid_t version, const_record_type value, size_type oldsz,
              size_type newsz, struct dbtuple *next, bool set_latest,
              bool copy_old_value)
  {
    INVARIANT(oldsz <= std::numeric_limits<node_size_type>::max());
    INVARIANT(newsz <= std::numeric_limits<node_size_type>::max());

    const size_t needed_sz =
      copy_old_value ? std::max(newsz, oldsz) : newsz;
    const size_t max_alloc_sz =
      std::numeric_limits<node_size_type>::max() + sizeof(dbtuple);
    const size_t alloc_sz =
      std::min(
          util::round_up<size_t, allocator::LgAllocAlignment>(sizeof(dbtuple) + needed_sz),
          max_alloc_sz);
    char *p = reinterpret_cast<char *>(rcu::s_instance.alloc(alloc_sz));
    INVARIANT(p);
    return new (p) dbtuple(
        version, value, oldsz, newsz,
        alloc_sz - sizeof(dbtuple), next, set_latest, copy_old_value);
  }


private:
  static inline void
  destruct_and_free(dbtuple *n)
  {
    const size_t alloc_sz = n->alloc_size + sizeof(*n);
    n->~dbtuple();
    rcu::s_instance.dealloc(n, alloc_sz);
  }

public:
  static void
  deleter(void *p)
  {
    dbtuple * const n = (dbtuple *) p;
    INVARIANT(!n->is_latest());
    INVARIANT(!n->is_locked());
    INVARIANT(!n->is_modifying());
    destruct_and_free(n);
  }

  static inline void
  release(dbtuple *n)
  {
    if (unlikely(!n))
      return;
    INVARIANT(n->is_locked());
    INVARIANT(!n->is_latest());
    rcu::s_instance.free_with_fn(n, deleter);
  }

  static inline void
  release_no_rcu(dbtuple *n)
  {
    if (unlikely(!n))
      return;
    INVARIANT(!n->is_latest());
    destruct_and_free(n);
  }

  static std::string
  VersionInfoStr(version_t v);

}
#if !defined(TUPLE_CHECK_KEY) && !defined(CHECK_INVARIANTS)
PACKED
#endif
;

#endif /* _NDB_TUPLE_H_ */