From 655d03592ffb4d484fba7240d933b283d2ff94ad Mon Sep 17 00:00:00 2001
From: Michael Krasnyk <michael.krasnyk@gmail.com>
Date: Mon, 26 Jun 2017 15:12:39 +0200
Subject: [PATCH] Use thread-safe lock-free assignment in
 PackedVector::set_value

PR uses TBB internal atomic's for atomic CAS on non-atomic data

Corresponding PR https://github.com/Project-OSRM/osrm-backend/pull/4199

Other options:

* use sequential update

* use an internal packed vector lock -> makes packed vector non-movable

* use boost.interprocess atomics implementation -> outdated and only 32 bit version

* use glib atomic's -> requires new dependency

*  wait for https://isocpp.org/blog/2014/05/n4013 as_atomic

*  use c11 _Atomic and atomic_compare_exchange_weak -> not possible to mix c++11 and c11

* use builtin functions gcc __sync_bool_compare_and_swap and msvc _InterlockedCompareExchange64 -> possible, but requires proper testing

boolean CompareAndSwapPointer(volatile * void * ptr,
                              void * new_value,
                              void * old_value) {
if defined(_MSC_VER)
   if (InterlockedCompareExchange(ptr, new_value, old_value) == old_value) return false;
   else return true;
elif (__GNUC__ * 10000 + __GNUC_MINOR__ * 100 + __GNUC_PATCHLEVEL__) > 40100
   return __sync_bool_compare_and_swap(ptr, old_value, new_value);
else
  error No implementation
endif
}

* use Boost.Atomic -> requires new dependency

        WordT local_lower_word = lower_word, new_lower_word;
        do
        {
            new_lower_word = set_lower_value<WordT, T>(local_lower_word,
                                                       lower_mask[internal_index.element],
                                                       lower_offset[internal_index.element],
                                                       value);
        } while (!boost::atomics::detail::operations<sizeof(WordT), false>::compare_exchange_weak(
            lower_word,
            local_lower_word,
            new_lower_word,
            boost::memory_order_release,
            boost::memory_order_relaxed));
---
 include/util/packed_vector.hpp | 40 +++++++++++++++++++++++++---------
 1 file changed, 30 insertions(+), 10 deletions(-)

diff --git a/include/util/packed_vector.hpp b/include/util/packed_vector.hpp
index 18b1e7bb8..8d4672f1e 100644
--- a/include/util/packed_vector.hpp
+++ b/include/util/packed_vector.hpp
@@ -10,6 +10,7 @@
 
 #include <boost/iterator/iterator_facade.hpp>
 #include <boost/iterator/reverse_iterator.hpp>
+#include <tbb/atomic.h>
 
 #include <array>
 #include <cmath>
@@ -348,7 +349,7 @@ template <typename T, std::size_t Bits, storage::Ownership Ownership> class Pack
         fill(initial_value);
     }
 
-    PackedVector(util::ViewOrVector<std::uint64_t, Ownership> vec_, std::size_t num_elements)
+    PackedVector(util::ViewOrVector<WordT, Ownership> vec_, std::size_t num_elements)
         : vec(std::move(vec_)), num_elements(num_elements)
     {
         initialize();
@@ -497,20 +498,39 @@ template <typename T, std::size_t Bits, storage::Ownership Ownership> class Pack
 
     inline void set_value(const InternalIndex internal_index, const T value)
     {
+        // ⚠ The method uses CAS spinlocks to prevent data races in parallel calls
+        // TBB internal atomic's are used for CAS on non-atomic data
+        // Parallel read and write access is not allowed
+
         auto &lower_word = vec[internal_index.lower_word];
         auto &upper_word = vec[internal_index.lower_word + 1];
 
-        lower_word = set_lower_value<WordT, T>(lower_word,
-                                               lower_mask[internal_index.element],
-                                               lower_offset[internal_index.element],
-                                               value);
-        upper_word = set_upper_value<WordT, T>(upper_word,
-                                               upper_mask[internal_index.element],
-                                               upper_offset[internal_index.element],
-                                               value);
+        // Lock-free update of the lower word
+        WordT local_lower_word, new_lower_word;
+        do
+        {
+            local_lower_word = lower_word;
+            new_lower_word = set_lower_value<WordT, T>(local_lower_word,
+                                                       lower_mask[internal_index.element],
+                                                       lower_offset[internal_index.element],
+                                                       value);
+        } while (tbb::internal::as_atomic(lower_word)
+                     .compare_and_swap(new_lower_word, local_lower_word) != local_lower_word);
+
+        // Lock-free update of the upper word
+        WordT local_upper_word, new_upper_word;
+        do
+        {
+            local_upper_word = upper_word;
+            new_upper_word = set_upper_value<WordT, T>(local_upper_word,
+                                                       upper_mask[internal_index.element],
+                                                       upper_offset[internal_index.element],
+                                                       value);
+        } while (tbb::internal::as_atomic(upper_word)
+                     .compare_and_swap(new_upper_word, local_upper_word) != local_upper_word);
     }
 
-    util::ViewOrVector<std::uint64_t, Ownership> vec;
+    util::ViewOrVector<WordT, Ownership> vec;
     std::uint64_t num_elements = 0;
 };
 }