txn_impl.h

#ifndef _NDB_TXN_IMPL_H_
#define _NDB_TXN_IMPL_H_

#include "txn.h"
#include "lockguard.h"

// base definitions

template <template <typename> class Protocol, typename Traits>
transaction<Protocol, Traits>::transaction(uint64_t flags, string_allocator_type &sa)
  : transaction_base(flags), sa(&sa)
{
  INVARIANT(rcu::s_instance.in_rcu_region());
#ifdef BTREE_LOCK_OWNERSHIP_CHECKING
  concurrent_btree::NodeLockRegionBegin();
#endif
}

template <template <typename> class Protocol, typename Traits>
transaction<Protocol, Traits>::~transaction()
{
  // transaction shouldn't fall out of scope w/o resolution
  // resolution means TXN_EMBRYO, TXN_COMMITED, and TXN_ABRT
  INVARIANT(state != TXN_ACTIVE);
  INVARIANT(rcu::s_instance.in_rcu_region());
  const unsigned cur_depth = rcu_guard_->sync()->depth();
  rcu_guard_.destroy();
  if (cur_depth == 1) {
    INVARIANT(!rcu::s_instance.in_rcu_region());
    cast()->on_post_rcu_region_completion();
  }
#ifdef BTREE_LOCK_OWNERSHIP_CHECKING
  concurrent_btree::AssertAllNodeLocksReleased();
#endif
}

template <template <typename> class Protocol, typename Traits>
void
transaction<Protocol, Traits>::clear()
{
  // it's actually *more* efficient to not call clear explicitly on the
  // read/write/absent sets, and let the destructors do the clearing- this is
  // because the destructors can take shortcuts since it knows the obj doesn't
  // have to end in a valid state
}

template <template <typename> class Protocol, typename Traits>
void
transaction<Protocol, Traits>::abort_impl(abort_reason reason)
{
  abort_trap(reason);
  switch (state) {
  case TXN_EMBRYO:
  case TXN_ACTIVE:
    break;
  case TXN_ABRT:
    return;
  case TXN_COMMITED:
    throw transaction_unusable_exception();
  }
  state = TXN_ABRT;
  this->reason = reason;

  // on abort, we need to go over all insert nodes and
  // release the locks
  typename write_set_map::iterator it     = write_set.begin();
  typename write_set_map::iterator it_end = write_set.end();
  for (; it != it_end; ++it) {
    dbtuple * const tuple = it->get_tuple();
    if (it->is_insert()) {
      INVARIANT(tuple->is_locked());
      this->cleanup_inserted_tuple_marker(tuple, it->get_key(), it->get_btree());
      tuple->unlock();
    }
  }

  clear();
}

template <template <typename> class Protocol, typename Traits>
void
transaction<Protocol, Traits>::cleanup_inserted_tuple_marker(
    dbtuple *marker, const std::string &key, concurrent_btree *btr)
{
  // XXX: this code should really live in txn_proto2_impl.h
  INVARIANT(marker->version == dbtuple::MAX_TID);
  INVARIANT(marker->is_locked());
  INVARIANT(marker->is_lock_owner());
  typename concurrent_btree::value_type removed = 0;
  const bool did_remove = btr->remove(varkey(key), &removed);
  if (unlikely(!did_remove)) {
#ifdef CHECK_INVARIANTS
    std::cerr << " *** could not remove key: " << util::hexify(key)  << std::endl;
#ifdef TUPLE_CHECK_KEY
    std::cerr << " *** original key        : " << util::hexify(marker->key) << std::endl;
#endif
#endif
    ALWAYS_ASSERT(false);
  }
  INVARIANT(removed == (typename concurrent_btree::value_type) marker);
  INVARIANT(marker->is_latest());
  marker->clear_latest();
  dbtuple::release(marker); // rcu free
}

namespace {
  inline const char *
  transaction_state_to_cstr(transaction_base::txn_state state)
  {
    switch (state) {
    case transaction_base::TXN_EMBRYO: return "TXN_EMBRYO";
    case transaction_base::TXN_ACTIVE: return "TXN_ACTIVE";
    case transaction_base::TXN_ABRT: return "TXN_ABRT";
    case transaction_base::TXN_COMMITED: return "TXN_COMMITED";
    }
    ALWAYS_ASSERT(false);
    return 0;
  }

