txn_proto2_impl.cc

#include <iostream>
#include <thread>
#include <fcntl.h>
#include <unistd.h>
#include <sys/uio.h>
#include <limits.h>
#include <numa.h>

#include "txn_proto2_impl.h"
#include "counter.h"
#include "util.h"

using namespace std;
using namespace util;

                    /** logger subsystem **/
/*{{{*/
bool txn_logger::g_persist = false;
bool txn_logger::g_call_fsync = true;
bool txn_logger::g_use_compression = false;
bool txn_logger::g_fake_writes = false;
size_t txn_logger::g_nworkers = 0;
txn_logger::epoch_array
  txn_logger::per_thread_sync_epochs_[txn_logger::g_nmax_loggers];
aligned_padded_elem<atomic<uint64_t>>
  txn_logger::system_sync_epoch_(0);
percore<txn_logger::persist_ctx>
  txn_logger::g_persist_ctxs;
percore<txn_logger::persist_stats>
  txn_logger::g_persist_stats;
event_counter
  txn_logger::g_evt_log_buffer_epoch_boundary("log_buffer_epoch_boundary");
event_counter
  txn_logger::g_evt_log_buffer_out_of_space("log_buffer_out_of_space");
event_counter
  txn_logger::g_evt_log_buffer_bytes_before_compress("log_buffer_bytes_before_compress");
event_counter
  txn_logger::g_evt_log_buffer_bytes_after_compress("log_buffer_bytes_after_compress");
event_counter
  txn_logger::g_evt_logger_writev_limit_met("logger_writev_limit_met");
event_counter
  txn_logger::g_evt_logger_max_lag_wait("logger_max_lag_wait");
event_avg_counter
  txn_logger::g_evt_avg_log_buffer_compress_time_us("avg_log_buffer_compress_time_us");
event_avg_counter
  txn_logger::g_evt_avg_log_entry_ntxns("avg_log_entry_ntxns_per_entry");
event_avg_counter
  txn_logger::g_evt_avg_logger_bytes_per_writev("avg_logger_bytes_per_writev");
event_avg_counter
  txn_logger::g_evt_avg_logger_bytes_per_sec("avg_logger_bytes_per_sec");

static event_avg_counter
  evt_avg_log_buffer_iov_len("avg_log_buffer_iov_len");

void
txn_logger::Init(
    size_t nworkers,
    const vector<string> &logfiles,
    const vector<vector<unsigned>> &assignments_given,
    vector<vector<unsigned>> *assignments_used,
    bool call_fsync,
    bool use_compression,
    bool fake_writes)
{
  INVARIANT(!g_persist);
  INVARIANT(g_nworkers == 0);
  INVARIANT(nworkers > 0);
  INVARIANT(!logfiles.empty());
  INVARIANT(logfiles.size() <= g_nmax_loggers);
  INVARIANT(!use_compression || g_perthread_buffers > 1); // need 1 as scratch buf
  vector<int> fds;
  for (auto &fname : logfiles) {
    int fd = open(fname.c_str(), O_CREAT|O_WRONLY|O_TRUNC, 0664);
    if (fd == -1) {
      perror("open");
      ALWAYS_ASSERT(false);
    }
    fds.push_back(fd);
  }
  g_persist = true;
  g_call_fsync = call_fsync;
  g_use_compression = use_compression;
  g_fake_writes = fake_writes;
  g_nworkers = nworkers;

  for (size_t i = 0; i < g_nmax_loggers; i++)
    for (size_t j = 0; j < g_nworkers; j++)
      per_thread_sync_epochs_[i].epochs_[j].store(0, memory_order_release);

  vector<thread> writers;
  vector<vector<unsigned>> assignments(assignments_given);

  if (assignments.empty()) {
    // compute assuming homogenous disks
    if (g_nworkers <= fds.size()) {
      // each thread gets its own logging worker
      for (size_t i = 0; i < g_nworkers; i++)
        assignments.push_back({(unsigned) i});
    } else {
      // XXX: currently we assume each logger is equally as fast- we should
      // adjust ratios accordingly for non-homogenous loggers
      const size_t threads_per_logger = g_nworkers / fds.size();
      for (size_t i = 0; i < fds.size(); i++) {
        assignments.emplace_back(
            MakeRange<unsigned>(
              i * threads_per_logger,
              ((i + 1) == fds.size()) ?  g_nworkers : (i + 1) * threads_per_logger));
      }
    }
  }

  INVARIANT(AssignmentsValid(assignments, fds.size(), g_nworkers));

  for (size_t i = 0; i < assignments.size(); i++) {
    writers.emplace_back(
        &txn_logger::writer,
        i, fds[i], assignments[i]);
    writers.back().detach();
  }

  thread persist_thread(&txn_logger::persister, assignments);
  persist_thread.detach();

  if (assignments_used)
    *assignments_used = assignments;
}

