Performance Fixes (#1375)

* Change RWSet to use tbb::concurrent_unordered_map

* Modify tile_group_locator to use cuckoohash

* Style Fix

* Remove unused attribute

Author: poojanilangekar

src/catalog/manager.cpp

@@ -25,6 +25,8 @@ std::shared_ptr<storage::TileGroup> Manager::empty_tile_group_;
 
 std::shared_ptr<storage::IndirectionArray> Manager::empty_indirection_array_;
 
+Manager::Manager() : tile_group_locator_(DEFAULT_LOCATOR_SIZE) {}
+
 Manager &Manager::GetInstance() {
   static Manager manager;
   return manager;
@@ -37,24 +39,24 @@ Manager &Manager::GetInstance() {
 void Manager::AddTileGroup(const oid_t oid,
                            std::shared_ptr<storage::TileGroup> location) {
   // add/update the catalog reference to the tile group
-  tile_group_locator_[oid] = location;
+  tile_group_locator_.Upsert(oid, location);
 }
 
 void Manager::DropTileGroup(const oid_t oid) {
   // drop the catalog reference to the tile group
-  tile_group_locator_[oid] = empty_tile_group_;
+  tile_group_locator_.Erase(oid);
 }
 
 std::shared_ptr<storage::TileGroup> Manager::GetTileGroup(const oid_t oid) {
-  auto iter = tile_group_locator_.find(oid);
-  if (iter == tile_group_locator_.end()) {
-    return empty_tile_group_;
+  std::shared_ptr<storage::TileGroup> location;
+  if (tile_group_locator_.Find(oid, location)) {
+    return location;
   }
-  return iter->second;
+  return empty_tile_group_;
 }
 
 // used for logging test
-void Manager::ClearTileGroup() { tile_group_locator_.clear(); }
+void Manager::ClearTileGroup() { tile_group_locator_.Clear(); }
 
 void Manager::AddIndirectionArray(
     const oid_t oid, std::shared_ptr<storage::IndirectionArray> location) {

src/common/container/cuckoo_map.cpp

@@ -48,17 +48,23 @@ bool CUCKOO_MAP_TYPE::Insert(const KeyType &key, ValueType value) {
   return status;
 }
 
+CUCKOO_MAP_TEMPLATE_ARGUMENTS
+void CUCKOO_MAP_TYPE::Upsert(const KeyType &key, ValueType value) {
+  auto update_fn = [](UNUSED_ATTRIBUTE ValueType &value) { return; };
+  cuckoo_map.upsert(key, update_fn, value);
+}
+
 CUCKOO_MAP_TEMPLATE_ARGUMENTS
 bool CUCKOO_MAP_TYPE::Update(const KeyType &key, ValueType value) {
   auto status = cuckoo_map.update(key, value);
+  LOG_TRACE("update status : %d", status);
   return status;
 }
 
 CUCKOO_MAP_TEMPLATE_ARGUMENTS
 bool CUCKOO_MAP_TYPE::Erase(const KeyType &key) {
   auto status = cuckoo_map.erase(key);
   LOG_TRACE("erase status : %d", status);
-
   return status;
 }
 
@@ -102,6 +108,7 @@ CUCKOO_MAP_TYPE::GetConstIterator() const {
 // Explicit template instantiation
 template class CuckooMap<uint32_t, uint32_t>;
 
+// Used in Catalog::Manager
 template class CuckooMap<oid_t, std::shared_ptr<storage::TileGroup>>;
 
 // Used in Shared Pointer test and iterator test
@@ -118,8 +125,4 @@ template class CuckooMap<std::shared_ptr<oid_t>, std::shared_ptr<oid_t>>;
 // Used in StatementCacheManager
 template class CuckooMap<StatementCache *, StatementCache *>;
 
-// Used in InternalTypes
-template class CuckooMap<ItemPointer, RWType, ItemPointerHasher,
-                         ItemPointerComparator>;
-
 }  // namespace peloton

src/concurrency/timestamp_ordering_transaction_manager.cpp

@@ -760,7 +760,7 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
   oid_t database_id = 0;
   if (static_cast<StatsType>(settings::SettingsManager::GetInt(
           settings::SettingId::stats_mode)) != StatsType::INVALID) {
-    for (const auto &tuple_entry : rw_set.GetConstIterator()) {
+    for (const auto &tuple_entry : rw_set) {
       // Call the GetConstIterator() function to explicitly lock the cuckoohash
       // and initilaize the iterator
       const auto tile_group_id = tuple_entry.first.block;
@@ -779,7 +779,7 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
 
   // TODO (Pooja): This might be inefficient since we will have to get the
   // tile_group_header for each entry. Check if this needs to be consolidated
-  for (const auto &tuple_entry : rw_set.GetConstIterator()) {
+  for (const auto &tuple_entry : rw_set) {
     ItemPointer item_ptr = tuple_entry.first;
     oid_t tile_group_id = item_ptr.block;
     oid_t tuple_slot = item_ptr.offset;
@@ -939,7 +939,7 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
   oid_t database_id = 0;
   if (static_cast<StatsType>(settings::SettingsManager::GetInt(
           settings::SettingId::stats_mode)) != StatsType::INVALID) {
-    for (const auto &tuple_entry : rw_set.GetConstIterator()) {
+    for (const auto &tuple_entry : rw_set) {
       // Call the GetConstIterator() function to explicitly lock the cuckoohash
       // and initilaize the iterator
       const auto tile_group_id = tuple_entry.first.block;
@@ -953,7 +953,7 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
   // Iterate through each item pointer in the read write set
   // TODO (Pooja): This might be inefficient since we will have to get the
   // tile_group_header for each entry. Check if this needs to be consolidated
-  for (const auto &tuple_entry : rw_set.GetConstIterator()) {
+  for (const auto &tuple_entry : rw_set) {
     ItemPointer item_ptr = tuple_entry.first;
     oid_t tile_group_id = item_ptr.block;
     oid_t tuple_slot = item_ptr.offset;

src/concurrency/transaction_context.cpp

@@ -50,25 +50,14 @@ namespace concurrency {
 
 TransactionContext::TransactionContext(const size_t thread_id,
                                        const IsolationLevelType isolation,
-                                       const cid_t &read_id)
-    : rw_set_(INTITIAL_RW_SET_SIZE) {
+                                       const cid_t &read_id) {
   Init(thread_id, isolation, read_id);
 }
 
 TransactionContext::TransactionContext(const size_t thread_id,
                                        const IsolationLevelType isolation,
                                        const cid_t &read_id,
-                                       const cid_t &commit_id)
-    : rw_set_(INTITIAL_RW_SET_SIZE) {
-  Init(thread_id, isolation, read_id, commit_id);
-}
-
-TransactionContext::TransactionContext(const size_t thread_id,
-                                       const IsolationLevelType isolation,
-                                       const cid_t &read_id,
-                                       const cid_t &commit_id,
-                                       const size_t rw_set_size)
-    : rw_set_(rw_set_size) {
+                                       const cid_t &commit_id) {
   Init(thread_id, isolation, read_id, commit_id);
 }
 
@@ -95,107 +84,95 @@ void TransactionContext::Init(const size_t thread_id,
 
   insert_count_ = 0;
 
-  rw_set_.Clear();
   gc_set_.reset(new GCSet());
   gc_object_set_.reset(new GCObjectSet());
 
   on_commit_triggers_.reset();
 }
 
 RWType TransactionContext::GetRWType(const ItemPointer &location) {
-  RWType rw_type;
+  RWType rw_type = RWType::INVALID;
 
-  if (!rw_set_.Find(location, rw_type)) {
-    rw_type = RWType::INVALID;
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
+    return rw_set_it->second;
   }
   return rw_type;
 }
 
 void TransactionContext::RecordRead(const ItemPointer &location) {
-  RWType rw_type;
 
-  if (rw_set_.Find(location, rw_type)) {
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
+    UNUSED_ATTRIBUTE RWType rw_type = rw_set_it->second;
     PELOTON_ASSERT(rw_type != RWType::DELETE && rw_type != RWType::INS_DEL);
     return;
-  } else {
-    rw_set_.Insert(location, RWType::READ);
   }
+  rw_set_[location] = RWType::READ;
 }
 
 void TransactionContext::RecordReadOwn(const ItemPointer &location) {
-  RWType rw_type;
-
-  if (rw_set_.Find(location, rw_type)) {
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
+    RWType rw_type = rw_set_it->second;
     PELOTON_ASSERT(rw_type != RWType::DELETE && rw_type != RWType::INS_DEL);
-    if (rw_type == RWType::READ) {
-      rw_set_.Update(location, RWType::READ_OWN);
+    if (rw_type != RWType::READ) {
+      return;
     }
-  } else {
-    rw_set_.Insert(location, RWType::READ_OWN);
   }
+  rw_set_[location] = RWType::READ_OWN;
 }
 
 void TransactionContext::RecordUpdate(const ItemPointer &location) {
-  RWType rw_type;
-
-  if (rw_set_.Find(location, rw_type)) {
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
+    RWType rw_type = rw_set_it->second;
     if (rw_type == RWType::READ || rw_type == RWType::READ_OWN) {
       is_written_ = true;
-      rw_set_.Update(location, RWType::UPDATE);
-      return;
-    }
-    if (rw_type == RWType::UPDATE) {
-      return;
-    }
-    if (rw_type == RWType::INSERT) {
+    } else if (rw_type == RWType::UPDATE || rw_type == RWType::INSERT) {
       return;
-    }
-    if (rw_type == RWType::DELETE) {
+    } else {
+      // DELETE or INS_DELETE
       PELOTON_ASSERT(false);
       return;
     }
-    PELOTON_ASSERT(false);
-  } else {
-    rw_set_.Insert(location, RWType::UPDATE);
   }
+  rw_set_[location] = RWType::UPDATE;
 }
 
 void TransactionContext::RecordInsert(const ItemPointer &location) {
-  if (IsInRWSet(location)) {
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
     PELOTON_ASSERT(false);
-  } else {
-    rw_set_.Insert(location, RWType::INSERT);
-    ++insert_count_;
+    return;
   }
+  rw_set_[location] = RWType::INSERT;
+  ++insert_count_;
 }
 
 bool TransactionContext::RecordDelete(const ItemPointer &location) {
-  RWType rw_type;
+  const auto rw_set_it = rw_set_.find(location);
+  if (rw_set_it != rw_set_.end()) {
+    RWType rw_type = rw_set_it->second;
 
-  if (rw_set_.Find(location, rw_type)) {
     if (rw_type == RWType::READ || rw_type == RWType::READ_OWN) {
-      rw_set_.Update(location, RWType::DELETE);
-      // record write
+      rw_set_[location] = RWType::DELETE;
       is_written_ = true;
       return false;
-    }
-    if (rw_type == RWType::UPDATE) {
-      rw_set_.Update(location, RWType::DELETE);
+    } else if (rw_type == RWType::UPDATE) {
+      rw_set_[location] = RWType::DELETE;
       return false;
-    }
-    if (rw_type == RWType::INSERT) {
-      rw_set_.Update(location, RWType::INS_DEL);
+    } else if (rw_type == RWType::INSERT) {
+      rw_set_[location] = RWType::INS_DEL;
       --insert_count_;
       return true;
-    }
-    if (rw_type == RWType::DELETE) {
+    } else {
+      // DELETE and INS_DEL
       PELOTON_ASSERT(false);
       return false;
     }
-    PELOTON_ASSERT(false);
-  } else {
-    rw_set_.Insert(location, RWType::DELETE);
   }
+  rw_set_[location] = RWType::DELETE;
   return false;
 }
 

src/include/catalog/manager.h

@@ -20,10 +20,13 @@
 #include <memory>
 #include "common/internal_types.h"
 
-#include "common/macros.h"
+#include "common/container/cuckoo_map.h"
 #include "common/internal_types.h"
-#include "common/container/lock_free_array.h"
+#include "common/macros.h"
 #include "tbb/concurrent_unordered_map.h"
+
+#define DEFAULT_LOCATOR_SIZE 1024
+
 namespace peloton {
 
 namespace storage {
@@ -39,7 +42,7 @@ namespace catalog {
 
 class Manager {
  public:
-  Manager() {}
+  Manager();
 
   // Singleton
   static Manager &GetInstance();
@@ -94,8 +97,7 @@ class Manager {
   //===--------------------------------------------------------------------===//
   std::atomic<oid_t> tile_group_oid_ = ATOMIC_VAR_INIT(START_OID);
 
-  tbb::concurrent_unordered_map<oid_t, std::shared_ptr<storage::TileGroup>>
-      tile_group_locator_;
+  CuckooMap<oid_t, std::shared_ptr<storage::TileGroup>> tile_group_locator_;
   static std::shared_ptr<storage::TileGroup> empty_tile_group_;
 
   //===--------------------------------------------------------------------===//

src/include/common/container/cuckoo_map.h

@@ -51,6 +51,10 @@ class CuckooMap {
   // Inserts a item
   bool Insert(const KeyType &key, ValueType value);
 
+  // Inserts the item if not present, updates value otherwise
+  // Upsert operations always succeed
+  void Upsert(const KeyType &key, ValueType value);
+
   // Extracts item with high priority
   bool Update(const KeyType &key, ValueType value);
 

src/include/common/internal_types.h

@@ -24,12 +24,12 @@
 #include <unordered_set>
 #include <vector>
 
+#include "tbb/concurrent_unordered_map.h"
 #include "tbb/concurrent_unordered_set.h"
 #include "tbb/concurrent_vector.h"
 
 #include "common/logger.h"
 #include "common/macros.h"
-#include "container/cuckoo_map.h"
 #include "parser/pg_trigger.h"
 #include "type/type_id.h"
 
@@ -1213,7 +1213,8 @@ RWType StringToRWType(const std::string &str);
 std::ostream &operator<<(std::ostream &os, const RWType &type);
 
 // ItemPointer -> type
-typedef CuckooMap<ItemPointer, RWType, ItemPointerHasher, ItemPointerComparator>
+typedef tbb::concurrent_unordered_map<ItemPointer, RWType, ItemPointerHasher,
+                                      ItemPointerComparator>
     ReadWriteSet;
 
 typedef tbb::concurrent_unordered_set<ItemPointer, ItemPointerHasher,

src/include/concurrency/transaction_context.h

@@ -24,8 +24,6 @@
 #include "common/printable.h"
 #include "common/internal_types.h"
 
-#define INTITIAL_RW_SET_SIZE 64
-
 namespace peloton {
 
 namespace trigger {
@@ -51,10 +49,6 @@ class TransactionContext : public Printable {
 
   TransactionContext(const size_t thread_id, const IsolationLevelType isolation,
               const cid_t &read_id, const cid_t &commit_id);
- 
-  TransactionContext(const size_t thread_id, const IsolationLevelType isolation,
-              const cid_t &read_id, const cid_t &commit_id, 
-              const size_t read_write_set_size);
 
   /**
    * @brief      Destroys the object.
@@ -201,7 +195,7 @@ class TransactionContext : public Printable {
    * @return     True if in rw set, False otherwise.
    */
   bool IsInRWSet(const ItemPointer &location) {
-    return rw_set_.Contains(location);
+    return (rw_set_.find(location) != rw_set_.end());
   }
 
   /**