fixing double destruction of CSL::data_type

Summary:
the currently code calls both ~SkipListNode() and node->data_.~value_type()
causes double destructing the object.

Test Plan: adding dihde's testing code to a test case

Reviewed By: emailweixu@fb.com

FB internal diff: D612289

folly/ConcurrentSkipList-inl.h

@@ -42,40 +42,28 @@ class SkipListNode : boost::noncopyable {
  public:
   typedef T value_type;
 
-  static SkipListNode* create(int height,
-      const value_type& data, bool isHead = false) {
+  template<typename U,
+    typename=typename std::enable_if<std::is_convertible<U, T>::value>::type>
+  static SkipListNode* create(int height, U&& data, bool isHead = false) {
     DCHECK(height >= 1 && height < 64) << height;
 
-    size_t size = sizeof(SkipListNode) + height * sizeof(SkipListNode*);
+    size_t size = sizeof(SkipListNode) +
+      height * sizeof(std::atomic<SkipListNode*>);
     auto* node = static_cast<SkipListNode*>(malloc(size));
-    new (node) SkipListNode(height);
-
-    node->spinLock_.init();
-    node->setFlags(0);
-
-    if (isHead) {
-      node->setIsHeadNode();
-    } else {
-      new (&(node->data_)) value_type(data);
-    }
+    // do placement new
+    new (node) SkipListNode(height, std::forward<U>(data), isHead);
     return node;
   }
 
   static void destroy(SkipListNode* node) {
-    if (!node->isHeadNode()) {
-      node->data_.~value_type();
-    }
     node->~SkipListNode();
     free(node);
   }
 
-  // assuming lock acquired
-  SkipListNode* promoteFrom(const SkipListNode* node) {
+  // copy the head node to a new head node assuming lock acquired
+  SkipListNode* copyHead(SkipListNode* node) {
     DCHECK(node != nullptr && height_ > node->height_);
     setFlags(node->getFlags());
-    if (!isHeadNode()) {
-      new (&(data_)) value_type(node->data());
-    }
     for (int i = 0; i < node->height_; ++i) {
       setSkip(i, node->skip(i));
     }
@@ -125,14 +113,22 @@ class SkipListNode : boost::noncopyable {
   }
 
  private:
-  ~SkipListNode() {
+  // Note! this can only be called from create() as a placement new.
+  template<typename U>
+  SkipListNode(uint8_t height, U&& data, bool isHead) :
+      height_(height), data_(std::forward<U>(data)) {
+    spinLock_.init();
+    setFlags(0);
+    if (isHead) setIsHeadNode();
+    // need to explicitly init the dynamic atomic pointer array
     for (uint8_t i = 0; i < height_; ++i) {
-      skip_[i].~atomic();
+      new (&skip_[i]) std::atomic<SkipListNode*>(nullptr);
     }
   }
-  explicit SkipListNode(uint8_t height) : height_(height) {
+
+  ~SkipListNode() {
     for (uint8_t i = 0; i < height_; ++i) {
-      new (&skip_[i]) std::atomic<SkipListNode*>(nullptr);
+      skip_[i].~atomic();
     }
   }
 

folly/ConcurrentSkipList.h

@@ -358,7 +358,8 @@ class ConcurrentSkipList {
   //     list with the same key.
   //   pair.second stores whether the data is added successfully:
   //     0 means not added, otherwise reutrns the new size.
-  std::pair<NodeType*, size_t> addOrGetData(const value_type &data) {
+  template<typename U>
+  std::pair<NodeType*, size_t> addOrGetData(U &&data) {
     NodeType *preds[MAX_HEIGHT], *succs[MAX_HEIGHT];
     NodeType *newNode;
     size_t newSize;
@@ -387,7 +388,7 @@ class ConcurrentSkipList {
       }
 
       // locks acquired and all valid, need to modify the links under the locks.
-      newNode = NodeType::create(nodeHeight, data);
+      newNode = NodeType::create(nodeHeight, std::forward<U>(data));
       for (int layer = 0; layer < nodeHeight; ++layer) {
         newNode->setSkip(layer, succs[layer]);
         preds[layer]->setSkip(layer, newNode);
@@ -553,7 +554,7 @@ class ConcurrentSkipList {
     { // need to guard the head node in case others are adding/removing
       // nodes linked to the head.
       ScopedLocker g = oldHead->acquireGuard();
-      newHead->promoteFrom(oldHead);
+      newHead->copyHead(oldHead);
       NodeType* expected = oldHead;
       if (!head_.compare_exchange_strong(expected, newHead,
           std::memory_order_release)) {
@@ -650,8 +651,10 @@ class ConcurrentSkipList<T, Comp, MAX_HEIGHT>::Accessor {
   const_iterator cbegin() const { return begin(); }
   const_iterator cend() const { return end(); }
 
-  std::pair<iterator, bool> insert(const key_type &data) {
-    auto ret = sl_->addOrGetData(data);
+  template<typename U,
+    typename=typename std::enable_if<std::is_convertible<U, T>::value>::type>
+  std::pair<iterator, bool> insert(U&& data) {
+    auto ret = sl_->addOrGetData(std::forward<U>(data));
     return std::make_pair(iterator(ret.first), ret.second);
   }
   size_t erase(const key_type &data) { return remove(data); }

folly/test/ConcurrentSkipListTest.cpp

@@ -16,6 +16,7 @@
 
 // @author: Xin Liu <xliux@fb.com>
 
+#include <memory>
 #include <set>
 #include <vector>
 #include <thread>
@@ -207,6 +208,55 @@ TEST(ConcurrentSkipList, SequentialAccess) {
 
 }
 
+static std::string makeRandomeString(int len) {
+  std::string s;
+  for (int j = 0; j < len; j++) {
+    s.push_back((rand() % 26) + 'A');
+  }
+  return s;
+}
+
+TEST(ConcurrentSkipList, TestStringType) {
+  typedef folly::ConcurrentSkipList<std::string> SkipListT;
+  boost::shared_ptr<SkipListT> skip = SkipListT::createInstance();
+  SkipListT::Accessor accessor(skip);
+  {
+    for (int i = 0; i < 100000; i++) {
+      std::string s = makeRandomeString(7);
+      accessor.insert(s);
+    }
+  }
+  EXPECT_TRUE(std::is_sorted(accessor.begin(), accessor.end()));
+}
+
+struct UniquePtrComp {
+  bool operator ()(
+      const std::unique_ptr<int> &x, const std::unique_ptr<int> &y) const {
+    if (!x) return false;
+    if (!y) return true;
+    return *x < *y;
+  }
+};
+
+TEST(ConcurrentSkipList, TestMovableData) {
+  typedef folly::ConcurrentSkipList<std::unique_ptr<int>, UniquePtrComp>
+    SkipListT;
+  auto sl = SkipListT::createInstance() ;
+  SkipListT::Accessor accessor(sl);
+
+  static const int N = 10;
+  for (int i = 0; i < N; ++i) {
+    accessor.insert(std::unique_ptr<int>(new int(i)));
+  }
+
+  for (int i = 0; i < N; ++i) {
+    EXPECT_TRUE(accessor.find(std::unique_ptr<int>(new int(i))) !=
+        accessor.end());
+  }
+  EXPECT_TRUE(accessor.find(std::unique_ptr<int>(new int(N))) ==
+      accessor.end());
+}
+
 void testConcurrentAdd(int numThreads) {
   auto skipList(SkipListType::create(kHeadHeight));