Garbage collection fixes (again) (#1428)

Author: mbutrovich

src/common/container/cuckoo_map.cpp

@@ -32,6 +32,9 @@ class IndexMetric;
 
 class StatementCache;
 
+template <typename T>
+class LockFreeQueue;
+
 CUCKOO_MAP_TEMPLATE_ARGUMENTS
 CUCKOO_MAP_TYPE::CuckooMap() {}
 
@@ -125,4 +128,8 @@ template class CuckooMap<std::shared_ptr<oid_t>, std::shared_ptr<oid_t>>;
 // Used in StatementCacheManager
 template class CuckooMap<StatementCache *, StatementCache *>;
 
+// Used in TransactionLevelGCManager
+template class CuckooMap<oid_t,
+                         std::shared_ptr<LockFreeQueue<ItemPointer>>>;
+
 }  // namespace peloton

src/common/init.cpp

@@ -52,7 +52,7 @@ void PelotonInit::Initialize() {
   threadpool::MonoQueuePool::GetExecutionInstance().Startup();
 
   int parallelism = (CONNECTION_THREAD_COUNT + 3) / 4;
-  storage::DataTable::SetActiveTileGroupCount(parallelism);
+  storage::DataTable::SetDefaultActiveTileGroupCount(parallelism);
   storage::DataTable::SetActiveIndirectionArrayCount(parallelism);
 
   // start epoch.

src/common/internal_types.cpp

@@ -3001,8 +3001,8 @@ std::string GCVersionTypeToString(GCVersionType type) {
     case GCVersionType::ABORT_UPDATE: {
       return "ABORT_UPDATE";
     }
-    case GCVersionType::ABORT_DELETE: {
-      return "ABORT_DELETE";
+    case GCVersionType::TOMBSTONE: {
+      return "TOMBSTONE";
     }
     case GCVersionType::ABORT_INSERT: {
       return "ABORT_INSERT";
@@ -3031,8 +3031,8 @@ GCVersionType StringToGCVersionType(const std::string &str) {
     return GCVersionType::COMMIT_INS_DEL;
   } else if (upper_str == "ABORT_UPDATE") {
     return GCVersionType::ABORT_UPDATE;
-  } else if (upper_str == "ABORT_DELETE") {
-    return GCVersionType::ABORT_DELETE;
+  } else if (upper_str == "TOMBSTONE") {
+    return GCVersionType::TOMBSTONE;
   } else if (upper_str == "ABORT_INSERT") {
     return GCVersionType::ABORT_INSERT;
   } else if (upper_str == "ABORT_INS_DEL") {

src/concurrency/timestamp_ordering_transaction_manager.cpp

@@ -709,6 +709,9 @@ ResultType TimestampOrderingTransactionManager::CommitTransaction(
       gc_set->operator[](tile_group_id)[tuple_slot] =
           GCVersionType::COMMIT_DELETE;
 
+      gc_set->operator[](new_version.block)[new_version.offset] =
+          GCVersionType::TOMBSTONE;
+
       log_manager.LogDelete(ItemPointer(tile_group_id, tuple_slot));
 
     } else if (tuple_entry.second == RWType::INSERT) {
@@ -827,9 +830,11 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
       // before we unlink the aborted version from version list
       ItemPointer *index_entry_ptr =
           tile_group_header->GetIndirection(tuple_slot);
-      UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
-          index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
-      PELOTON_ASSERT(res == true);
+      if (index_entry_ptr) {
+        UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
+            index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
+        PELOTON_ASSERT(res == true);
+      }
       //////////////////////////////////////////////////
 
       // we should set the version before releasing the lock.
@@ -875,9 +880,11 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
       // before we unlink the aborted version from version list
       ItemPointer *index_entry_ptr =
           tile_group_header->GetIndirection(tuple_slot);
-      UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
-          index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
-      PELOTON_ASSERT(res == true);
+      if (index_entry_ptr) {
+        UNUSED_ATTRIBUTE auto res = AtomicUpdateItemPointer(
+            index_entry_ptr, ItemPointer(tile_group_id, tuple_slot));
+        PELOTON_ASSERT(res == true);
+      }
       //////////////////////////////////////////////////
 
       // we should set the version before releasing the lock.
@@ -895,7 +902,7 @@ ResultType TimestampOrderingTransactionManager::AbortTransaction(
 
       // add the version to gc set.
       gc_set->operator[](new_version.block)[new_version.offset] =
-          GCVersionType::ABORT_DELETE;
+          GCVersionType::TOMBSTONE;
 
     } else if (tuple_entry.second == RWType::INSERT) {
       tile_group_header->SetBeginCommitId(tuple_slot, MAX_CID);