数据库内核月报 － 2018 / 08

 inline bool IsInSnapshot(uint64_t prep_seq, uint64_t snapshot_seq,
                           uint64_t min_uncommitted = 0) const {
    ROCKS_LOG_DETAILS(info_log_,
                      "IsInSnapshot %" PRIu64 " in %" PRIu64
                      " min_uncommitted %" PRIu64,
                      prep_seq, snapshot_seq, min_uncommitted);
    // Here we try to infer the return value without looking into prepare list.
    // This would help avoiding synchronization over a shared map.
    // TODO(myabandeh): optimize this. This sequence of checks must be correct
    // but not necessary efficient
    if (prep_seq == 0) {
      // Compaction will output keys to bottom-level with sequence number 0 if
      // it is visible to the earliest snapshot.
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, 1);
      return true;
    }
    if (snapshot_seq < prep_seq) {
      // snapshot_seq < prep_seq <= commit_seq => snapshot_seq < commit_seq
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, 0);
      return false;
    }
    if (!delayed_prepared_empty_.load(std::memory_order_acquire)) {
      // We should not normally reach here
      WPRecordTick(TXN_PREPARE_MUTEX_OVERHEAD);
      ReadLock rl(&prepared_mutex_);
      ROCKS_LOG_WARN(info_log_, "prepared_mutex_ overhead %" PRIu64,
                     static_cast<uint64_t>(delayed_prepared_.size()));
      if (delayed_prepared_.find(prep_seq) != delayed_prepared_.end()) {
        // Then it is not committed yet
        ROCKS_LOG_DETAILS(info_log_,
                          "IsInSnapshot %" PRIu64 " in %" PRIu64
                          " returns %" PRId32,
                          prep_seq, snapshot_seq, 0);
        return false;
      }
    }
// Note: since min_uncommitted does not include the delayed_prepared_ we
    // should check delayed_prepared_ first before applying this optimization.
    // TODO(myabandeh): include delayed_prepared_ in min_uncommitted
    if (prep_seq < min_uncommitted) {
      ROCKS_LOG_DETAILS(info_log_,
                        "IsInSnapshot %" PRIu64 " in %" PRIu64
                        " returns %" PRId32
                        " because of min_uncommitted %" PRIu64,
                        prep_seq, snapshot_seq, 1, min_uncommitted);
      return true;
    }
    auto indexed_seq = prep_seq % COMMIT_CACHE_SIZE;
    CommitEntry64b dont_care;
    CommitEntry cached;
    bool exist = GetCommitEntry(indexed_seq, &dont_care, &cached);
    if (exist && prep_seq == cached.prep_seq) {
      // It is committed and also not evicted from commit cache
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, cached.commit_seq <= snapshot_seq);
      return cached.commit_seq <= snapshot_seq;
    }
    // else it could be committed but not inserted in the map which could happen
    // after recovery, or it could be committed and evicted by another commit,
    // or never committed.

    // At this point we dont know if it was committed or it is still prepared
    auto max_evicted_seq = max_evicted_seq_.load(std::memory_order_acquire);
    // max_evicted_seq_ when we did GetCommitEntry <= max_evicted_seq now
    if (max_evicted_seq < prep_seq) {
      // Not evicted from cache and also not present, so must be still prepared
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, 0);
      return false;
    }
// When advancing max_evicted_seq_, we move older entires from prepared to
    // delayed_prepared_. Also we move evicted entries from commit cache to
    // old_commit_map_ if it overlaps with any snapshot. Since prep_seq <=
    // max_evicted_seq_, we have three cases: i) in delayed_prepared_, ii) in
    // old_commit_map_, iii) committed with no conflict with any snapshot. Case
    // (i) delayed_prepared_ is checked above
    if (max_evicted_seq < snapshot_seq) {  // then (ii) cannot be the case
      // only (iii) is the case: committed
      // commit_seq <= max_evicted_seq_ < snapshot_seq => commit_seq <
      // snapshot_seq
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, 1);
      return true;
    }
    // else (ii) might be the case: check the commit data saved for this
    // snapshot. If there was no overlapping commit entry, then it is committed
    // with a commit_seq lower than any live snapshot, including snapshot_seq.
    if (old_commit_map_empty_.load(std::memory_order_acquire)) {
      ROCKS_LOG_DETAILS(
          info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
          prep_seq, snapshot_seq, 1);
      return true;
    }
    {
      // We should not normally reach here unless sapshot_seq is old. This is a
      // rare case and it is ok to pay the cost of mutex ReadLock for such old,
      // reading transactions.
      WPRecordTick(TXN_OLD_COMMIT_MAP_MUTEX_OVERHEAD);
      ROCKS_LOG_WARN(info_log_, "old_commit_map_mutex_ overhead");
      ReadLock rl(&old_commit_map_mutex_);
      auto prep_set_entry = old_commit_map_.find(snapshot_seq);
      bool found = prep_set_entry != old_commit_map_.end();
      if (found) {
        auto& vec = prep_set_entry->second;
        found = std::binary_search(vec.begin(), vec.end(), prep_seq);
      }
      if (!found) {
        ROCKS_LOG_DETAILS(info_log_,
                          "IsInSnapshot %" PRIu64 " in %" PRIu64
                          " returns %" PRId32,
                          prep_seq, snapshot_seq, 1);
       return true;
      }
    }
    // (ii) it the case: it is committed but after the snapshot_seq
    ROCKS_LOG_DETAILS(
        info_log_, "IsInSnapshot %" PRIu64 " in %" PRIu64 " returns %" PRId32,
        prep_seq, snapshot_seq, 0);
    return false;
  }

  s = db_->GetImpl(roptions, cf_handle, key, &pinnable_val, &not_used,
                   &callback);
  assert(s.ok() || s.IsNotFound());
  if (s.ok()) {
    s = rollback_batch_->Put(cf_handle, key, pinnable_val);
    assert(s.ok());
  } else if (s.IsNotFound()) {
    // There has been no readable value before txn. By adding a delete we
    // make sure that there will be none afterwards either.
    s = rollback_batch_->Delete(cf_handle, key);
    assert(s.ok());
  } else {
    // Unexpected status. Return it to the user.
  }