void
txn_logger::persister(
    vector<vector<unsigned>> assignments)
{
  timer loop_timer;
  for (;;) {
    const uint64_t last_loop_usec = loop_timer.lap();
    const uint64_t delay_time_usec = ticker::tick_us;
    if (last_loop_usec < delay_time_usec) {
      const uint64_t sleep_ns = (delay_time_usec - last_loop_usec) * 1000;
      struct timespec t;
      t.tv_sec  = sleep_ns / ONE_SECOND_NS;
      t.tv_nsec = sleep_ns % ONE_SECOND_NS;
      nanosleep(&t, nullptr);
    }
    advance_system_sync_epoch(assignments);
  }
}

void
txn_logger::advance_system_sync_epoch(
    const vector<vector<unsigned>> &assignments)
{
  uint64_t min_so_far = numeric_limits<uint64_t>::max();
  const uint64_t best_tick_ex =
    ticker::s_instance.global_current_tick();
  // special case 0
  const uint64_t best_tick_inc =
    best_tick_ex ? (best_tick_ex - 1) : 0;

  for (size_t i = 0; i < assignments.size(); i++)
    for (auto j : assignments[i])
      for (size_t k = j; k < NMAXCORES; k += g_nworkers) {
        persist_ctx &ctx = persist_ctx_for(k, INITMODE_NONE);
        // we need to arbitrarily advance threads which are not "doing
        // anything", so they don't drag down the persistence of the system. if
        // we can see that a thread is NOT in a guarded section AND its
        // core->logger queue is empty, then that means we can advance its sync
        // epoch up to best_tick_inc, b/c it is guaranteed that the next time
        // it does any actions will be in epoch > best_tick_inc
        if (!ctx.persist_buffers_.peek()) {
          spinlock &l = ticker::s_instance.lock_for(k);
          if (!l.is_locked()) {
            bool did_lock = false;
            for (size_t c = 0; c < 3; c++) {
              if (l.try_lock()) {
                did_lock = true;
                break;
              }
            }
            if (did_lock) {
              if (!ctx.persist_buffers_.peek()) {
                min_so_far = min(min_so_far, best_tick_inc);
                per_thread_sync_epochs_[i].epochs_[k].store(
                    best_tick_inc, memory_order_release);
                l.unlock();
                continue;
              }
              l.unlock();
            }
          }
        }
        min_so_far = min(
            per_thread_sync_epochs_[i].epochs_[k].load(
              memory_order_acquire),
            min_so_far);
      }

  const uint64_t syssync =
    system_sync_epoch_->load(memory_order_acquire);

  INVARIANT(min_so_far < numeric_limits<uint64_t>::max());
  INVARIANT(syssync <= min_so_far);

  // need to aggregate from [syssync + 1, min_so_far]
  const uint64_t now_us = timer::cur_usec();
  for (size_t i = 0; i < g_persist_stats.size(); i++) {
    auto &ps = g_persist_stats[i];
    for (uint64_t e = syssync + 1; e <= min_so_far; e++) {
        auto &pes = ps.d_[e % g_max_lag_epochs];
        const uint64_t ntxns_in_epoch = pes.ntxns_.load(memory_order_acquire);
        const uint64_t start_us = pes.earliest_start_us_.load(memory_order_acquire);
        INVARIANT(now_us >= start_us);
        non_atomic_fetch_add(ps.ntxns_persisted_, ntxns_in_epoch);
        non_atomic_fetch_add(
            ps.latency_numer_,
            (now_us - start_us) * ntxns_in_epoch);
        pes.ntxns_.store(0, memory_order_release);
        pes.earliest_start_us_.store(0, memory_order_release);
    }
  }

  system_sync_epoch_->store(min_so_far, memory_order_release);
}

