用只插入的跳表组织 MemTable 版本记录

84  size_t key_size = key.size();85  size_t val_size = value.size();86  size_t internal_key_size = key_size + 8;87  const size_t encoded_len = VarintLength(internal_key_size) +88                             internal_key_size + VarintLength(val_size) +89                             val_size;90  char* buf = arena_.Allocate(encoded_len);91  char* p = EncodeVarint32(buf, internal_key_size);92  std::memcpy(p, key.data(), key_size);93  p += key_size;94  EncodeFixed64(p, (s << 8) | type);95  p += 8;96  p = EncodeVarint32(p, val_size);97  std::memcpy(p, value.data(), val_size);98  assert(p + val_size == buf + encoded_len);99  table_.Insert(buf);100}

47int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {48  // Order by:49  //    increasing user key (according to user-supplied comparator)50  //    decreasing sequence number51  //    decreasing type (though sequence# should be enough to disambiguate)52  int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));53  if (r == 0) {54    const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);55    const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);56    if (anum > bnum) {57      r = -1;58    } else if (anum < bnum) {59      r = +1;60    }61  }62  return r;63}

259typename SkipList<Key, Comparator>::Node*260SkipList<Key, Comparator>::FindGreaterOrEqual(const Key& key,261                                              Node** prev) const {262  Node* x = head_;263  int level = GetMaxHeight() - 1;264  while (true) {265    Node* next = x->Next(level);266    if (KeyIsAfterNode(key, next)) {267      // Keep searching in this list268      x = next;269    } else {270      if (prev != nullptr) prev[level] = x;271      if (level == 0) {272        return next;273      } else {274        // Switch to next list275        level--;276      }277    }278  }279}

149  // Accessors/mutators for links.  Wrapped in methods so we can150  // add the appropriate barriers as necessary.151  Node* Next(int n) {152    assert(n >= 0);153    // Use an 'acquire load' so that we observe a fully initialized154    // version of the returned Node.155    return next_[n].load(std::memory_order_acquire);156  }157  void SetNext(int n, Node* x) {158    assert(n >= 0);159    // Use a 'release store' so that anybody who reads through this160    // pointer observes a fully initialized version of the inserted node.161    next_[n].store(x, std::memory_order_release);162  }

359  x = NewNode(key, height);360  for (int i = 0; i < height; i++) {361    // NoBarrier_SetNext() suffices since we will add a barrier when362    // we publish a pointer to "x" in prev[i].363    x->NoBarrier_SetNext(i, prev[i]->NoBarrier_Next(i));364    prev[i]->SetNext(i, x);365  }366}

8// Thread safety9// -------------10//11// Writes require external synchronization, most likely a mutex.12// Reads require a guarantee that the SkipList will not be destroyed13// while the read is in progress.  Apart from that, reads progress14// without any internal locking or synchronization.15//16// Invariants:17//18// (1) Allocated nodes are never deleted until the SkipList is19// destroyed.  This is trivially guaranteed by the code since we20// never delete any skip list nodes.21//22// (2) The contents of a Node except for the next/prev pointers are23// immutable after the Node has been linked into the SkipList.24// Only Insert() modifies the list, and it is careful to initialize25// a node and use release-stores to publish the nodes in one or26// more lists.

102bool MemTable::Get(const LookupKey& key, std::string* value, Status* s) {103  Slice memkey = key.memtable_key();104  Table::Iterator iter(&table_);105  iter.Seek(memkey.data());106  if (iter.Valid()) {107    // entry format is:108    //    klength  varint32109    //    userkey  char[klength]110    //    tag      uint64111    //    vlength  varint32112    //    value    char[vlength]113    // Check that it belongs to same user key.  We do not check the114    // sequence number since the Seek() call above should have skipped115    // all entries with overly large sequence numbers.116    const char* entry = iter.key();117    uint32_t key_length;118    const char* key_ptr = GetVarint32Ptr(entry, entry + 5, &key_length);119    if (comparator_.comparator.user_comparator()->Compare(120            Slice(key_ptr, key_length - 8), key.user_key()) == 0) {

122      const uint64_t tag = DecodeFixed64(key_ptr + key_length - 8);123      switch (static_cast<ValueType>(tag & 0xff)) {124        case kTypeValue: {125          Slice v = GetLengthPrefixedSlice(key_ptr + key_length);126          value->assign(v.data(), v.size());127          return true;128        }129        case kTypeDeletion:130          *s = Status::NotFound(Slice());131          return true;132      }133    }134  }135  return false;