  inline std::string
  transaction_flags_to_str(uint64_t flags)
  {
    bool first = true;
    std::ostringstream oss;
    if (flags & transaction_base::TXN_FLAG_LOW_LEVEL_SCAN) {
      oss << "TXN_FLAG_LOW_LEVEL_SCAN";
      first = false;
    }
    if (flags & transaction_base::TXN_FLAG_READ_ONLY) {
      if (first)
        oss << "TXN_FLAG_READ_ONLY";
      else
        oss << " | TXN_FLAG_READ_ONLY";
      first = false;
    }
    return oss.str();
  }
}

template <template <typename> class Protocol, typename Traits>
void
transaction<Protocol, Traits>::dump_debug_info() const
{
  std::cerr << "Transaction (obj=" << util::hexify(this) << ") -- state "
       << transaction_state_to_cstr(state) << std::endl;
  std::cerr << "  Abort Reason: " << AbortReasonStr(reason) << std::endl;
  std::cerr << "  Flags: " << transaction_flags_to_str(flags) << std::endl;
  std::cerr << "  Read/Write sets:" << std::endl;

  std::cerr << "      === Read Set ===" << std::endl;
  // read-set
  for (typename read_set_map::const_iterator rs_it = read_set.begin();
       rs_it != read_set.end(); ++rs_it)
    std::cerr << *rs_it << std::endl;

  std::cerr << "      === Write Set ===" << std::endl;
  // write-set
  for (typename write_set_map::const_iterator ws_it = write_set.begin();
       ws_it != write_set.end(); ++ws_it)
    std::cerr << *ws_it << std::endl;

  std::cerr << "      === Absent Set ===" << std::endl;
  // absent-set
  for (typename absent_set_map::const_iterator as_it = absent_set.begin();
       as_it != absent_set.end(); ++as_it)
    std::cerr << "      B-tree Node " << util::hexify(as_it->first)
              << " : " << as_it->second << std::endl;

}

template <template <typename> class Protocol, typename Traits>
std::map<std::string, uint64_t>
transaction<Protocol, Traits>::get_txn_counters() const
{
  std::map<std::string, uint64_t> ret;

  // max_read_set_size
  ret["read_set_size"] = read_set.size();;
  ret["read_set_is_large?"] = !read_set.is_small_type();

  // max_absent_set_size
  ret["absent_set_size"] = absent_set.size();
  ret["absent_set_is_large?"] = !absent_set.is_small_type();

  // max_write_set_size
  ret["write_set_size"] = write_set.size();
  ret["write_set_is_large?"] = !write_set.is_small_type();

  return ret;
}

template <template <typename> class Protocol, typename Traits>
bool
transaction<Protocol, Traits>::handle_last_tuple_in_group(
    dbtuple_write_info &last,
    bool did_group_insert)
{
  if (did_group_insert) {
    // don't need to lock
    if (!last.is_insert())
      // we inserted the last run, and then we did 1+ more overwrites
      // to it, so we do NOT need to lock the node (again), but we DO
      // need to apply the latest write
      last.entry->set_do_write();
  } else {
    dbtuple *tuple = last.get_tuple();
    if (unlikely(tuple->version == dbtuple::MAX_TID)) {
      // if we race to put/insert w/ another txn which has inserted a new
      // record, we *must* abort b/c the other txn could try to put/insert
      // into a new record which we hold the lock on, so we must abort
      //
      // other ideas:
      // we could *not* abort if this txn did not insert any new records.
      // we could also release our insert locks and try to acquire them
      // again in sorted order
      return false; // signal abort
    }
    const dbtuple::version_t v = tuple->lock(true); // lock for write
    INVARIANT(dbtuple::IsLatest(v) == tuple->is_latest());
    last.mark_locked();
    if (unlikely(!dbtuple::IsLatest(v) ||
                 !cast()->can_read_tid(tuple->version))) {
      // XXX(stephentu): overly conservative (with the can_read_tid() check)
      return false; // signal abort
    }
    last.entry->set_do_write();
  }
  return true;
}

