folly::ConcurrentSkipList 详解

这个函数中调用的几个方法：

findInsertionPointGetMaxLayer：首先它返回 skiplist 的高度，然后按照 right down 的方式查找节点，不同的是在查找过程中会保留前继指针 preds[] 和后继指针 succs[]。

// find node for insertion/deleting
int findInsertionPointGetMaxLayer(
    const value_type& data,
    NodeType* preds[],
    NodeType* succs[],
    int* max_layer) const {
  *max_layer = maxLayer();
  return findInsertionPoint(
      head_.load(std::memory_order_consume), *max_layer, data, preds, succs);
}
 
static int findInsertionPoint(
    NodeType* cur,
    int cur_layer,
    const value_type& data,
    NodeType* preds[],
    NodeType* succs[]) {
  int foundLayer = -1;
  NodeType* pred = cur;
  NodeType* foundNode = nullptr;
  for (int layer = cur_layer; layer >= 0; --layer) {
    NodeType* node = pred->skip(layer);
    while (greater(data, node)) {
      pred = node;
      node = node->skip(layer);
    }
    if (foundLayer == -1 && !less(data, node)) { // the two keys equal
      foundLayer = layer;
      foundNode = node;
    }
    preds[layer] = pred;
 
    // if found, succs[0..foundLayer] need to point to the cached foundNode,
    // as foundNode might be deleted at the same time thus pred->skip() can
    // return nullptr or another node.
    succs[layer] = foundNode ? foundNode : node;
  }
  return foundLayer;
}

SkipListRandomHeight::getHeight 和 SkipListRandomHeight::getSizeLimit

在 SkipListRandomHeight 构造的时候会初始化两张表：lookupTable_ 是高度的概率表，sizeLimitTable_ 是 SkipList 的高度对应的最大的 list size。

getHeight 方法用随机函数生成一个 0~1 之间的 double 值 p，然后在 lookupTable 中找比 p 大的值对应的表索引 i，找到后获得的高度就是 i+1。getSizeLimit 方法也类似，以参数 height 为 sizeLimitTable 的索引，返回对应高度的 sizeLimit。

lockNodesForChange 的实现：

// lock all the necessary nodes for changing (adding or removing) the list.
// returns true if all the lock acquried successfully and the related nodes
// are all validate (not in certain pending states), false otherwise.
bool lockNodesForChange(
    int nodeHeight,
    ScopedLocker guards[MAX_HEIGHT],
    NodeType* preds[MAX_HEIGHT],    // 插入或删除节点的前继
    NodeType* succs[MAX_HEIGHT],    // 插入或删除节点的后继
    bool adding = true) {
    // adding 为 true 表明该函数是在 add 里被调用，否则是在 remove 里被调用
  NodeType *pred, *succ, *prevPred = nullptr;
  bool valid = true;
  for (int layer = 0; valid && layer < nodeHeight; ++layer) {
    pred = preds[layer];
    DCHECK(pred != nullptr) << "layer=" << layer << " height=" << height()
                            << " nodeheight=" << nodeHeight;
    succ = succs[layer];
    if (pred != prevPred) {
    // 可能连续多层的前继指针都是一个节点，这里可以避免多次上锁
      guards[layer] = pred->acquireGuard();
      prevPred = pred;
    }
 
    // 对于 remove 来说只要判断前继没有被删除并且前继的后继是后继节点即可
    valid = !pred->markedForRemoval() &&
        pred->skip(layer) == succ; // check again after locking
 
    // 对于 adding 来说还需要判断后继节点没有被删除
    if (adding) { // when adding a node, the succ shouldn't be going away
      valid = valid && (succ == nullptr || !succ->markedForRemoval());
    }
  }
 
  return valid;
}

growHeight 的实现：

void growHeight(int height) {
  NodeType* oldHead = head_.load(std::memory_order_consume);
  if (oldHead->height() >= height) { // someone else already did this
    return;
  }
 
  // 生成新的 head 节点，height 参数就是在原来的 heigth 基础上加1
  NodeType* newHead =
      NodeType::create(recycler_.alloc(), height, value_type(), true);
 
  { // need to guard the head node in case others are adding/removing
    // nodes linked to the head.
    ScopedLocker g = oldHead->acquireGuard();
    // 从 oldHead 中把数据拷贝过来，类似拷贝构造函数
    newHead->copyHead(oldHead);
    NodeType* expected = oldHead;
    // 原子替换 head_ 指针指向 newHead
    if (!head_.compare_exchange_strong(
            expected, newHead, std::memory_order_release)) {
      // if someone has already done the swap, just return.
      NodeType::destroy(recycler_.alloc(), newHead);
      return;
    }
    oldHead->setMarkedForRemoval();
  }
 
  recycle(oldHead);
}

SkipList 删除