void
txn_logger::writer(
    unsigned id, int fd,
    vector<unsigned> assignment)
{

  if (g_pin_loggers_to_numa_nodes) {
    ALWAYS_ASSERT(!numa_run_on_node(id % numa_num_configured_nodes()));
    ALWAYS_ASSERT(!sched_yield());
  }

  vector<iovec> iovs(
      min(size_t(IOV_MAX), g_nworkers * g_perthread_buffers));
  vector<pbuffer *> pxs;
  timer loop_timer;

  // XXX: sense is not useful for now, unless we want to
  // fsync in the background...
  bool sense = false; // cur is at sense, prev is at !sense
  uint64_t epoch_prefixes[2][NMAXCORES];

  NDB_MEMSET(&epoch_prefixes[0], 0, sizeof(epoch_prefixes[0]));
  NDB_MEMSET(&epoch_prefixes[1], 0, sizeof(epoch_prefixes[1]));

  // NOTE: a core id in the persistence system really represets
  // all cores in the regular system modulo g_nworkers
  size_t nbufswritten = 0, nbyteswritten = 0;
  for (;;) {

    const uint64_t last_loop_usec = loop_timer.lap();
    const uint64_t delay_time_usec = ticker::tick_us;
    // don't allow this loop to proceed less than an epoch's worth of time,
    // so we can batch IO
    if (last_loop_usec < delay_time_usec && nbufswritten < iovs.size()) {
      const uint64_t sleep_ns = (delay_time_usec - last_loop_usec) * 1000;
      struct timespec t;
      t.tv_sec  = sleep_ns / ONE_SECOND_NS;
      t.tv_nsec = sleep_ns % ONE_SECOND_NS;
      nanosleep(&t, nullptr);
    }

    // we need g_persist_stats[cur_sync_epoch_ex % g_nmax_loggers]
    // to remain untouched (until the syncer can catch up), so we
    // cannot read any buffers with epoch >=
    // (cur_sync_epoch_ex + g_max_lag_epochs)
    const uint64_t cur_sync_epoch_ex =
      system_sync_epoch_->load(memory_order_acquire) + 1;
    nbufswritten = nbyteswritten = 0;
    for (auto idx : assignment) {
      INVARIANT(idx >= 0 && idx < g_nworkers);
      for (size_t k = idx; k < NMAXCORES; k += g_nworkers) {
        persist_ctx &ctx = persist_ctx_for(k, INITMODE_NONE);
        ctx.persist_buffers_.peekall(pxs);
        for (auto px : pxs) {
          INVARIANT(px);
          INVARIANT(!px->io_scheduled_);
          INVARIANT(nbufswritten <= iovs.size());
          INVARIANT(px->header()->nentries_);
          INVARIANT(px->core_id_ == k);
          if (nbufswritten == iovs.size()) {
            ++g_evt_logger_writev_limit_met;
            goto process;
          }
          if (transaction_proto2_static::EpochId(px->header()->last_tid_) >=
              cur_sync_epoch_ex + g_max_lag_epochs) {
            ++g_evt_logger_max_lag_wait;
            break;
          }
          iovs[nbufswritten].iov_base = (void *) &px->buf_start_[0];

#ifdef LOGGER_UNSAFE_REDUCE_BUFFER_SIZE
  #define PXLEN(px) (((px)->curoff_ < 4) ? (px)->curoff_ : ((px)->curoff_ / 4))
#else
  #define PXLEN(px) ((px)->curoff_)
#endif

          const size_t pxlen = PXLEN(px);

          iovs[nbufswritten].iov_len = pxlen;
          evt_avg_log_buffer_iov_len.offer(pxlen);
          px->io_scheduled_ = true;
          nbufswritten++;
          nbyteswritten += pxlen;

#ifdef CHECK_INVARIANTS
          auto last_tid_cid = transaction_proto2_static::CoreId(px->header()->last_tid_);
          auto px_cid = px->core_id_;
          if (last_tid_cid != px_cid) {
            cerr << "header: " << *px->header() << endl;
            cerr << g_proto_version_str(last_tid_cid) << endl;
            cerr << "last_tid_cid: " << last_tid_cid << endl;
            cerr << "px_cid: " << px_cid << endl;
          }
#endif

          const uint64_t px_epoch =
            transaction_proto2_static::EpochId(px->header()->last_tid_);
          INVARIANT(
              transaction_proto2_static::CoreId(px->header()->last_tid_) ==
              px->core_id_);
          INVARIANT(epoch_prefixes[sense][k] <= px_epoch);
          INVARIANT(px_epoch > 0);
          epoch_prefixes[sense][k] = px_epoch - 1;
          auto &pes = g_persist_stats[k].d_[px_epoch % g_max_lag_epochs];
          if (!pes.ntxns_.load(memory_order_acquire))
            pes.earliest_start_us_.store(px->earliest_start_us_, memory_order_release);
          non_atomic_fetch_add(pes.ntxns_, px->header()->nentries_);
          g_evt_avg_log_entry_ntxns.offer(px->header()->nentries_);
        }
      }
    }

  process:
    if (!nbufswritten) {
      // XXX: should probably sleep here
      nop_pause();
      continue;
    }

    const bool dosense = sense;

    if (!g_fake_writes) {
#ifdef ENABLE_EVENT_COUNTERS
      timer write_timer;
#endif
      const ssize_t ret = writev(fd, &iovs[0], nbufswritten);
      if (unlikely(ret == -1)) {
        perror("writev");
        ALWAYS_ASSERT(false);
      }

      if (g_call_fsync) {
        const int fret = fdatasync(fd);
        if (unlikely(fret == -1)) {
          perror("fdatasync");
          ALWAYS_ASSERT(false);
        }
      }

#ifdef ENABLE_EVENT_COUNTERS
      {
        g_evt_avg_logger_bytes_per_writev.offer(nbyteswritten);
        const double bytes_per_sec =
          double(nbyteswritten)/(write_timer.lap_ms() / 1000.0);
        g_evt_avg_logger_bytes_per_sec.offer(bytes_per_sec);
      }
#endif
    }

    // update metadata from previous write
    //
    // return all buffers that have been io_scheduled_ - we can do this as
    // soon as write returns. we take care to return to the proper buffer
    epoch_array &ea = per_thread_sync_epochs_[id];
    for (auto idx: assignment) {
      for (size_t k = idx; k < NMAXCORES; k += g_nworkers) {
        const uint64_t x0 = ea.epochs_[k].load(memory_order_acquire);
        const uint64_t x1 = epoch_prefixes[dosense][k];
        if (x1 > x0)
          ea.epochs_[k].store(x1, memory_order_release);

        persist_ctx &ctx = persist_ctx_for(k, INITMODE_NONE);
        pbuffer *px, *px0;
        while ((px = ctx.persist_buffers_.peek()) && px->io_scheduled_) {
#ifdef LOGGER_STRIDE_OVER_BUFFER
          {
            const size_t pxlen = PXLEN(px);
            const size_t stridelen = 1;
            for (size_t p = 0; p < pxlen; p += stridelen)
              if ((&px->buf_start_[0])[p] & 0xF)
                non_atomic_fetch_add(ea.dummy_work_, 1UL);
          }
#endif
          px0 = ctx.persist_buffers_.deq();
          INVARIANT(px == px0);
          INVARIANT(px->header()->nentries_);
          px0->reset();
          INVARIANT(ctx.init_);
          INVARIANT(px0->core_id_ == k);
          ctx.all_buffers_.enq(px0);
        }
      }
    }

    // bump the sense
    sense = !sense;
  }
}

