Persistent-aware self_relative_ptr and Single-writer-multi-reader (SWMR) skiplist

CMakeLists.txt

@@ -99,6 +99,7 @@ option(TEST_SEGMENT_VECTOR_VECTOR_EXPSIZE "enable testing of pmem::obj::segment_
 option(TEST_SEGMENT_VECTOR_VECTOR_FIXEDSIZE "enable testing of pmem::obj::segment_vector with vector as segment_vector_type and fixed_size_policy" ON)
 option(TEST_ENUMERABLE_THREAD_SPECIFIC "enable testing of pmem::obj::enumerable_thread_specific" ON)
 option(TEST_CONCURRENT_MAP "enable testing of pmem::obj::experimental::concurrent_map (depends on TEST_STRING)" ON)
+option(TEST_SWMR_MAP "enable testing of pmem::obj::experimental::swmr_map (depends on TEST_STRING)" ON)
 option(TEST_SELF_RELATIVE_POINTER "enable testing of pmem::obj::experimental::self_relative_ptr" ON)
 option(TEST_RADIX_TREE "enable testing of pmem::obj::experimental::radix_tree" ON)
 option(TEST_MPSC_QUEUE "enable testing of pmem::obj::experimental::mpsc_queue" ON)

benchmarks/self_relative_pointer/assignment.cpp

@@ -37,7 +37,8 @@ using size_type = std::ptrdiff_t;
 template <typename U>
 using persistent_ptr = pmem::obj::persistent_ptr<U>;
 template <typename U>
-using self_relative_ptr = pmem::obj::experimental::self_relative_ptr<U>;
+using self_relative_ptr =
+	pmem::obj::experimental::self_relative_ptr<U, std::false_type>;
 template <typename U>
 using vector = pmem::obj::vector<U>;
 
@@ -81,6 +82,28 @@ benchmark_assignment(pointer<pointer<value_type>[]> &array,
 	}
 }
 
