Shared memory table scan (phase 2) (#1403)

* Insert new locators into type index

* Remove obsolete locators from type index

* Clean up comments

* cleanup

* rename method

* remove allocate_type()

* Address review feedback

Author: senderista

production/db/core/inc/db_helpers.hpp

@@ -32,19 +32,13 @@ inline common::gaia_id_t allocate_id()
     return ++(counters->last_id);
 }
 
-inline common::gaia_type_t allocate_type()
-{
-    counters_t* counters = gaia::db::get_counters();
-    return ++(counters->last_type_id);
-}
-
 inline gaia_txn_id_t allocate_txn_id()
 {
     counters_t* counters = gaia::db::get_counters();
     return ++(counters->last_txn_id);
 }
 
-inline gaia_locator_t allocate_locator()
+inline gaia_locator_t allocate_locator(common::gaia_type_t type)
 {
     counters_t* counters = gaia::db::get_counters();
 
@@ -53,7 +47,14 @@ inline gaia_locator_t allocate_locator()
         throw system_object_limit_exceeded_internal();
     }
 
-    return ++(counters->last_locator);
+    gaia_locator_t locator = ++(counters->last_locator);
+
+    type_index_t* type_index = get_type_index();
+    // Ignore failure if type is already registered.
+    type_index->register_type(type);
+    type_index->add_locator(type, locator);
+
+    return locator;
 }
 
 inline void update_locator(gaia_locator_t locator, gaia_offset_t offset)

production/db/core/inc/db_object_helpers.hpp