tuple<uint64_t, uint64_t, double>
txn_logger::compute_ntxns_persisted_statistics()
{
  uint64_t acc = 0, acc1 = 0, acc2 = 0;
  uint64_t num = 0;
  for (size_t i = 0; i < g_persist_stats.size(); i++) {
    acc  += g_persist_stats[i].ntxns_persisted_.load(memory_order_acquire);
    acc1 += g_persist_stats[i].ntxns_pushed_.load(memory_order_acquire);
    acc2 += g_persist_stats[i].ntxns_committed_.load(memory_order_acquire);
    num  += g_persist_stats[i].latency_numer_.load(memory_order_acquire);
  }
  INVARIANT(acc <= acc1);
  INVARIANT(acc1 <= acc2);
  if (acc == 0)
    return make_tuple(0, acc1, 0.0);
  return make_tuple(acc, acc1, double(num)/double(acc));
}

void
txn_logger::clear_ntxns_persisted_statistics()
{
  for (size_t i = 0; i < g_persist_stats.size(); i++) {
    auto &ps = g_persist_stats[i];
    ps.ntxns_persisted_.store(0, memory_order_release);
    ps.ntxns_pushed_.store(0, memory_order_release);
    ps.ntxns_committed_.store(0, memory_order_release);
    ps.latency_numer_.store(0, memory_order_release);
    for (size_t e = 0; e < g_max_lag_epochs; e++) {
      auto &pes = ps.d_[e];
      pes.ntxns_.store(0, memory_order_release);
      pes.earliest_start_us_.store(0, memory_order_release);
    }
  }
}

