Delegate Cleanables

Makefile

@@ -324,6 +324,7 @@ TESTS = \
 	db_properties_test \
 	db_table_properties_test \
 	autovector_test \
+	cleanable_test \
 	column_family_test \
 	table_properties_collector_test \
 	arena_test \
@@ -1147,6 +1148,9 @@ full_filter_block_test: table/full_filter_block_test.o $(LIBOBJECTS) $(TESTHARNE
 log_test: db/log_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 
+cleanable_test: table/cleanable_test.o $(LIBOBJECTS) $(TESTHARNESS)
+	$(AM_LINK)
+
 table_test: table/table_test.o $(LIBOBJECTS) $(TESTHARNESS)
 	$(AM_LINK)
 

db/pinned_iterators_manager.h

@@ -16,7 +16,7 @@ namespace rocksdb {
 // PinnedIteratorsManager will be notified whenever we need to pin an Iterator
 // and it will be responsible for deleting pinned Iterators when they are
 // not needed anymore.
-class PinnedIteratorsManager {
+class PinnedIteratorsManager : public Cleanable {
  public:
   PinnedIteratorsManager() : pinning_enabled(false) {}
   ~PinnedIteratorsManager() {

include/rocksdb/iterator.h

@@ -36,6 +36,7 @@ class Cleanable {
   // not abstract and therefore clients should not override it.
   typedef void (*CleanupFunction)(void* arg1, void* arg2);
   void RegisterCleanup(CleanupFunction function, void* arg1, void* arg2);
+  void DelegateCleanupsTo(Cleanable* other);
 
  protected:
   struct Cleanup {
@@ -45,6 +46,8 @@ class Cleanable {
     Cleanup* next;
   };
   Cleanup cleanup_;
+  // It also becomes the owner of c
+  void RegisterCleanup(Cleanup* c);
 };
 
 class Iterator : public Cleanable {

table/block_based_table_reader.cc

@@ -1559,15 +1559,8 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
         break;
       } else {
         BlockIter stack_biter;
-        if (pin_blocks) {
-          // We need to create the BlockIter on heap because we may need to
-          // pin it if we encounterd merge operands
-          biter = static_cast<BlockIter*>(
-              NewDataBlockIterator(rep_, read_options, iiter.value()));
-        } else {
-          biter = &stack_biter;
-          NewDataBlockIterator(rep_, read_options, iiter.value(), biter);
-        }
+        biter = &stack_biter;
+        NewDataBlockIterator(rep_, read_options, iiter.value(), biter);
 
         if (read_options.read_tier == kBlockCacheTier &&
             biter->status().IsIncomplete()) {
@@ -1596,22 +1589,12 @@ Status BlockBasedTable::Get(const ReadOptions& read_options, const Slice& key,
         }
         s = biter->status();
 
-        if (pin_blocks) {
-          if (get_context->State() == GetContext::kMerge) {
-            // Pin blocks as long as we are merging
-            pinned_iters_mgr->PinIterator(biter);
-          } else {
-            delete biter;
-          }
-          biter = nullptr;
-        } else {
-          // biter is on stack, Nothing to clean
+        if (pin_blocks && get_context->State() == GetContext::kMerge) {
+          // Pin blocks as long as we are merging
+          biter->DelegateCleanupsTo(pinned_iters_mgr);
         }
       }
     }
-    if (pin_blocks && biter != nullptr) {
-      delete biter;
-    }
     if (s.ok()) {
       s = iiter.status();
     }

table/iterator.cc

@@ -31,6 +31,37 @@ Cleanable::~Cleanable() {
   }
 }
 
+// TODO(myabandeh): if the list is too long we should maintain a tail pointer
+void Cleanable::DelegateCleanupsTo(Cleanable* other) {
+  assert(other != nullptr);
+  if (cleanup_.function == nullptr) {
+    return;
+  }
+  Cleanup* c = &cleanup_;
+  other->RegisterCleanup(c->function, c->arg1, c->arg2);
+  c = c->next;
+  while (c != nullptr) {
+    Cleanup* next = c->next;
+    other->RegisterCleanup(c);
+    c = next;
+  }
+  cleanup_.function = nullptr;
+  cleanup_.next = nullptr;
+}
+
+void Cleanable::RegisterCleanup(Cleanable::Cleanup* c) {
+  assert(c != nullptr);
+  if (cleanup_.function == nullptr) {
+    cleanup_.function = c->function;
+    cleanup_.arg1 = c->arg1;
+    cleanup_.arg2 = c->arg2;
+    delete c;
+  } else {
+    c->next = cleanup_.next;
+    cleanup_.next = c;
+  }
+}
+
 void Cleanable::RegisterCleanup(CleanupFunction func, void* arg1, void* arg2) {
   assert(func != nullptr);
   Cleanup* c;