@@ -32,7 +32,7 @@ inline db_object_t* create_object(
     size_t ref_len = refs_count * sizeof(*refs);
     size_t total_len = obj_data_size + ref_len;
     gaia::db::hash_node_t* hash_node = db_hash_map::insert(id);
-    hash_node->locator = allocate_locator();
+    hash_node->locator = allocate_locator(type);
     gaia::db::allocate_object(hash_node->locator.load(), total_len);
     db_object_t* obj_ptr = locator_to_ptr(hash_node->locator.load());
     obj_ptr->id = id;
@@ -56,7 +56,7 @@ inline db_object_t* create_object(
     const void* obj_data)
 {
     gaia::db::hash_node_t* hash_node = db_hash_map::insert(id);
-    hash_node->locator = allocate_locator();
+    hash_node->locator = allocate_locator(type);
     gaia::db::allocate_object(hash_node->locator.load(), obj_data_size);
     db_object_t* obj_ptr = locator_to_ptr(hash_node->locator.load());
     obj_ptr->id = id;

production/db/core/inc/type_index.hpp

@@ -68,61 +68,64 @@ static_assert(
 static_assert(std::atomic<locator_list_node_t>::is_always_lock_free);
 
 // Holds a type ID and the head of its linked list of locators.
+// This structure is always updated atomically via CAS.
 struct type_index_entry_t
 {
     // The type ID of this locator list.
-    std::atomic<common::gaia_type_t::value_type> type;
+    common::gaia_type_t::value_type type;
 
     // The first locator in the list.
-    std::atomic<gaia_locator_t::value_type> first_locator;
+    gaia_locator_t::value_type first_locator;
 };
 
-static_assert(sizeof(type_index_entry_t) == 8, "Expected sizeof(type_index_entry_t) to be 8!");
+static_assert(
+    sizeof(type_index_entry_t) == sizeof(uint64_t),
+    "Expected type_index_entry_t to occupy 8 bytes!");
+static_assert(std::atomic<type_index_entry_t>::is_always_lock_free);
 
-// An index enabling efficient retrieval of all locators belonging to a
-// registered type. Each type ID maps to a linked list containing all locators
-// of that type.
+// An index enabling efficient retrieval of all locators of a registered type.
+// Each type ID maps to a linked list containing all locators of that type.
 struct type_index_t
 {
     // Mapping of registered type IDs to the heads of their locator lists.
-    type_index_entry_t type_index_entries[c_max_types];
-
-    // Atomic counter incremented during type registration to determine the
-    // index of the registered type in the `type_index_entries` array.
-    std::atomic<size_t> type_index_entries_count;
+    std::atomic<type_index_entry_t> type_index_entries[c_max_types];
 
     // Pool of nodes used for the linked lists representing the sets of locators
     // belonging to each registered type. The array index of each node
     // corresponds to the locator it represents.
     locator_list_node_t locator_lists_array[c_max_locators + 1];
 
+    // Returns a reference to the type index entry containing `type`.
+    // PRECONDITION: `type` has already been registered.
+    inline std::atomic<type_index_entry_t>& get_type_index_entry(common::gaia_type_t type);
+
     // Claims a slot in `type_index_entries` by atomically incrementing
     // `type_index_entries_count` (slots are not reused).
-    inline void register_type(common::gaia_type_t type);
+    // Returns true if `type` was not already registered, false otherwise.
+    inline bool register_type(common::gaia_type_t type);
 
-    // Returns the head of the locator list for the given type.
+    // Returns the head of the locator list for `type`.
     inline gaia_locator_t get_first_locator(common::gaia_type_t type);
 
-    // Changes the head of the locator list for the given type to
-    // `desired_locator`, if the head is still `expected_locator`.
-    // Returns true if the head is still `expected_locator`, false otherwise.
-    // This has CAS semantics because we need to retry if the head of the list
-    // changes during the operation.
-    inline bool set_first_locator(
+    // Changes the head of the locator list for `type` to `desired_locator`, if
+    // the head is still `expected_locator`. Returns true if the head is still
+    // `expected_locator`, false otherwise. (This has CAS semantics because we
+    // need to retry if the head of the list changes during the operation.)
+    inline bool try_set_first_locator(
         common::gaia_type_t type, gaia_locator_t expected_locator, gaia_locator_t desired_locator);
 
-    // Gets the list node corresponding to the given locator.
+    // Gets the list node corresponding to `locator`.
     inline locator_list_node_t* get_list_node(gaia_locator_t locator);
 
-    // Inserts the node for a locator at the head of the list for its type.
+    // Inserts the node for `locator` at the head of the list for `type`.
     // PRECONDITION: `type` is already registered in `type_index_entries`.
     // PRECONDITION: The list node for `locator` has not been previously used.
     // POSTCONDITION: `type_index_cursor_t(type).current_locator()` returns
     // `locator` (in the absence of concurrent invocations).
     inline void add_locator(common::gaia_type_t type, gaia_locator_t locator);
 
-    // Logically deletes the given locator from the list for its type.
-    // Returns false if the given locator was already logically deleted, true otherwise.
+    // Logically deletes `locator` from the list for its type.
+    // Returns false if `locator` was already logically deleted, true otherwise.
     // PRECONDITION: `locator` was previously allocated.
     // POSTCONDITION: returns false if list node for `locator` was already marked for deletion.
     inline bool delete_locator(gaia_locator_t locator);

production/db/core/inc/type_index.inc

@@ -36,6 +36,7 @@ bool locator_list_node_t::mark_for_deletion()
         {
             return false;
         }
+
         uint64_t desired_word{data_word | c_deleted_flag_mask};
         if (data_word.compare_exchange_strong(expected_word, desired_word))
         {
@@ -46,45 +47,93 @@ bool locator_list_node_t::mark_for_deletion()
     return true;
 }
 
-void type_index_t::register_type(common::gaia_type_t type)
+std::atomic<type_index_entry_t>& type_index_t::get_type_index_entry(common::gaia_type_t type)
 {
-    if (type_index_entries_count >= c_max_types)
-    {
-        throw type_limit_exceeded_internal();
-    }
-    type_index_entries[type_index_entries_count++].type = type;
-}
+    ASSERT_PRECONDITION(type.is_valid(), "Cannot call get_type_index_entry() with an invalid type!");
 
-gaia_locator_t type_index_t::get_first_locator(common::gaia_type_t type)
-{
     // REVIEW: With our current limit of 64 types, linear search should be
     // fine (the whole array is at most 8 cache lines, so should almost
     // always be in L1 cache), but with more types we'll eventually need
-    // sublinear search complexity.
-    for (size_t i = 0; i < type_index_entries_count; ++i)
+    // sublinear search complexity (e.g., a hash table).
+
+    // Scan until the end of the array. (We could stop at the first
+    // uninitialized entry, but the branch is likely not worth it for such a
+    // small array; see e.g.
+    // https://dirtyhandscoding.wordpress.com/2017/08/25/performance-comparison-linear-search-vs-binary-search/.)
+    for (size_t i = 0; i < std::size(type_index_entries); ++i)
     {
-        if (type_index_entries[i].type == type)
+        auto& entry_ref = type_index_entries[i];
+        auto entry_val = entry_ref.load();
+        if (entry_val.type == type)
         {
-            return gaia_locator_t(type_index_entries[i].first_locator);
+            return entry_ref;
         }
     }
-    ASSERT_UNREACHABLE("Type must be registered before accessing its locator list!");
+    // If we reach the end of the array without finding the entry for this type,
+    // the precondition has been violated.
+    ASSERT_UNREACHABLE("Type must be registered before calling get_type_index_entry()!");
 }
 
-bool type_index_t::set_first_locator(
-    common::gaia_type_t type, gaia_locator_t expected_locator, gaia_locator_t desired_locator)
+bool type_index_t::register_type(common::gaia_type_t type)
 {
-    gaia_locator_t::value_type expected_value = expected_locator.value();
-    gaia_locator_t::value_type desired_value = desired_locator.value();
-
-    for (size_t i = 0; i < type_index_entries_count; ++i)
+    ASSERT_PRECONDITION(type.is_valid(), "Cannot call register_type() with an invalid type!");
+
+    // This implements the insert operation on a lock-free set. Inserting a
+    // duplicate element is prevented by CAS semantics: each concurrent insert
+    // uses the next uninitialized array entry (there can be no "holes" in the
+    // array because entries only go from zero to nonzero, and we never scan
+    // past an entry initially read as zero until a CAS shows it is nonzero), so
+    // for any two concurrent inserts, one of them (the one that initializes the
+    // higher-indexed entry) must see the other's insert, and abort if has the
+    // same value.
+    //
+    // Scan until the first uninitialized entry or the end of the array,
+    // whichever comes first.
+    for (size_t i = 0; i < std::size(type_index_entries); ++i)
     {
-        if (type_index_entries[i].type == type)
+        auto& entry_ref = type_index_entries[i];
+        auto entry_val = entry_ref.load();
+        // The type was already registered.
+        if (entry_val.type == type)
         {
-            return type_index_entries[i].first_locator.compare_exchange_strong(expected_value, desired_value);
+            return false;
+        }
+
+        // Try to initialize the first uninitialized entry.
+        //
+        // REVIEW: This could technically be a relaxed load, because the
+        // subsequent CAS will detect a stale read. However, we don't currently
+        // specify non-default memory orderings anywhere, and I think we should
+        // only change this policy on the basis of profiling data.
+        if (entry_val.type == common::c_invalid_gaia_type)
+        {
+            type_index_entry_t expected_entry{common::c_invalid_gaia_type, c_invalid_gaia_locator};
+            type_index_entry_t desired_entry{type, c_invalid_gaia_locator};
+
+            // If the CAS succeeds, we are done, otherwise try the next entry.
+            if (entry_ref.compare_exchange_strong(expected_entry, desired_entry))
+            {
+                return true;
+            }
         }
     }
-    ASSERT_UNREACHABLE("Type must be registered before accessing its locator list!");
+
+    // We reached the end of the array without finding an uninitialized entry.
+    throw type_limit_exceeded_internal();
+}
+
+gaia_locator_t type_index_t::get_first_locator(common::gaia_type_t type)
+{
+    return get_type_index_entry(type).load().first_locator;
+}
+
+bool type_index_t::try_set_first_locator(
+    common::gaia_type_t type, gaia_locator_t expected_locator, gaia_locator_t desired_locator)
+{
+    type_index_entry_t expected_entry{type, expected_locator};
+    type_index_entry_t desired_entry{type, desired_locator};
+
+    return get_type_index_entry(type).compare_exchange_strong(expected_entry, desired_entry);
 }
 
 locator_list_node_t* type_index_t::get_list_node(gaia_locator_t locator)
@@ -127,7 +176,7 @@ void type_index_t::add_locator(common::gaia_type_t type, gaia_locator_t locator)
 
         // Now try to point the list head to the new node, retrying if it
         // was concurrently pointed to another node.
-        if (set_first_locator(type, first_locator, locator))
+        if (try_set_first_locator(type, first_locator, locator))
         {
             break;
         }

production/db/core/inc/type_index_cursor.inc

@@ -68,7 +68,7 @@ bool type_index_cursor_t::unlink_for_deletion()
     // Handle the special case where the first marked node is at the head of the list.
     if (!prev_node)
     {
-        return m_type_index->set_first_locator(m_type, unlinked_locator, current_locator());
+        return m_type_index->try_set_first_locator(m_type, unlinked_locator, current_locator());
     }
 
     // Otherwise, set the previous node to point to the new current locator

production/db/core/src/db_server.cpp

@@ -2006,22 +2006,22 @@ void server_t::gc_txn_log_from_offset(log_offset_t log_offset, bool is_committed
 {
     txn_log_t* txn_log = get_txn_log_from_offset(log_offset);
 
+    // Remove index entries that might be referencing obsolete versions before
+    // actually deallocating them.
+    bool deallocate_new_offsets = !is_committed;
+    index::index_builder_t::gc_indexes_from_txn_log(txn_log, deallocate_new_offsets);
+
     // If the txn committed, we deallocate only undo versions, because the
     // redo versions may still be visible after the txn has fallen
     // behind the watermark. If the txn aborted, then we deallocate only
     // redo versions, because the undo versions may still be visible. Note
     // that we could deallocate intermediate versions (i.e., those
     // superseded within the same txn) immediately, but we do it here
     // for simplicity.
-    bool deallocate_new_offsets = !is_committed;
-
-    // Remove index entries that might be referencing obsolete versions before
-    // actually deallocating them.
-    index::index_builder_t::gc_indexes_from_txn_log(txn_log, deallocate_new_offsets);
-    deallocate_txn_log(txn_log, deallocate_new_offsets);
+    deallocate_txn_log(txn_log, is_committed);
 }
 
-void server_t::deallocate_txn_log(txn_log_t* txn_log, bool deallocate_new_offsets)
+void server_t::deallocate_txn_log(txn_log_t* txn_log, bool is_committed)
 {
     ASSERT_PRECONDITION(txn_log, "txn_log must be a valid address!");
     ASSERT_PRECONDITION(
@@ -2043,17 +2043,16 @@ void server_t::deallocate_txn_log(txn_log_t* txn_log, bool deallocate_new_offset
         // txn log of an aborted txn is after it falls behind the watermark,
         // because at that point it cannot be in the conflict window of any
         // committing txn.
-        gaia_offset_t offset_to_free = deallocate_new_offsets
-            ? log_record->new_offset
-            : log_record->old_offset;
+        gaia_offset_t offset_to_free = is_committed
+            ? log_record->old_offset
+            : log_record->new_offset;
 
         // If we're gc-ing the old version of an object that is being deleted,
         // then request the deletion of its locator from the corresponding record list.
-        if (!deallocate_new_offsets && !log_record->new_offset.is_valid())
+        if (is_committed && log_record->operation() == gaia_operation_t::remove)
         {
             // Get the old object data to extract its type.
             db_object_t* db_object = offset_to_ptr(log_record->old_offset);
-
             // Retrieve the record_list_t instance corresponding to the type.
             std::shared_ptr<record_list_t> record_list = record_list_manager_t::get()->get_record_list(db_object->type);
 
@@ -2065,6 +2064,18 @@ void server_t::deallocate_txn_log(txn_log_t* txn_log, bool deallocate_new_offset
         {
             deallocate_object(offset_to_free);
         }
+
+        // For committed txns, we need to remove any deleted locators from the
+        // type index. For aborted or rolled-back txns, we need to remove any
+        // allocated locators from the type index.
+        bool is_locator_removal_committed = is_committed && log_record->operation() == gaia_operation_t::remove;
+        bool is_locator_creation_aborted = !is_committed && log_record->operation() == gaia_operation_t::create;
+        if (is_locator_removal_committed || is_locator_creation_aborted)
+        {
+            type_index_t* type_index = get_type_index();
+            bool has_succeeded = type_index->delete_locator(log_record->locator);
+            ASSERT_INVARIANT(has_succeeded, "A locator cannot be deleted twice!");
+        }
     }
 
     // We've deallocated all garbage versions, and we have no shared references,

production/db/core/src/gaia_ptr_client.cpp

@@ -121,7 +121,7 @@ gaia_ptr_t gaia_ptr_t::create_no_txn(gaia_id_t id, gaia_type_t type, reference_o
     // TODO: this constructor allows creating a gaia_ptr_t in an invalid state;
     //  the db_object_t should either be initialized before and passed in
     //  or it should be initialized inside the constructor.
-    gaia_locator_t locator = allocate_locator();
+    gaia_locator_t locator = allocate_locator(type);
     hash_node_t* hash_node = db_hash_map::insert(id);
     hash_node->locator = locator;
     allocate_object(locator, total_payload_size);