void
txn_logger::wait_for_idle_state()
{
  for (size_t i = 0; i < NMAXCORES; i++) {
    persist_ctx &ctx = persist_ctx_for(i, INITMODE_NONE);
    if (!ctx.init_)
      continue;
    pbuffer *px;
    while (!(px = ctx.all_buffers_.peek()) || px->header()->nentries_)
      nop_pause();
    while (ctx.persist_buffers_.peek())
      nop_pause();
  }
}

void
txn_logger::wait_until_current_point_persisted()
{
  const uint64_t e = ticker::s_instance.global_current_tick();
  cerr << "waiting for system_sync_epoch_="
       << system_sync_epoch_->load(memory_order_acquire)
       << " to be < e=" << e << endl;
  while (system_sync_epoch_->load(memory_order_acquire) < e)
    nop_pause();
}
/*}}}*/

                /** garbage collection subsystem **/

static event_counter evt_local_chain_cleanups("local_chain_cleanups");
static event_counter evt_try_delete_unlinks("try_delete_unlinks");
static event_avg_counter evt_avg_time_inbetween_ro_epochs_usec(
    "avg_time_inbetween_ro_epochs_usec");

void
transaction_proto2_static::InitGC()
{
  g_flags->g_gc_init.store(true, memory_order_release);
}

static void
sleep_ro_epoch()
{
  const uint64_t sleep_ns = transaction_proto2_static::ReadOnlyEpochUsec * 1000;
  struct timespec t;
  t.tv_sec  = sleep_ns / ONE_SECOND_NS;
  t.tv_nsec = sleep_ns % ONE_SECOND_NS;
  nanosleep(&t, nullptr);
}

void
transaction_proto2_static::PurgeThreadOutstandingGCTasks()
{
#ifdef PROTO2_CAN_DISABLE_GC
  if (!IsGCEnabled())
    return;
#endif
  INVARIANT(!rcu::s_instance.in_rcu_region());
  threadctx &ctx = g_threadctxs.my();
  uint64_t e;
  if (!ctx.queue_.get_latest_epoch(e))
    return;
  // wait until we can clean up e
  for (;;) {
    const uint64_t last_tick_ex = ticker::s_instance.global_last_tick_exclusive();
    const uint64_t ro_tick_ex = to_read_only_tick(last_tick_ex);
    if (unlikely(!ro_tick_ex)) {
      sleep_ro_epoch();
      continue;
    }
    const uint64_t ro_tick_geq = ro_tick_ex - 1;
    if (ro_tick_geq < e) {
      sleep_ro_epoch();
      continue;
    }
    break;
  }
  clean_up_to_including(ctx, e);
  INVARIANT(ctx.queue_.empty());
}

//#ifdef CHECK_INVARIANTS
//// make sure hidden is blocked by version e, when traversing from start
//static bool
//IsBlocked(dbtuple *start, dbtuple *hidden, uint64_t e)
//{
//  dbtuple *c = start;
//  while (c) {
//    if (c == hidden)
//      return false;
//    if (c->is_not_behind(e))
//      // blocked
//      return true;
//    c = c->next;
//  }
//  ALWAYS_ASSERT(false); // hidden should be found on chain
//}
//#endif

void
transaction_proto2_static::clean_up_to_including(threadctx &ctx, uint64_t ro_tick_geq)
{
  INVARIANT(!rcu::s_instance.in_rcu_region());
  INVARIANT(ctx.last_reaped_epoch_ <= ro_tick_geq);
  INVARIANT(ctx.scratch_.empty());
  if (ctx.last_reaped_epoch_ == ro_tick_geq)
    return;

#ifdef ENABLE_EVENT_COUNTERS
  const uint64_t now = timer::cur_usec();
  if (ctx.last_reaped_timestamp_us_ > 0) {
    const uint64_t diff = now - ctx.last_reaped_timestamp_us_;
    evt_avg_time_inbetween_ro_epochs_usec.offer(diff);
  }
  ctx.last_reaped_timestamp_us_ = now;
#endif
  ctx.last_reaped_epoch_ = ro_tick_geq;

#ifdef CHECK_INVARIANTS
  const uint64_t last_tick_ex = ticker::s_instance.global_last_tick_exclusive();
  INVARIANT(last_tick_ex);
  const uint64_t last_consistent_tid = ComputeReadOnlyTid(last_tick_ex - 1);
  const uint64_t computed_last_tick_ex = ticker::s_instance.compute_global_last_tick_exclusive();
  INVARIANT(last_tick_ex <= computed_last_tick_ex);
  INVARIANT(to_read_only_tick(last_tick_ex) > ro_tick_geq);
#endif

  // XXX: hacky
  char rcu_guard[sizeof(scoped_rcu_base<false>)] = {0};
  const size_t max_niters_with_rcu = 128;
#define ENTER_RCU() \
    do { \
      new (&rcu_guard[0]) scoped_rcu_base<false>(); \
    } while (0)
#define EXIT_RCU() \
    do { \
      scoped_rcu_base<false> *px = (scoped_rcu_base<false> *) &rcu_guard[0]; \
      px->~scoped_rcu_base<false>(); \
    } while (0)

  ctx.scratch_.empty_accept_from(ctx.queue_, ro_tick_geq);
  ctx.scratch_.transfer_freelist(ctx.queue_);
  px_queue &q = ctx.scratch_;
  if (q.empty())
    return;
  bool in_rcu = false;
  size_t niters_with_rcu = 0, n = 0;
  for (auto it = q.begin(); it != q.end(); ++it, ++n, ++niters_with_rcu) {
    auto &delent = *it;
    INVARIANT(delent.tuple()->opaque.load(std::memory_order_acquire) == 1);
    if (!delent.key_.get_flags()) {
      // guaranteed to be gc-able now (even w/o RCU)
#ifdef CHECK_INVARIANTS
      if (delent.trigger_tid_ > last_consistent_tid /*|| !IsBlocked(delent.tuple_ahead_, delent.tuple(), last_consistent_tid) */) {
        cerr << "tuple ahead     : " << g_proto_version_str(delent.tuple_ahead_->version) << endl;
        cerr << "tuple ahead     : " << *delent.tuple_ahead_ << endl;
        cerr << "trigger tid     : " << g_proto_version_str(delent.trigger_tid_) << endl;
        cerr << "tuple           : " << g_proto_version_str(delent.tuple()->version) << endl;
        cerr << "last_consist_tid: " << g_proto_version_str(last_consistent_tid) << endl;
        cerr << "last_tick_ex    : " << last_tick_ex << endl;
        cerr << "ro_tick_geq     : " << ro_tick_geq << endl;
        cerr << "rcu_block_tick  : " << it.tick() << endl;
      }
      INVARIANT(delent.trigger_tid_ <= last_consistent_tid);
      delent.tuple()->opaque.store(0, std::memory_order_release);
#endif
      dbtuple::release_no_rcu(delent.tuple());
    } else {
      INVARIANT(!delent.tuple_ahead_);
      INVARIANT(delent.btr_);
      // check if an element preceeds the (deleted) tuple before doing the delete
      ::lock_guard<dbtuple> lg_tuple(delent.tuple(), false);
#ifdef CHECK_INVARIANTS
      if (!delent.tuple()->is_not_behind(last_consistent_tid)) {
        cerr << "trigger tid     : " << g_proto_version_str(delent.trigger_tid_) << endl;
        cerr << "tuple           : " << g_proto_version_str(delent.tuple()->version) << endl;
        cerr << "last_consist_tid: " << g_proto_version_str(last_consistent_tid) << endl;
        cerr << "last_tick_ex    : " << last_tick_ex << endl;
        cerr << "ro_tick_geq     : " << ro_tick_geq << endl;
        cerr << "rcu_block_tick  : " << it.tick() << endl;
      }
      INVARIANT(delent.tuple()->version == delent.trigger_tid_);
      INVARIANT(delent.tuple()->is_not_behind(last_consistent_tid));
      INVARIANT(delent.tuple()->is_deleting());
#endif
      if (unlikely(!delent.tuple()->is_latest())) {
        // requeue it up, except this time as a regular delete
        const uint64_t my_ro_tick = to_read_only_tick(
            ticker::s_instance.global_current_tick());
        ctx.queue_.enqueue(
            delete_entry(
              nullptr,
              MakeTid(CoreMask, NumIdMask >> NumIdShift, (my_ro_tick + 1) * ReadOnlyEpochMultiplier - 1),
              delent.tuple(),
              marked_ptr<string>(),
              nullptr),
            my_ro_tick);
        ++g_evt_proto_gc_delete_requeue;
        // reclaim string ptrs
        string *spx = delent.key_.get();
        if (unlikely(spx))
          ctx.pool_.emplace_back(spx);
        continue;
      }
#ifdef CHECK_INVARIANTS
      delent.tuple()->opaque.store(0, std::memory_order_release);
#endif
      // if delent.key_ is nullptr, then the key is stored in the tuple
      // record storage location, and the size field contains the length of
      // the key
      //
      // otherwise, delent.key_ is a pointer to a string containing the
      // key
      varkey k;
      string *spx = delent.key_.get();
      if (likely(!spx)) {
        k = varkey(delent.tuple()->get_value_start(), delent.tuple()->size);
      } else {
        k = varkey(*spx);
        ctx.pool_.emplace_back(spx);
      }

      if (!in_rcu) {
        ENTER_RCU();
        niters_with_rcu = 0;
        in_rcu = true;
      }
      typename concurrent_btree::value_type removed = 0;
      const bool did_remove = delent.btr_->remove(k, &removed);
      ALWAYS_ASSERT(did_remove);
      INVARIANT(removed == (typename concurrent_btree::value_type) delent.tuple());
      delent.tuple()->clear_latest();
      dbtuple::release(delent.tuple()); // rcu free it
    }

    if (in_rcu && niters_with_rcu >= max_niters_with_rcu) {
      EXIT_RCU();
      niters_with_rcu = 0;
      in_rcu = false;
    }
  }
  q.clear();
  g_evt_avg_proto_gc_queue_len.offer(n);

  if (in_rcu)
    EXIT_RCU();
  INVARIANT(!rcu::s_instance.in_rcu_region());
}

aligned_padded_elem<transaction_proto2_static::hackstruct>
  transaction_proto2_static::g_hack;
aligned_padded_elem<transaction_proto2_static::flags>
  transaction_proto2_static::g_flags;
percore_lazy<transaction_proto2_static::threadctx>
  transaction_proto2_static::g_threadctxs;
event_counter
  transaction_proto2_static::g_evt_worker_thread_wait_log_buffer(
      "worker_thread_wait_log_buffer");
event_counter
  transaction_proto2_static::g_evt_dbtuple_no_space_for_delkey(
      "dbtuple_no_space_for_delkey");
event_counter
  transaction_proto2_static::g_evt_proto_gc_delete_requeue(
      "proto_gc_delete_requeue");
event_avg_counter
  transaction_proto2_static::g_evt_avg_log_entry_size(
      "avg_log_entry_size");
event_avg_counter
  transaction_proto2_static::g_evt_avg_proto_gc_queue_len(
      "avg_proto_gc_queue_len");