template <template <typename> class Protocol, typename Traits>
bool
transaction<Protocol, Traits>::commit(bool doThrow)
{
#ifdef TUPLE_MAGIC
  try {
#endif

  PERF_DECL(
      static std::string probe0_name(
        std::string(__PRETTY_FUNCTION__) + std::string(":total:")));
  ANON_REGION(probe0_name.c_str(), &transaction_base::g_txn_commit_probe0_cg);

  switch (state) {
  case TXN_EMBRYO:
  case TXN_ACTIVE:
    break;
  case TXN_COMMITED:
    return true;
  case TXN_ABRT:
    if (doThrow)
      throw transaction_abort_exception(reason);
    return false;
  }

  dbtuple_write_info_vec write_dbtuples;
  std::pair<bool, tid_t> commit_tid(false, 0);

  // copy write tuples to vector for sorting
  if (!write_set.empty()) {
    PERF_DECL(
        static std::string probe1_name(
          std::string(__PRETTY_FUNCTION__) + std::string(":lock_write_nodes:")));
    ANON_REGION(probe1_name.c_str(), &transaction_base::g_txn_commit_probe1_cg);
    INVARIANT(!is_snapshot());
    typename write_set_map::iterator it     = write_set.begin();
    typename write_set_map::iterator it_end = write_set.end();
    for (size_t pos = 0; it != it_end; ++it, ++pos) {
      INVARIANT(!it->is_insert() || it->get_tuple()->is_locked());
      write_dbtuples.emplace_back(it->get_tuple(), &(*it), it->is_insert(), pos);
    }
  }

  // read_only txns require consistent snapshots
  INVARIANT(!is_snapshot() || read_set.empty());
  INVARIANT(!is_snapshot() || write_set.empty());
  INVARIANT(!is_snapshot() || absent_set.empty());
  if (!is_snapshot()) {
    // we don't have consistent tids, or not a read-only txn

    // lock write nodes
    if (!write_dbtuples.empty()) {
      PERF_DECL(
          static std::string probe2_name(
            std::string(__PRETTY_FUNCTION__) + std::string(":lock_write_nodes:")));
      ANON_REGION(probe2_name.c_str(), &transaction_base::g_txn_commit_probe2_cg);
      // lock the logical nodes in sort order
      {
        PERF_DECL(
            static std::string probe6_name(
              std::string(__PRETTY_FUNCTION__) + std::string(":sort_write_nodes:")));
        ANON_REGION(probe6_name.c_str(), &transaction_base::g_txn_commit_probe6_cg);
        write_dbtuples.sort(); // in-place
      }
      typename dbtuple_write_info_vec::iterator it     = write_dbtuples.begin();
      typename dbtuple_write_info_vec::iterator it_end = write_dbtuples.end();
      dbtuple_write_info *last_px = nullptr;
      bool inserted_last_run = false;
      for (; it != it_end; last_px = &(*it), ++it) {
        if (likely(last_px && last_px->tuple != it->tuple)) {
          // on boundary
          if (unlikely(!handle_last_tuple_in_group(*last_px, inserted_last_run))) {
            abort_trap((reason = ABORT_REASON_WRITE_NODE_INTERFERENCE));
            goto do_abort;
          }
          inserted_last_run = false;
        }
        if (it->is_insert()) {
          INVARIANT(!last_px || last_px->tuple != it->tuple);
          INVARIANT(it->is_locked());
          INVARIANT(it->get_tuple()->is_locked());
          INVARIANT(it->get_tuple()->is_lock_owner());
          it->entry->set_do_write(); // all inserts are marked do-write
          inserted_last_run = true;
        } else {
          INVARIANT(!it->is_locked());
        }
      }
      if (likely(last_px) &&
          unlikely(!handle_last_tuple_in_group(*last_px, inserted_last_run))) {
        abort_trap((reason = ABORT_REASON_WRITE_NODE_INTERFERENCE));
        goto do_abort;
      }
      commit_tid.first = true;
      PERF_DECL(
          static std::string probe5_name(
            std::string(__PRETTY_FUNCTION__) + std::string(":gen_commit_tid:")));
      ANON_REGION(probe5_name.c_str(), &transaction_base::g_txn_commit_probe5_cg);
      commit_tid.second = cast()->gen_commit_tid(write_dbtuples);
      VERBOSE(std::cerr << "commit tid: " << g_proto_version_str(commit_tid.second) << std::endl);
    } else {
      VERBOSE(std::cerr << "commit tid: <read-only>" << std::endl);
    }

    // do read validation
    {
      PERF_DECL(
          static std::string probe3_name(
            std::string(__PRETTY_FUNCTION__) + std::string(":read_validation:")));
      ANON_REGION(probe3_name.c_str(), &transaction_base::g_txn_commit_probe3_cg);

      // check the nodes we actually read are still the latest version
      if (!read_set.empty()) {
        typename read_set_map::iterator it     = read_set.begin();
        typename read_set_map::iterator it_end = read_set.end();
        for (; it != it_end; ++it) {
          VERBOSE(std::cerr << "validating dbtuple " << util::hexify(it->get_tuple())
                            << " at snapshot_tid "
                            << g_proto_version_str(cast()->snapshot_tid())
                            << std::endl);
          const bool found = sorted_dbtuples_contains(
              write_dbtuples, it->get_tuple());
          if (likely(found ?
                it->get_tuple()->is_latest_version(it->get_tid()) :
                it->get_tuple()->stable_is_latest_version(it->get_tid())))
            continue;

          VERBOSE(std::cerr << "validating dbtuple " << util::hexify(it->get_tuple()) << " at snapshot_tid "
                            << g_proto_version_str(cast()->snapshot_tid()) << " FAILED" << std::endl
                            << "  txn read version: " << g_proto_version_str(it->get_tid()) << std::endl
                            << "  tuple=" << *it->get_tuple() << std::endl);

          //std::cerr << "failed tuple: " << *it->get_tuple() << std::endl;

          abort_trap((reason = ABORT_REASON_READ_NODE_INTEREFERENCE));
          goto do_abort;
        }
      }

      // check btree versions have not changed
      if (!absent_set.empty()) {
        typename absent_set_map::iterator it     = absent_set.begin();
        typename absent_set_map::iterator it_end = absent_set.end();
        for (; it != it_end; ++it) {
          const uint64_t v = concurrent_btree::ExtractVersionNumber(it->first);
          if (unlikely(v != it->second.version)) {
            VERBOSE(std::cerr << "expected node " << util::hexify(it->first) << " at v="
                              << it->second.version << ", got v=" << v << std::endl);
            abort_trap((reason = ABORT_REASON_NODE_SCAN_READ_VERSION_CHANGED));
            goto do_abort;
          }
        }
      }
    }

    // commit actual records
    if (!write_dbtuples.empty()) {
      PERF_DECL(
          static std::string probe4_name(
            std::string(__PRETTY_FUNCTION__) + std::string(":write_records:")));
      ANON_REGION(probe4_name.c_str(), &transaction_base::g_txn_commit_probe4_cg);
      typename write_set_map::iterator it     = write_set.begin();
      typename write_set_map::iterator it_end = write_set.end();
      for (; it != it_end; ++it) {
        if (unlikely(!it->do_write()))
          continue;
        dbtuple * const tuple = it->get_tuple();
        INVARIANT(tuple->is_locked());
        VERBOSE(std::cerr << "writing dbtuple " << util::hexify(tuple)
                          << " at commit_tid " << g_proto_version_str(commit_tid.second)
                          << std::endl);
        if (it->is_insert()) {
          INVARIANT(tuple->version == dbtuple::MAX_TID);
          tuple->version = commit_tid.second; // allows write_record_ret() to succeed
                                              // w/o creating a new chain
        } else {
          tuple->prefetch();
          const dbtuple::write_record_ret ret =
            tuple->write_record_at(
                cast(), commit_tid.second,
                it->get_value(), it->get_writer());
          bool unlock_head = false;
          if (unlikely(ret.head_ != tuple)) {
            // tuple was replaced by ret.head_
            INVARIANT(ret.rest_ == tuple);
            // XXX: write_record_at() should acquire this lock
            ret.head_->lock(true);
            unlock_head = true;
            // need to unlink tuple from underlying btree, replacing
            // with ret.rest_ (atomically)
            typename concurrent_btree::value_type old_v = 0;
            if (it->get_btree()->insert(
                  varkey(it->get_key()), (typename concurrent_btree::value_type) ret.head_, &old_v, NULL))
              // should already exist in tree
              INVARIANT(false);
            INVARIANT(old_v == (typename concurrent_btree::value_type) tuple);
            // we don't RCU free this, because it is now part of the chain
            // (the cleaners will take care of this)
            ++evt_dbtuple_latest_replacement;
          }
          if (unlikely(ret.rest_))
            // spill happened: schedule GC task
            cast()->on_dbtuple_spill(ret.head_, ret.rest_);
          if (!it->get_value())
            // logical delete happened: schedule GC task
            cast()->on_logical_delete(ret.head_, it->get_key(), it->get_btree());
          if (unlikely(unlock_head))
            ret.head_->unlock();
        }
        VERBOSE(std::cerr << "dbtuple " << util::hexify(tuple) << " is_locked? " << tuple->is_locked() << std::endl);
      }
      // unlock
      // NB: we can no longer un-lock after doing the writes above
      for (typename dbtuple_write_info_vec::iterator it = write_dbtuples.begin();
           it != write_dbtuples.end(); ++it) {
        if (it->is_locked())
          it->tuple->unlock();
        else
          INVARIANT(!it->is_insert());
      }
    }
  }
  state = TXN_COMMITED;
  if (commit_tid.first)
    cast()->on_tid_finish(commit_tid.second);
  clear();
  return true;

do_abort:
  // XXX: these values are possibly un-initialized
  if (this->is_snapshot())
    VERBOSE(std::cerr << "aborting txn @ snapshot_tid " << cast()->snapshot_tid() << std::endl);
  else
    VERBOSE(std::cerr << "aborting txn" << std::endl);

  for (typename dbtuple_write_info_vec::iterator it = write_dbtuples.begin();
       it != write_dbtuples.end(); ++it) {
    if (it->is_locked()) {
      if (it->is_insert()) {
        INVARIANT(it->entry);
        this->cleanup_inserted_tuple_marker(
            it->tuple.get(), it->entry->get_key(), it->entry->get_btree());
      }
      // XXX: potential optimization: on unlock() for abort, we don't
      // technically need to change the version number
      it->tuple->unlock();
    } else {
      INVARIANT(!it->is_insert());
    }
  }

  state = TXN_ABRT;
  if (commit_tid.first)
    cast()->on_tid_finish(commit_tid.second);
  clear();
  if (doThrow)
    throw transaction_abort_exception(reason);
  return false;

#ifdef TUPLE_MAGIC
  } catch (dbtuple::magic_failed_exception &) {
    dump_debug_info();
    ALWAYS_ASSERT(false);
  }
#endif
}

template <template <typename> class Protocol, typename Traits>
std::pair< dbtuple *, bool >
transaction<Protocol, Traits>::try_insert_new_tuple(
    concurrent_btree &btr,
    const std::string *key,
    const void *value,
    dbtuple::tuple_writer_t writer)
{
  INVARIANT(key);
  const size_t sz =
    value ? writer(dbtuple::TUPLE_WRITER_COMPUTE_NEEDED,
      value, nullptr, 0) : 0;

  // perf: ~900 tsc/alloc on istc11.csail.mit.edu
  dbtuple * const tuple = dbtuple::alloc_first(sz, true);
  if (value)
    writer(dbtuple::TUPLE_WRITER_DO_WRITE,
        value, tuple->get_value_start(), 0);
  INVARIANT(find_read_set(tuple) == read_set.end());
  INVARIANT(tuple->is_latest());
  INVARIANT(tuple->version == dbtuple::MAX_TID);
  INVARIANT(tuple->is_locked());
  INVARIANT(tuple->is_write_intent());
#ifdef TUPLE_CHECK_KEY
  tuple->key.assign(key->data(), key->size());
  tuple->tree = (void *) &btr;
#endif

  // XXX: underlying btree api should return the existing value if insert
  // fails- this would allow us to avoid having to do another search
  typename concurrent_btree::insert_info_t insert_info;
  if (unlikely(!btr.insert_if_absent(
          varkey(*key), (typename concurrent_btree::value_type) tuple, &insert_info))) {
    VERBOSE(std::cerr << "insert_if_absent failed for key: " << util::hexify(key) << std::endl);
    tuple->clear_latest();
    tuple->unlock();
    dbtuple::release_no_rcu(tuple);
    ++transaction_base::g_evt_dbtuple_write_insert_failed;
    return std::pair< dbtuple *, bool >(nullptr, false);
  }
  VERBOSE(std::cerr << "insert_if_absent suceeded for key: " << util::hexify(key) << std::endl
                    << "  new dbtuple is " << util::hexify(tuple) << std::endl);
  // update write_set
  // too expensive to be practical
  // INVARIANT(find_write_set(tuple) == write_set.end());
  write_set.emplace_back(tuple, key, value, writer, &btr, true);

  // update node #s
  INVARIANT(insert_info.node);
  if (!absent_set.empty()) {
    auto it = absent_set.find(insert_info.node);
    if (it != absent_set.end()) {
      if (unlikely(it->second.version != insert_info.old_version)) {
        abort_trap((reason = ABORT_REASON_WRITE_NODE_INTERFERENCE));
        return std::make_pair(tuple, true);
      }
      VERBOSE(std::cerr << "bump node=" << util::hexify(it->first) << " from v=" << insert_info.old_version
                        << " -> v=" << insert_info.new_version << std::endl);
      // otherwise, bump the version
      it->second.version = insert_info.new_version;
      SINGLE_THREADED_INVARIANT(concurrent_btree::ExtractVersionNumber(it->first) == it->second);
    }
  }
  return std::make_pair(tuple, false);
}

