containers_cpp: fixes in LlxScx, added unit test for LlxScx

a3c62f0c · Tobias Fuchs · a425b12e · a3c62f0c · a3c62f0c · a3c62f0c
Commit a3c62f0c authored Mar 25, 2015 by Tobias Fuchs
6 changed files
--- a/containers_cpp/include/embb/containers/internal/fixed_size_list.h
+++ b/containers_cpp/include/embb/containers/internal/fixed_size_list.h
@@ -75,7 +75,7 @@ class FixedSizeList {
    size_t capacity
    /**< [IN] Capacity of the list */
  );
  /**
   * Copy constructor.
   */

--- a/containers_cpp/include/embb/containers/internal/primitives/llx_scx-inl.h
+++ b/containers_cpp/include/embb/containers/internal/primitives/llx_scx-inl.h
@@ -44,99 +44,8 @@ namespace embb {
 namespace containers {
 namespace primitives {
-namespace internal {
+template< typename UserData, typename ValuePool >
+unsigned int LlxScx<UserData, ValuePool>::ThreadId() {
-#if 0
-/**
- * RAII-style implementation of a simplified hazard
- * pointer scheme. SmartHazardPointer instances should
- * reside only on the stack, never the heap, and T*
- * values should reside only in the heap. A NodePtr can
- * be automatically converted to a Node*, but an explicit
- * constructor is needed to go the other way.
- *
- * Example:
- *   SmartHazardPointer<Node> shp = SmartHazardPointer(
- *     nodePool?>allocate());
- *   // assign via default copy constructor:
- *   *shp = Node(...);
- */
-template< typename T >
-class SmartHazardPointer {
- public:
-  SmartHazardPointer(T ** node) {
-    while (true) {
-      ptr = *node;
-      table_->add(ptr);
-      // Full fence:
-      embb_atomic_memory_barrier();
-      T * reread = *node;
-      // @TODO: Prove practical wait-freedom
-      if (read == reread) {
-        return;
-      }
-    }
-  }
-  SmartHazardPointer(T * node) {
-    ptr = node;
-  }
-  /**
-   * Dereference operator.
-   */
-  T * operator->() {
-    return ptr;
-  }
-  /**
-   * Dereference lvalue.
-   */
-  T & operator*() {
-    return *ptr;
-  }
-  /**
-   * Equality test with regular pointer
-   */
-  bool operator==(T * const other) {
-    return ptr == other;
-  }
-  /**
-   * Equality test with another SmartHazardPointer
-   */
-  bool operator==(const SmartHazardPointer & other) {
-    return this->ptr == other->ptr;
-  }
-  /**
-   * Destructor, retires hazard pointer.
-   */
-  ~SmartHazardPointer() {
-    table_->remove(ptr);
-  }
-  /**
-   * Conversion to regular pointer.
-   */
-  operator T*() {
-    return ptr;
-  }
- private:
-  static HazardPointerTable * table_;
-  T * ptr;
-};
-#endif
-}  // namespace internal
-template< class UserData >
-unsigned int LlxScx<UserData>::ThreadId() {
  unsigned int thread_index;
  int return_val = embb_internal_thread_index(&thread_index);
  if (return_val != EMBB_SUCCESS)
@@ -144,34 +53,33 @@ unsigned int LlxScx<UserData>::ThreadId() {
  return thread_index;
 }
-template< class UserData >
+template< typename UserData, typename ValuePool >
-LlxScx<UserData>::LlxScx(size_t max_links)
+LlxScx<UserData, ValuePool>::LlxScx(size_t max_links)
-: max_links_(max_links) {
+: max_links_(max_links),
+  max_threads_(embb::base::Thread::GetThreadsMaxCount()),
+  scx_record_list_pool_(max_threads_) {
  typedef embb::containers::internal::FixedSizeList<LlxResult>
    llx_result_list_t;
  typedef embb::containers::internal::FixedSizeList<ScxRecord_t>
    scx_record_list_t;
-  unsigned int num_threads = embb::base::Thread::GetThreadsMaxCount();
+  // Allocate a list of LLX results for every thread:
-  // Table in shared memory containing for each r in TryStoreConditional's
+  thread_llx_results_ = static_cast<llx_result_list_t *>(
-  // dependent links, a copy of r's info value in this threads local table
-  // of LLX results.
-  scx_record_list_t empty_scx_list(3);
-  info_fields_ = static_cast<scx_record_list_t *>(
    embb::base::Allocation::AllocateCacheAligned(
-      num_threads * sizeof(empty_scx_list)));
+      max_threads_ * sizeof(llx_result_list_t)));
 }
-template< class UserData >
+template< typename UserData, typename ValuePool >
-LlxScx<UserData>::~LlxScx() {
+LlxScx<UserData, ValuePool>::~LlxScx() {
-  embb::base::Allocation::FreeAligned(info_fields_);
+  embb::base::Allocation::FreeAligned(thread_llx_results_);
 }
-template< class UserData >
+template< typename UserData, typename ValuePool >
-bool LlxScx<UserData>::TryLoadLinked(
+bool LlxScx<UserData, ValuePool>::TryLoadLinked(
  DataRecord_t * const data_record,
  DataRecord_t & user_data,
  bool & finalized) {
  finalized = false;
+  unsigned int thread_id = ThreadId();
  // Order of initialization matters:
  bool           marked_1   = data_record->IsMarkedForFinalize();
  ScxRecord_t *  curr_scx   = data_record->ScxInfo().Load();
@@ -188,7 +96,7 @@ bool LlxScx<UserData>::TryLoadLinked(
      llx_result.data_record = data_record;
      llx_result.scx_record  = curr_scx;
      llx_result.user_data   = user_data_local;
-      thread_llx_results_.Get().PushBack(llx_result);
+      thread_llx_results_[thread_id].PushBack(llx_result);
      // Set return value:
      user_data = user_data_local;
      return true;
@@ -210,15 +118,15 @@ bool LlxScx<UserData>::TryLoadLinked(
  return false;
 }
-template< class UserData >
+template< typename UserData, typename ValuePool >
 template< typename FieldType >
-bool LlxScx<UserData>::TryStoreConditional(
+bool LlxScx<UserData, ValuePool>::TryStoreConditional(
  embb::base::Atomic<FieldType> * field,
  FieldType value,
  embb::containers::internal::FixedSizeList<DataRecord_t *> & linked_deps,
  embb::containers::internal::FixedSizeList<DataRecord_t *> & finalize_deps) {
  typedef embb::containers::internal::FixedSizeList<DataRecord_t *> dr_list_t;
-//typedef embb::containers::internal::FixedSizeList<ScxRecord_t *> op_list_t;
+  typedef embb::containers::internal::FixedSizeList<ScxRecord_t> scx_op_list_t;
  // Preconditions:
  // 1. For each r in linked_deps, this thread has performed an invocation
  //    I_r of LLX(r) linked to this SCX.
@@ -228,17 +136,18 @@ bool LlxScx<UserData>::TryStoreConditional(
  unsigned int thread_id = ThreadId();
  // Let info_fields be a table in shared memory containing for each r in V,
  // a copy of r's info value in this threads local table of LLX results:
-  info_fields_[thread_id].clear();
+  scx_op_list_t * info_fields = scx_record_list_pool_.Allocate(max_links_);
  dr_list_t::const_iterator it;
  dr_list_t::const_iterator end;
  end = linked_deps.end();
-  // for each r in V ...
+  // for each r in linked_deps ...
  for (it = linked_deps.begin(); it != end; ++it) {
    // Find LLX result of r in thread-local table of LLX results:
    typedef embb::containers::internal::FixedSizeList<LlxResult>
      llx_result_list;
-    llx_result_list::iterator l_it  = thread_llx_results_.Get().begin();
+    llx_result_list::iterator l_it  = thread_llx_results_[thread_id].begin();
-    llx_result_list::iterator l_end = thread_llx_results_.Get().end();
+    llx_result_list::iterator l_end = thread_llx_results_[thread_id].end();
+    // Find LLX result of r in thread-local LLX results:
    for (; l_it != l_end && l_it->data_record != *it; ++l_it);
    if (l_it == l_end) {
      // Missing LLX result for given linked data record, user did not
@@ -246,9 +155,9 @@ bool LlxScx<UserData>::TryStoreConditional(
      EMBB_THROW(embb::base::ErrorException,
        "Missing preceding LLX on a data record used for SCX");
    }
-    // ... copy of r's info value in this threads local table of LLX results
+    // Copy of r's info value in this threads local table of LLX results
    ScxRecord_t scx_op(*(l_it->data_record->ScxInfo().Load()));
-    info_fields_[thread_id].PushBack(scx_op);
+    info_fields->PushBack(scx_op);
  }
  // Announce SCX operation. Lists linked_deps and finalize_dep are 
  // guaranteed to remain on the stack until this announced operation
@@ -263,27 +172,27 @@ bool LlxScx<UserData>::TryStoreConditional(
    // old value:
    reinterpret_cast<cas_t>(field->Load()),
    // linked SCX operations:
-    &info_fields_[thread_id],
+    info_fields,
    // initial operation state:
    OperationState::InProgress);
  return scx.Help();
 }
-template< class UserData >
+template< typename UserData, typename ValuePool >
-bool LlxScx<UserData>::TryValidateLink(
+bool LlxScx<UserData, ValuePool>::TryValidateLink(
  const DataRecord_t & field) {
  return true; // @TODO
 }
 // LlxScxRecord
-template< class UserData >
+template< typename UserData >
 LlxScxRecord<UserData>::LlxScxRecord()
 : marked_for_finalize_(false) {
  scx_op_.Store(&dummy_scx);
 }
-template< class UserData >
+template< typename UserData >
 LlxScxRecord<UserData>::LlxScxRecord(
  const UserData & user_data)
 : user_data_(user_data),
@@ -293,24 +202,26 @@ LlxScxRecord<UserData>::LlxScxRecord(
 // internal::ScxRecord
-template< class DataRecord >
+template< typename DataRecord >
 bool internal::ScxRecord<DataRecord>::Help() {
  // We ensure that an SCX S does not change a data record 
  // while it is frozen for another SCX S'. Instead, S uses 
  // the information in the SCX record of S' to help S'
  // complete, so that the data record can be unfrozen.
  typedef embb::containers::internal::FixedSizeList<DataRecord *> dr_list_t;
+  typedef embb::containers::internal::FixedSizeList<self_t> op_list_t;
  // Freeze all data records in data_records to protect their 
  // mutable fields from being changed by other SCXs:
  dr_list_t::iterator linked_it  = linked_data_records_->begin();
  dr_list_t::iterator linked_end = linked_data_records_->end();
-  for (unsigned int fieldIdx = 0;
+  op_list_t::iterator scx_op_it  = scx_ops_->begin();
-       linked_it != linked_end;
+  op_list_t::iterator scx_op_end = scx_ops_->end();
-       ++linked_it, ++fieldIdx) {
+  for (; linked_it != linked_end && scx_op_it != scx_op_end;
+       ++linked_it, ++scx_op_it) {
    DataRecord * r = *linked_it;
    // pointer indexed by r in this->info_fields:
-    ScxRecord<DataRecord> * rinfo = &info_fields_[fieldIdx];
+    ScxRecord<DataRecord> * rinfo_exp = &(*scx_op_it);
-    if (!r->ScxInfo().CompareAndSwap(rinfo, this)) {
+    if (!r->ScxInfo().CompareAndSwap(rinfo_exp, this)) {
      if (r->ScxInfo().Load() != this) {
        // could not freeze r because it is frozen for 
        // another SCX:
@@ -331,10 +242,8 @@ bool internal::ScxRecord<DataRecord>::Help() {
  // mark step:
  dr_list_t::iterator finalize_it  = finalize_data_records_->begin();
  dr_list_t::iterator finalize_end = finalize_data_records_->end();
-  for (unsigned int field_idx = finalize_range_.first;
+  for (; finalize_it != finalize_end; ++finalize_it) {
-       finalize_it != finalize_end;
+    (*finalize_it)->MarkForFinalize();
-       ++finalize_it, ++fieldRangeIdx) {
-    linked_data_records_[fieldRangeIdx]->MarkForFinalize();
  }
  // update CAS:
  cas_t expected_old_value = old_value_;
@@ -347,6 +256,11 @@ bool internal::ScxRecord<DataRecord>::Help() {
  return true;
 }
+template< typename UserData >
+internal::ScxRecord< LlxScxRecord<UserData> > 
+  LlxScxRecord<UserData>::dummy_scx = 
+    internal::ScxRecord< LlxScxRecord<UserData> >();
 } // namespace primitives
 } // namespace containers
 } // namespace embb

--- a/containers_cpp/include/embb/containers/primitives/llx_scx.h
+++ b/containers_cpp/include/embb/containers/primitives/llx_scx.h
@@ -30,6 +30,8 @@
 #include <embb/base/thread.h>
 #include <embb/base/atomic.h>
 #include <embb/base/thread_specific_storage.h>
+#include <embb/containers/object_pool.h>
+#include <embb/containers/lock_free_tree_value_pool.h>
 #include <embb/containers/internal/fixed_size_list.h>
 namespace embb { 
@@ -42,7 +44,7 @@ namespace internal {
 * SCX operation description. An SCX record contains all information
 * required to allow any process to complete a pending SCX operation.
 */
-template< class DataRecord >
+template< typename DataRecord >
 class ScxRecord {
 private:
@@ -66,11 +68,11 @@ class ScxRecord {
   */
  ScxRecord()
  : linked_data_records_(0),
-    finalize_data_records_(0)
+    finalize_data_records_(0),
    new_value_(0),
    old_value_(0),
    scx_ops_(0),
-    state_(OperationState::Undefined),
+    state_(OperationState::Comitted),
    all_frozen_(false) {
    field_ = 0;
  }
@@ -84,7 +86,7 @@ class ScxRecord {
    embb::base::Atomic<cas_t> * field,
    cas_t new_value,
    cas_t old_value,
-    embb::containers::internal::FixedSizeList<self_t *> * scx_ops,
+    embb::containers::internal::FixedSizeList<self_t> * scx_ops,
    OperationState operation_state)
  : linked_data_records_(&linked_data_records),
    finalize_data_records_(&finalize_data_records),
@@ -143,7 +145,7 @@ class ScxRecord {
   * List of SCX operation descriptions associated with data records
   * linked with this SCX operation.
   */
-  embb::containers::internal::FixedSizeList<self_t *> scx_ops_;
+  embb::containers::internal::FixedSizeList<self_t> * scx_ops_;
  /**
   * Current state of this SCX record.
@@ -165,7 +167,7 @@ class ScxRecord {
 * Wraps user-defined data with fields required for LLX/SCX algorithm.
 * Mutable fields must each be contained in a single word.
 */
-template< class UserData >
+template< typename UserData >
 class LlxScxRecord {
 private:
@@ -203,9 +205,12 @@ class LlxScxRecord {
   * Assignment operator.
   */
  LlxScxRecord & operator=(const LlxScxRecord & rhs) {
-    user_data_ = rhs.user_data_;
+    if (this != &rhs) {
-    scx_op_.Store(rhs.scx_info_.Load());
+      user_data_ = rhs.user_data_;
-    marked_for_finalize_ = rhs.marked_for_finalize_;
+      scx_op_.Store(rhs.scx_op_.Load());
+      marked_for_finalize_ = rhs.marked_for_finalize_;
+    }
+    return *this;
  }
  /**
@@ -299,10 +304,13 @@ class LlxScxRecord {
 * "Pragmatic Primitives for Non-blocking Data Structures" 
 * (Brown et al., 2013).
 *
- * \tparam MaxLinks Maximum number of active LL-dependencies per thread
 * \tparam UserData Type containing mutable fields
+ * \tparam ValuePool Type containing mutable fields
 */
-template< class UserData >
+template<
+  typename UserData,
+  typename ValuePool = embb::containers::LockFreeTreeValuePool< bool, false >
+>
 class LlxScx {
 private:  
@@ -335,8 +343,7 @@ class LlxScx {
    DataRecord_t * const data_record,
    /**< [IN] Pointer to data record to load */
    DataRecord_t & data,
-    /**< [OUT] Atomic snapshot of \c NumMutableFields fields in data
+    /**< [OUT] Atomic snapshot of data record */
-               record at given index */
    bool & finalized
    /**< [OUT] Indicating whether requested fields have been finalized */
  );
@@ -393,6 +400,11 @@ class LlxScx {
  size_t max_links_;
  /**
+   * Maximum number of threads engaging in operations on this LLX/SCX instance.
+   */
+  unsigned int max_threads_;
+  /**
   * Shared table containing for each r in V, a copy of r's info 
   * value in this thread's local table of LLX results.
   * 
@@ -402,23 +414,38 @@ class LlxScx {
   *    r_i in V -> *ScxRecord(thread_llx_results_[r_i].data_record.ScxInfo())
   *  }
   */
-  embb::containers::internal::FixedSizeList<ScxRecord_t> * info_fields_;
+// embb::containers::internal::FixedSizeList<ScxRecord_t> * info_fields_;
+  /**
+   * Shared table containing for each r in V, a copy of r's info 
+   * value in this thread's local table of LLX results.
+   *
+   *   thread_id -> {
+   *     r_1 in V -> *ScxRecord(thread_llx_results_[r_1].data_record.ScxInfo()),
+   *     ...
+   *     r_i in V -> *ScxRecord(thread_llx_results_[r_i].data_record.ScxInfo())
+   *   }
+   */
+  embb::containers::ObjectPool<
+    embb::containers::internal::FixedSizeList<ScxRecord_t>, ValuePool >
+      scx_record_list_pool_;
  /**
   * Thread-specific list of LLX results performed by the thread.
   */
-  embb::base::ThreadSpecificStorage<
+  embb::containers::internal::FixedSizeList<LlxResult> *
-    embb::containers::internal::FixedSizeList<LlxResult> >
+    thread_llx_results_;
-      thread_llx_results_;
  /**
   * Prevent default construction.
   */
  LlxScx();
  /**
   * Prevent copy construction.
   */
  LlxScx(const LlxScx &);
  /**
   * Prevent assignment.
   */

--- a/containers_cpp/test/llx_scx_test.cc
+++ b/containers_cpp/test/llx_scx_test.cc
@@ -37,14 +37,13 @@ using embb::containers::primitives::LlxScxRecord;
 using embb::containers::primitives::LlxScx;
 LlxScxTest::LlxScxTest() :
-  num_threads_(static_cast<int>
+  num_threads_(
-  (partest::TestSuite::GetDefaultNumThreads())) {
+    static_cast<int>(partest::TestSuite::GetDefaultNumThreads())) {
  CreateUnit("SerialTest").Add(&LlxScxTest::SerialTest, this);
 }
 void LlxScxTest::SerialTest() {
  typedef LlxScxTest::Node Node;
  // Global:
  LlxScx<Node> llxscx(3);
@@ -72,7 +71,7 @@ void LlxScxTest::SerialTest() {
  PT_ASSERT(llxscx.TryLoadLinked(&dr3, l3, finalized));
  PT_ASSERT(!finalized);
-  FixedSizeList< LlxScxRecord<Node> * > 
+  FixedSizeList< LlxScxRecord<Node> * >
    linked_deps(3);
  linked_deps.PushBack(&dr1);
  linked_deps.PushBack(&dr2);
@@ -92,6 +91,12 @@ void LlxScxTest::SerialTest() {
      linked_deps,   // V: dependencies, must be LL'd before
      finalize_deps  // R: Subsequence of V to be finalized
    ));
+  // Following LLX calls on finalized data records are
+  // expected to fail:
+  PT_ASSERT(!llxscx.TryLoadLinked(&dr2, l2, finalized));
+  PT_ASSERT(finalized);
+  PT_ASSERT(!llxscx.TryLoadLinked(&dr3, l3, finalized));
+  PT_ASSERT(finalized);
 }
 } // namespace test

--- a/containers_cpp/test/llx_scx_test.h
+++ b/containers_cpp/test/llx_scx_test.h
@@ -66,10 +66,19 @@ class LlxScxTest : public partest::TestCase {
      next_.Store(next_node);
    }
+    Node(const Node & other)
+    : value_(other.value_) {
+      next_.Store(other.next_.Load());
+      count_.Store(other.count_.Load());
+    }
    Node & operator=(const Node & rhs) {
-      count_.Store(rhs.count_.Load());
+      if (this != &rhs) {
-      next_.Store(rhs.next_.Load());
+        count_.Store(rhs.count_.Load());
-      value_ = rhs.value_;
+        next_.Store(rhs.next_.Load());
+        value_ = rhs.value_;
+      }
+      return *this;
    }
  };

--- a/containers_cpp/test/main.cc
+++ b/containers_cpp/test/main.cc
@@ -40,6 +40,7 @@
 #include "./stack_test.h"
 #include "./hazard_pointer_test.h"
 #include "./object_pool_test.h"
+#include "./llx_scx_test.h"
 #define COMMA ,
@@ -55,6 +56,7 @@ using embb::containers::test::HazardPointerTest;
 using embb::containers::test::QueueTest;
 using embb::containers::test::StackTest;
 using embb::containers::test::ObjectPoolTest;
+using embb::containers::test::LlxScxTest;
 PT_MAIN("Data Structures C++") {
  unsigned int max_threads = static_cast<unsigned int>(
@@ -70,6 +72,7 @@ PT_MAIN("Data Structures C++") {
  PT_RUN(StackTest< LockFreeStack<int> >);
  PT_RUN(ObjectPoolTest< LockFreeTreeValuePool<bool COMMA false > >);
  PT_RUN(ObjectPoolTest< WaitFreeArrayValuePool<bool COMMA false> >);
+  PT_RUN(LlxScxTest);
  PT_EXPECT(embb_get_bytes_allocated() == 0);
 }