+template <template <typename U, typename PersistentAware> class pointer>
+void
+benchmark_swap(
+	pointer<pointer<value_type, std::false_type>[], std::false_type> &array,
+	pointer<value_type, std::false_type> value)
+{
+	for (size_type i = 0; i < ARR_SIZE; i++) {
+		swap(array[i], value);
+	}
+}
+
+template <template <typename U, typename PersistentAware> class pointer>
+void
+benchmark_assignment(
+	pointer<pointer<value_type, std::false_type>[], std::false_type> &array,
+	pointer<value_type, std::false_type> value)
+{
+	for (size_type i = 0; i < ARR_SIZE; i++) {
+		array[i] = value;
+	}
+}
+
 int
 main(int argc, char *argv[])
 {

benchmarks/self_relative_pointer/get.cpp

@@ -82,7 +82,8 @@ main(int argc, char *argv[])
 		root->pptr = prepare_array(pop, ARR_SIZE);
 
 		pmem::obj::persistent_ptr<value_type> pptr = root->pptr;
-		pmem::obj::experimental::self_relative_ptr<value_type>
+		pmem::obj::experimental::self_relative_ptr<value_type,
+							   std::false_type>
 			offset_ptr = root->pptr;
 		int *vptr = root->pptr.get();
 

include/libpmemobj++/container/detail/concurrent_skip_list_impl.hpp

@@ -120,7 +120,8 @@ allocator_swap(MyAlloc &, OtherAlloc &, std::false_type)
 { /* NO SWAP */
 }
 
-template <typename Value, typename Mutex = pmem::obj::mutex,
+template <typename Value, typename UsePersistentAwarePtr,
+	  typename Mutex = pmem::obj::mutex,
 	  typename LockType = std::unique_lock<Mutex>>
 class skip_list_node {
 public:
@@ -131,7 +132,8 @@ class skip_list_node {
 	using pointer = value_type *;
 	using const_pointer = const value_type *;
 	using node_pointer =
-		obj::experimental::self_relative_ptr<skip_list_node>;
+		obj::experimental::self_relative_ptr<skip_list_node,
+						     UsePersistentAwarePtr>;
 	using atomic_node_pointer = std::atomic<node_pointer>;
 	using mutex_type = Mutex;
 	using lock_type = LockType;
@@ -210,6 +212,18 @@ class skip_list_node {
 		return get_next(level).load(std::memory_order_acquire);
 	}
 
+	template <typename U = void,
+		  typename = typename std::enable_if<
+			  std::is_same<UsePersistentAwarePtr,
+				       std::true_type>::value,
+			  U>::type>
+	node_pointer
+	next(size_type level)
+	{
+		assert(level < height());
+		return get_next(level).persist_load(std::memory_order_acquire);
+	}
+
 	/**
 	 * Can`t be called concurrently
 	 * Should be called inside a transaction
@@ -233,14 +247,27 @@ class skip_list_node {
 		pop.persist(&node, sizeof(node));
 	}
 
+	void
+	set_next(size_type level, node_pointer next)
+	{
+		assert(level < height());
+		auto &node = get_next(level);
+		node.store(node_pointer{next.get(), true},
+			   std::memory_order_release);
+		/* instead of persist it immediately, mark it dirty,
+		 * and rely on consequent get_next operation to flush.
+		 */
+	}
+
 	void
 	set_nexts(const node_pointer *new_nexts, size_type h)
 	{
 		assert(h == height());
 		auto *nexts = get_nexts();
 
 		for (size_type i = 0; i < h; i++) {
-			nexts[i].store(new_nexts[i], std::memory_order_relaxed);
+			nexts[i].store(node_pointer{new_nexts[i].get(), true},
+				       std::memory_order_relaxed);
 		}
 	}
 
@@ -497,8 +524,11 @@ class concurrent_skip_list {
 	using const_reference = const value_type &;
 	using pointer = typename allocator_traits_type::pointer;
 	using const_pointer = typename allocator_traits_type::const_pointer;
+	using use_persistent_aware_ptr =
+		typename traits_type::use_persistent_aware_ptr;
 
-	using list_node_type = skip_list_node<value_type>;
+	using list_node_type =
+		skip_list_node<value_type, use_persistent_aware_ptr>;
 
 	using iterator = skip_list_iterator<list_node_type, false>;
 	using const_iterator = skip_list_iterator<list_node_type, true>;
@@ -514,7 +544,8 @@ class concurrent_skip_list {
 	using node_ptr = list_node_type *;
 	using const_node_ptr = const list_node_type *;
 	using persistent_node_ptr =
-		obj::experimental::self_relative_ptr<list_node_type>;
+		obj::experimental::self_relative_ptr<list_node_type,
+						     use_persistent_aware_ptr>;
 
 	using prev_array_type = std::array<node_ptr, MAX_LEVEL>;
 	using next_array_type = std::array<persistent_node_ptr, MAX_LEVEL>;
@@ -2306,6 +2337,33 @@ class concurrent_skip_list {
 			      sizeof(decltype(size_diff)) -
 			      sizeof(decltype(insert_stage))];
 	};
+
+	class swmr_tls_data : public obj::segment_vector<tls_entry_type> {
+	public:
+		using base_type = obj::segment_vector<tls_entry_type>;
+		swmr_tls_data()
+		{
+		}
+		~swmr_tls_data() = default;
+		tls_entry_type &
+		local()
+		{
+			if (this->size() == 0) {
+				this->resize(1);
+			}
+			return this->front();
+		}
+
+	private:
+		obj::pool_base
+		get_pool() const noexcept
+		{
+			auto pop = pmemobj_pool_by_ptr(this);
+			assert(pop != nullptr);
+			return obj::pool_base(pop);
+		}
+	};
+
 	static_assert(sizeof(tls_entry_type) == 64,
 		      "The size of tls_entry_type should be 64 bytes.");
 
@@ -2430,14 +2488,12 @@ class concurrent_skip_list {
 		assert(level < prev->height());
 		persistent_node_ptr next = prev->next(level);
 		pointer_type curr = next.get();
-
 		while (curr && cmp(get_key(curr), key)) {
 			prev = curr;
 			assert(level < prev->height());
 			next = prev->next(level);
 			curr = next.get();
 		}
-
 		return next;
 	}
 
@@ -2645,7 +2701,6 @@ class concurrent_skip_list {
 
 		do {
 			find_insert_pos(prev_nodes, next_nodes, key);
-
 			node_ptr next = next_nodes[0].get();
 			if (next && !allow_multimapping &&
 			    !_compare(key, get_key(next))) {
@@ -2753,9 +2808,11 @@ class concurrent_skip_list {
 		return true;
 	}
 
-	bool
-	try_lock_nodes(size_type height, prev_array_type &prevs,
-		       const next_array_type &nexts, lock_array &locks)
+	template <typename Is_SWMR>
+	typename std::enable_if<std::is_same<Is_SWMR, std::false_type>::value,
+				bool>::type
+	try_lock_nodes_impl(size_type height, prev_array_type &prevs,
+			    const next_array_type &nexts, lock_array &locks)
 	{
 		assert(check_prev_array(prevs, height));
 
@@ -2775,6 +2832,23 @@ class concurrent_skip_list {
 		return true;
 	}
 
+	template <typename Is_SWMR>
+	typename std::enable_if<std::is_same<Is_SWMR, std::true_type>::value,
+				bool>::type
+	try_lock_nodes_impl(size_type height, prev_array_type &prevs,
+			    const next_array_type &nexts, lock_array &locks)
+	{
+		return true;
+	}
+
+	bool
+	try_lock_nodes(size_type height, prev_array_type &prevs,
+		       const next_array_type &nexts, lock_array &locks)
+	{
+		return try_lock_nodes_impl<use_persistent_aware_ptr>(
+			height, prevs, nexts, locks);
+	}
+
 	/**
 	 * Returns an iterator pointing to the first element from the list for
 	 * which cmp(element, key) is false.
@@ -3075,11 +3149,9 @@ class concurrent_skip_list {
 	{
 		assert(pmemobj_tx_stage() == TX_STAGE_WORK);
 		size_type sz = calc_node_size(height);
-
 		persistent_node_ptr n =
 			node_allocator_traits::allocate(_node_allocator, sz)
 				.raw();
-
 		assert(n != nullptr);
 
 		node_allocator_traits::construct(_node_allocator, n.get(),
@@ -3243,7 +3315,12 @@ class concurrent_skip_list {
 	random_level_generator_type _rnd_generator;
 	persistent_node_ptr dummy_head;
 
-	enumerable_thread_specific<tls_entry_type> tls_data;
+	using tls_data_storage = typename std::conditional<
+		std::is_same<use_persistent_aware_ptr, std::true_type>::value,
+		swmr_tls_data,
+		enumerable_thread_specific<tls_entry_type>>::type;
+
+	tls_data_storage tls_data;
 
 	std::atomic<size_type> _size;
 
@@ -3257,7 +3334,7 @@ class concurrent_skip_list {
 
 template <typename Key, typename Value, typename KeyCompare,
 	  typename RND_GENERATOR, typename Allocator, bool AllowMultimapping,
-	  size_t MAX_LEVEL>
+	  size_t MAX_LEVEL, typename UsePersistentAwarePtr = std::false_type>
 class map_traits {
 public:
 	static constexpr size_t max_level = MAX_LEVEL;
@@ -3269,6 +3346,7 @@ class map_traits {
 	using reference = value_type &;
 	using const_reference = const value_type &;
 	using allocator_type = Allocator;
+	using use_persistent_aware_ptr = UsePersistentAwarePtr;
 
 	/**
 	 * pmem::detail::concurrent_skip_list allows multimapping. If this flag

include/libpmemobj++/detail/self_relative_ptr_base_impl.hpp

@@ -44,7 +44,8 @@ class self_relative_ptr_base_impl {
 	using byte_type = uint8_t;
 	using byte_ptr_type = byte_type *;
 	using const_byte_ptr_type = const byte_type *;
-
+	static constexpr difference_type dirty_flag = ~(1L << 1);
+	static constexpr difference_type dirty_mask = dirty_flag & (1 << 1 | 1);
 	/*
 	 * Constructors
 	 */
@@ -190,6 +191,30 @@ class self_relative_ptr_base_impl {
 	{
 		return offset == nullptr_offset;
 	}
+	/**
+	 * check if offset is dirty
+	 */
+	static inline bool
+	is_dirty(difference_type other_offset)
+	{
+		return ((other_offset != nullptr_offset) &&
+			(other_offset & dirty_mask) ==
+				(dirty_flag & dirty_mask));
+	}
+	bool
+	is_dirty()
+	{
+		return (!is_null() &&
+			(difference_type(offset) & dirty_mask) ==
+				(dirty_flag & dirty_mask));
+	}
+	void
+	set_dirty_flag(bool dirty)
+	{
+		intptr_t dirty_mask = dirty == true;
+		--dirty_mask;
+		offset &= (dirty_mask | dirty_flag);
+	}
 
 protected:
 	/**
@@ -222,8 +247,12 @@ class self_relative_ptr_base_impl {
 		*/
 		uintptr_t mask = other_offset == nullptr_offset;
 		--mask;
+		intptr_t mask_dirty = is_dirty(other_offset) == true;
+		--mask_dirty;
+		/* clear the dirty_flag if it's set to get the correct ptr. */
 		uintptr_t ptr = static_cast<uintptr_t>(
-			reinterpret_cast<intptr_t>(this) + other_offset + 1);
+			reinterpret_cast<intptr_t>(this) +
+			(other_offset | ~(dirty_flag | mask_dirty)) + 1);
 		ptr &= mask;
 		return reinterpret_cast<void *>(ptr);
 	}

include/libpmemobj++/detail/tagged_ptr.hpp

@@ -169,12 +169,13 @@ struct tagged_ptr_impl {
 	PointerType ptr;
 
 	friend struct std::atomic<tagged_ptr_impl<
-		P1, P2, obj::experimental::self_relative_ptr<void>>>;
+		P1, P2,
+		obj::experimental::self_relative_ptr<void, std::false_type>>>;
 };
 
 template <typename P1, typename P2>
-using tagged_ptr =
-	tagged_ptr_impl<P1, P2, obj::experimental::self_relative_ptr<void>>;
+using tagged_ptr = tagged_ptr_impl<
+	P1, P2, obj::experimental::self_relative_ptr<void, std::false_type>>;
 
 } /* namespace detail */
 } /* namespace pmem */
@@ -187,7 +188,8 @@ struct atomic<pmem::detail::tagged_ptr<P1, P2>> {
 private:
 	using ptr_type = pmem::detail::tagged_ptr_impl<
 		P1, P2,
-		atomic<pmem::obj::experimental::self_relative_ptr<void>>>;
+		atomic<pmem::obj::experimental::self_relative_ptr<
+			void, std::false_type>>>;
 	using value_type = pmem::detail::tagged_ptr<P1, P2>;
 
 public: