in progress

db/blob/blob_source.cc

@@ -144,7 +144,7 @@ void BlobSource::PinCachedBlob(CacheHandleGuard<BlobContents>* cached_blob,
   constexpr Cleanable* cleanable = nullptr;
   value->PinSlice(cached_blob->GetValue()->data(), cleanable);
 
-  cached_blob->TransferTo(value);
+  cached_blob->TransferTo(value->Cleaner());
 }
 
 void BlobSource::PinOwnedBlob(std::unique_ptr<BlobContents>* owned_blob,

db/db_impl/db_impl.cc

@@ -2331,10 +2331,10 @@ Status DBImpl::GetImpl(const ReadOptions& read_options, const Slice& key,
                   sl, nullptr /* cleanable */);
               if (i == state->merge_context.GetOperands().size() - 1) {
                 shared_cleanable.MoveAsCleanupTo(
-                    get_impl_options.merge_operands);
+                    get_impl_options.merge_operands->Cleaner());
               } else {
                 shared_cleanable.RegisterCopyWith(
-                    get_impl_options.merge_operands);
+                    get_impl_options.merge_operands->Cleaner());
               }
               get_impl_options.merge_operands++;
             }

include/rocksdb/cleanable.h

@@ -33,6 +33,9 @@ class Cleanable {
   // not abstract and therefore clients should not override it.
   using CleanupFunction = void (*)(void* arg1, void* arg2);
 
+  inline Cleanable(CleanupFunction function, void* arg1, void* arg2)
+    : cleanup_{function, arg1, arg2, nullptr} {}
+
   // Add another Cleanup to the list
   void RegisterCleanup(CleanupFunction function, void* arg1, void* arg2);
 
@@ -75,6 +78,8 @@ class Cleanable {
       }
     }
   }
+ public:
+  inline const Cleanup& GetCleanup() { return cleanup_; }
 };
 
 // A copyable, reference-counted pointer to a simple Cleanable that only

include/rocksdb/slice.h

@@ -159,10 +159,14 @@ class Slice {
  * to avoid memcpy by having the PinnableSlice object referring to the data
  * that is locked in the memory and release them after the data is consumed.
  */
-class PinnableSlice : public Slice, public Cleanable {
+class PinnableSlice : public Slice {
  public:
-  PinnableSlice() { buf_ = &self_space_; }
-  explicit PinnableSlice(std::string* buf) { buf_ = buf; }
+  using CleanupFunction = Cleanable::CleanupFunction;
+  PinnableSlice() { handle_ = kNotPinned; }
+  explicit PinnableSlice(std::string* buf) {
+    handle_ = size_t(buf) | kNotPinned;
+  }
+  ~PinnableSlice();
 
   PinnableSlice(PinnableSlice&& other);
   PinnableSlice& operator=(PinnableSlice&& other);
@@ -171,89 +175,66 @@ class PinnableSlice : public Slice, public Cleanable {
   PinnableSlice(PinnableSlice&) = delete;
   PinnableSlice& operator=(PinnableSlice&) = delete;
 
-  inline void PinSlice(const Slice& s, CleanupFunction f, void* arg1,
-                       void* arg2) {
-    assert(!pinned_);
-    pinned_ = true;
-    data_ = s.data();
-    size_ = s.size();
-    RegisterCleanup(f, arg1, arg2);
-    assert(pinned_);
-  }
+  void PinSlice(const Slice&, CleanupFunction, void* arg1, void* arg2 = nullptr);
+  void PinSlice(const Slice&, Cleanable*);
 
-  inline void PinSlice(const Slice& s, Cleanable* cleanable) {
-    assert(!pinned_);
-    pinned_ = true;
-    data_ = s.data();
-    size_ = s.size();
-    if (cleanable != nullptr) {
-      cleanable->DelegateCleanupsTo(this);
-    }
-    assert(pinned_);
-  }
+  Cleanable* Cleaner();
 
   inline void SyncToString(std::string* s) const {
-    assert(s == buf_);
-    if (pinned_) {
+    if (IsPinned()) {
       s->assign(data_, size_);
     } else {
       assert(size_ == s->size());
       assert(data_ == s->data() || size_ == 0);
     }
   }
-  inline void SyncToString() const { SyncToString(buf_); }
-
-  inline void PinSelf(const Slice& slice) {
-    assert(!pinned_);
-    buf_->assign(slice.data(), slice.size());
-    data_ = buf_->data();
-    size_ = buf_->size();
-    assert(!pinned_);
-  }
-
-  inline void PinSelf() {
-    assert(!pinned_);
-    data_ = buf_->data();
-    size_ = buf_->size();
-    assert(!pinned_);
-  }
+  void SyncToString() const;
+  void PinSelf(const Slice& slice);
+  void PinSelf();
 
   void remove_suffix(size_t n) {
     assert(n <= size());
-    if (pinned_) {
+    if (IsPinned()) {
       size_ -= n;
     } else {
-      buf_->erase(size() - n, n);
+      GetSelf()->erase(size() - n, n);
       PinSelf();
     }
   }
 
   void remove_prefix(size_t n) {
     assert(n <= size());
-    if (pinned_) {
+    if (IsPinned()) {
       data_ += n;
       size_ -= n;
     } else {
-      buf_->erase(0, n);
+      GetSelf()->erase(0, n);
       PinSelf();
     }
   }
 
   void Reset() {
-    Cleanable::Reset();
-    pinned_ = false;
-    size_ = 0;
+    this->~PinnableSlice();
+    new(this)PinnableSlice();
   }
 
-  inline std::string* GetSelf() { return buf_; }
-
-  inline bool IsPinned() const { return pinned_; }
+  std::string* GetSelf() const;
+  inline bool IsPinned() const { return (handle_ & 7) != kNotPinned; }
 
  private:
+  struct CleanerString : public Cleanable {
+    using Cleanable::Cleanable;
+    std::string self_space;
+    std::string *str_ptr = &self_space;
+  };
+  enum HandleType {
+    kNotPinned,      // handle is std::string, no need free this->data_, maybe null
+    kFreeSlice,      // handle is std::string,    need free this->data_
+    kFreeHandle,     // handle is a malloc block, no std::string, need free handle
+    kCleanerString,  // this is rare case
+  };
+  size_t handle_;
   friend class PinnableSlice4Test;
-  std::string self_space_;
-  std::string* buf_;
-  bool pinned_ = false;
 };
 
 // A set of Slices that are virtually concatenated together.  'parts' points

table/cleanable_test.cc

@@ -209,6 +209,7 @@ static void ReleaseStringHeap(void* s, void*) {
 class PinnableSlice4Test : public PinnableSlice {
  public:
   void TestStringIsRegistered(std::string* s) {
+    auto& cleanup_ = Cleaner()->GetCleanup();
     ASSERT_TRUE(cleanup_.function == ReleaseStringHeap);
     ASSERT_EQ(cleanup_.arg1, s);
     ASSERT_EQ(cleanup_.arg2, nullptr);

util/slice.cc

@@ -17,6 +17,7 @@
 #include "rocksdb/slice_transform.h"
 #include "rocksdb/utilities/object_registry.h"
 #include "rocksdb/utilities/options_type.h"
+#include "port/likely.h"
 #include "util/string_util.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -372,32 +373,167 @@ bool Slice::DecodeHex(std::string* result) const {
 }
 
 PinnableSlice::PinnableSlice(PinnableSlice&& other) {
-  *this = std::move(other);
+  data_ = other.data_;
+  size_ = other.size_;
+  handle_ = other.handle_;
+  new(&other)PinnableSlice();
 }
 
-PinnableSlice& PinnableSlice::operator=(PinnableSlice&& other) {
-  if (this != &other) {
-    Cleanable::Reset();
-    Cleanable::operator=(std::move(other));
-    size_ = other.size_;
-    pinned_ = other.pinned_;
-    if (pinned_) {
-      data_ = other.data_;
-      // When it's pinned, buf should no longer be of use.
+void PinnableSlice::PinSlice(const Slice& s, CleanupFunction f, void* arg1,
+                             void* arg2) {
+  assert(!IsPinned());
+  data_ = s.data();
+  size_ = s.size();
+  if (CleanupFunction(::free) == f) {
+    if (arg1 == s.data_) {
+      handle_ = (handle_ & ~3) | kFreeSlice;
+    } else if ((handle_ & ~7) == 0) {
+      handle_ = size_t(arg1) | kFreeHandle;
     } else {
-      if (other.buf_ == &other.self_space_) {
-        self_space_ = std::move(other.self_space_);
-        buf_ = &self_space_;
-        data_ = buf_->data();
+      auto cs = new CleanerString(f, arg1, arg2);
+      cs->str_ptr = (std::string*)(handle_ & ~7);
+      handle_ = size_t(cs) | kCleanerString;
+    }
+  }
+  else {
+    auto cs = new CleanerString(f, arg1, arg2);
+    if (auto str_ptr = (std::string*)(handle_ & ~7)) {
+      cs->str_ptr = str_ptr;
+    }
+    handle_ = size_t(cs) | kCleanerString;
+  }
+  assert(IsPinned());
+}
+
+void PinnableSlice::PinSlice(const Slice& s, Cleanable* cleanable) {
+  assert(!IsPinned());
+  data_ = s.data();
+  size_ = s.size();
+  if (cleanable != nullptr) {
+    auto& c = cleanable->GetCleanup();
+    if (CleanupFunction(::free) == c.function && !c.next) {
+      if (c.arg1 == s.data_) {
+        handle_ = (handle_ & ~3) | kFreeSlice;
+      } else if ((handle_ & ~7) == 0) {
+        handle_ = size_t(c.arg1) | (handle_ & 4) | kFreeHandle;
       } else {
-        buf_ = other.buf_;
-        data_ = other.data_;
+        auto cs = new CleanerString(c.function, c.arg1, c.arg2);
+        cs->str_ptr = (std::string*)(handle_ & ~7);
+        handle_ = size_t(cs) | kCleanerString;
+      }
+    }
+    else {
+      auto cs = new CleanerString;
+      if (auto str_ptr = (std::string*)(handle_ & ~7)) {
+        cs->str_ptr = str_ptr;
+      }
+      cleanable->DelegateCleanupsTo(cs);
+      handle_ = size_t(cs) | kCleanerString;
+    }
+  }
+  assert(IsPinned());
+}
+
+void PinnableSlice::PinSelf(const Slice& slice) {
+  assert(!IsPinned());
+  auto str_ptr = (std::string*)(handle_ & ~7);
+  if (!str_ptr) {
+    str_ptr = new std::string(slice.data_, slice.size_);
+    handle_ = size_t(str_ptr) | 4 | (handle_ & 3);
+  } else {
+    str_ptr->assign(slice.data(), slice.size());
+  }
+  data_ = str_ptr->data();
+  size_ = str_ptr->size();
+  assert(!IsPinned());
+}
+
+std::string* PinnableSlice::GetSelf() const {
+  void* ptr = (void*)(handle_ & ~7);
+  const auto ht = HandleType(handle_ & 3);
+  if (ht < kCleanerString) {
+    if (!ptr) {
+      ptr = new std::string;
+      const_cast<PinnableSlice*>(this)->handle_ = size_t(ptr) | 4 | ht;
+    }
+    return (std::string*)(ptr);
+  }
+  else {
+    assert(ptr != nullptr);
+    return ((CleanerString*)(ptr))->str_ptr;
+  }
+}
+
+Cleanable* PinnableSlice::Cleaner() {
+  const auto ht = HandleType(handle_ & 3);
+  if (ht < kCleanerString) {
+    auto cs = new CleanerString;
+    if (ht < kFreeHandle) {
+      if (auto str_ptr = (std::string*)(handle_ & ~7)) {
+        if (handle_ & 4) { // owner
+          cs->self_space = std::move(*str_ptr);
+          delete str_ptr;
+        } else {
+          cs->str_ptr = str_ptr;
+        }
+      }
+      if (ht == kFreeSlice) {
+        cs->RegisterCleanup(CleanupFunction(::free), (void*)(data_), nullptr);
       }
     }
-    other.self_space_.clear();
-    other.buf_ = &other.self_space_;
-    other.pinned_ = false;
-    other.PinSelf();
+    else if (ht == kFreeHandle) {
+      cs->RegisterCleanup(CleanupFunction(::free), (void*)(handle_ & ~7), nullptr);
+    }
+    handle_ = size_t(cs) | kCleanerString;
+  }
+  return (Cleanable*)(handle_ & ~7);
+}
+
+void PinnableSlice::PinSelf() {
+  assert(!IsPinned());
+  auto str_ptr = (std::string*)(handle_ & ~7);
+  if (!str_ptr) {
+    data_ = "";
+    size_ = 0;
+  } else {
+    data_ = str_ptr->data();
+    size_ = str_ptr->size();
+  }
+  assert(!IsPinned());
+}
+
+void PinnableSlice::SyncToString() const {
+  std::string* str_ptr = GetSelf();
+  assert(str_ptr != nullptr);
+  SyncToString(str_ptr);
+}
+
+PinnableSlice::~PinnableSlice() {
+  const auto ht = HandleType(handle_ & 3);
+  if (ht < kCleanerString) {
+    if (ht == kFreeSlice) {
+      ::free((void*)data_);
+    }
+    else if (ht == kFreeHandle) {
+      assert((handle_ & 4) == 0);
+      ::free((void*)(handle_ & ~7));
+    }
+    if (handle_ & 4) {
+      auto str_ptr = (std::string*)(handle_ & ~7);
+      delete str_ptr;
+    }
+  }
+  else {
+    auto cs = (CleanerString*)(handle_ & ~7);
+    assert(cs != nullptr);
+    delete cs;
+  }
+}
+
+PinnableSlice& PinnableSlice::operator=(PinnableSlice&& other) {
+  if (this != &other) {
+    this->~PinnableSlice();
+    new(this)PinnableSlice(std::move(other));
   }
   return *this;
 }