Accessor 的接口
bool remove(const key_type& data) { return sl_->remove(data); }
size_t erase(const key_type& data) { return remove(data); }
 
具体调用
bool remove(const value_type& data) {
  NodeType* nodeToDelete = nullptr;
  ScopedLocker nodeGuard;
  bool isMarked = false;
  int nodeHeight = 0;
  NodeType *preds[MAX_HEIGHT], *succs[MAX_HEIGHT];
 
  while (true) {
    int max_layer = 0;
    int layer = findInsertionPointGetMaxLayer(data, preds, succs, &max_layer);
    if (!isMarked && (layer < 0 || !okToDelete(succs[layer], layer))) {
      return false;
    }
 
    if (!isMarked) {
      nodeToDelete = succs[layer];
      nodeHeight = nodeToDelete->height();
      nodeGuard = nodeToDelete->acquireGuard();
      if (nodeToDelete->markedForRemoval()) {
        return false;
      }
      nodeToDelete->setMarkedForRemoval();
      isMarked = true;
    }
 
    // acquire pred locks from bottom layer up
    ScopedLocker guards[MAX_HEIGHT];
    if (!lockNodesForChange(nodeHeight, guards, preds, succs, false)) {
      continue; // this will unlock all the locks
    }
 
    for (int k = nodeHeight - 1; k >= 0; --k) {
      preds[k]->setSkip(k, nodeToDelete->skip(k));
    }
 
    incrementSize(-1);
    break;
  }
  recycle(nodeToDelete);
  return true;
}

注意：recycler 负责回收被删除的节点，但其实它只是把节点加入一个 vector 中，然后在 recycler 对象析构或者显示调用 release 方法时才会去释放这些节点。

void recycle(NodeType* node) { recycler_.add(node); }
detail::NodeRecycler recycler_;
 
template <typename NodeType, typename NodeAlloc>
class NodeRecycler<
    NodeType,
    NodeAlloc,
    typename std::enable_if<
        !NodeType::template DestroyIsNoOp::value>::type> {
 public:
  explicit NodeRecycler(const NodeAlloc& alloc)
      : refs_(0), dirty_(false), alloc_(alloc) {
    lock_.init();
  }
 
  explicit NodeRecycler() : refs_(0), dirty_(false) { lock_.init(); }
 
  ~NodeRecycler() {
    CHECK_EQ(refs(), 0);
    if (nodes_) {
      for (auto& node : *nodes_) {
        NodeType::destroy(alloc_, node);
      }
    }
  }
 
  void add(NodeType* node) {
    std::lock_guard g(lock_);
    if (nodes_.get() == nullptr) {
      nodes_ = std::make_unique>(1, node);
    } else {
      nodes_->push_back(node);
    }
    DCHECK_GT(refs(), 0);
    dirty_.store(true, std::memory_order_relaxed);
  }
 
  int addRef() { return refs_.fetch_add(1, std::memory_order_relaxed); }
 
  int releaseRef() {
    // We don't expect to clean the recycler immediately everytime it is OK
    // to do so. Here, it is possible that multiple accessors all release at
    // the same time but nobody would clean the recycler here. If this
    // happens, the recycler will usually still get cleaned when
    // such a race doesn't happen. The worst case is the recycler will
    // eventually get deleted along with the skiplist.
    if (LIKELY(!dirty_.load(std::memory_order_relaxed) || refs() > 1)) {
      return refs_.fetch_add(-1, std::memory_order_relaxed);
    }
 
    std::unique_ptr> newNodes;
    {
      std::lock_guard g(lock_);
      if (nodes_.get() == nullptr || refs() > 1) {
        return refs_.fetch_add(-1, std::memory_order_relaxed);
      }
      // once refs_ reaches 1 and there is no other accessor, it is safe to
      // remove all the current nodes in the recycler, as we already acquired
      // the lock here so no more new nodes can be added, even though new
      // accessors may be added after that.
      newNodes.swap(nodes_);
      dirty_.store(false, std::memory_order_relaxed);
    }
 
    // TODO(xliu) should we spawn a thread to do this when there are large
    // number of nodes in the recycler?
    for (auto& node : *newNodes) {
      NodeType::destroy(alloc_, node);
    }
 
    // decrease the ref count at the very end, to minimize the
    // chance of other threads acquiring lock_ to clear the deleted
    // nodes again.
    return refs_.fetch_add(-1, std::memory_order_relaxed);
  }
 
  NodeAlloc& alloc() { return alloc_; }
 
 private:
  int refs() const { return refs_.load(std::memory_order_relaxed); }
 
  std::unique_ptr> nodes_;
  std::atomic<int32_t> refs_; // current number of visitors to the list
  std::atomic<bool> dirty_; // whether *nodes_ is non-empty
  MicroSpinLock lock_; // protects access to *nodes_
  NodeAlloc alloc_;
};