template <template <typename> class Protocol, typename Traits>
template <typename ValueReader>
bool
transaction<Protocol, Traits>::do_tuple_read(
    const dbtuple *tuple, ValueReader &value_reader)
{
  INVARIANT(tuple);
  ++evt_local_search_lookups;

  const bool is_snapshot_txn = is_snapshot();
  const transaction_base::tid_t snapshot_tid = is_snapshot_txn ?
    cast()->snapshot_tid() : static_cast<transaction_base::tid_t>(dbtuple::MAX_TID);
  transaction_base::tid_t start_t = 0;

  if (Traits::read_own_writes) {
    // this is why read_own_writes is not performant, because we have
    // to do linear scan
    auto write_set_it = find_write_set(const_cast<dbtuple *>(tuple));
    if (unlikely(write_set_it != write_set.end())) {
      ++evt_local_search_write_set_hits;
      if (!write_set_it->get_value())
        return false;
      const typename ValueReader::value_type * const px =
        reinterpret_cast<const typename ValueReader::value_type *>(
            write_set_it->get_value());
      value_reader.dup(*px, this->string_allocator());
      return true;
    }
  }

  // do the actual tuple read
  dbtuple::ReadStatus stat;
  {
    PERF_DECL(static std::string probe0_name(std::string(__PRETTY_FUNCTION__) + std::string(":do_read:")));
    ANON_REGION(probe0_name.c_str(), &private_::txn_btree_search_probe0_cg);
    tuple->prefetch();
    stat = tuple->stable_read(snapshot_tid, start_t, value_reader, this->string_allocator(), is_snapshot_txn);
    if (unlikely(stat == dbtuple::READ_FAILED)) {
      const transaction_base::abort_reason r = transaction_base::ABORT_REASON_UNSTABLE_READ;
      abort_impl(r);
      throw transaction_abort_exception(r);
    }
  }
  if (unlikely(!cast()->can_read_tid(start_t))) {
    const transaction_base::abort_reason r = transaction_base::ABORT_REASON_FUTURE_TID_READ;
    abort_impl(r);
    throw transaction_abort_exception(r);
  }
  INVARIANT(stat == dbtuple::READ_EMPTY ||
            stat == dbtuple::READ_RECORD);
  const bool v_empty = (stat == dbtuple::READ_EMPTY);
  if (v_empty)
    ++transaction_base::g_evt_read_logical_deleted_node_search;
  if (!is_snapshot_txn)
    // read-only txns do not need read-set tracking
    // (b/c we know the values are consistent)
    read_set.emplace_back(tuple, start_t);
  return !v_empty;
}

template <template <typename> class Protocol, typename Traits>
void
transaction<Protocol, Traits>::do_node_read(
    const typename concurrent_btree::node_opaque_t *n, uint64_t v)
{
  INVARIANT(n);
  if (is_snapshot())
    return;
  auto it = absent_set.find(n);
  if (it == absent_set.end()) {
    absent_set[n].version = v;
  } else if (it->second.version != v) {
    const transaction_base::abort_reason r =
      transaction_base::ABORT_REASON_NODE_SCAN_READ_VERSION_CHANGED;
    abort_impl(r);
    throw transaction_abort_exception(r);
  }
}

#endif /* _NDB_TXN_IMPL_H_ */