From 9a41b7b865c9845a7f1d434eadcb5bcdad033963 Mon Sep 17 00:00:00 2001
From: Radoslaw Zarzynski <rzarzyns@redhat.com>
Date: Tue, 9 Oct 2018 01:17:31 +0200
Subject: [PATCH] common: hypercombined bufferlist.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>
---
 src/common/buffer.cc     | 29 +++++++++++++++++++++++++++--
 src/include/buffer.h     | 27 +++++++++++++++++++++++----
 src/include/buffer_raw.h |  6 +++++-
 3 files changed, 55 insertions(+), 7 deletions(-)

diff --git a/src/common/buffer.cc b/src/common/buffer.cc
index 4634f14f7a502..ab7dc0d3d09b3 100644
--- a/src/common/buffer.cc
+++ b/src/common/buffer.cc
@@ -639,14 +639,18 @@ using namespace ceph;
     if (_raw) {
       bdout << "ptr " << this << " release " << _raw << bendl;
       if (--_raw->nref == 0) {
+        // BE CAREFUL: this is called also for hypercombined ptr_node. After
+        // freeing underlying raw, `*this` can become inaccessible as well!
+        const auto* delete_raw = _raw;
+        _raw = nullptr;
 	//cout << "hosing raw " << (void*)_raw << " len " << _raw->len << std::endl;
         ANNOTATE_HAPPENS_AFTER(&_raw->nref);
         ANNOTATE_HAPPENS_BEFORE_FORGET_ALL(&_raw->nref);
-	delete _raw;  // dealloc old (if any)
+	delete delete_raw;  // dealloc old (if any)
       } else {
         ANNOTATE_HAPPENS_BEFORE(&_raw->nref);
+        _raw = nullptr;
       }
-      _raw = 0;
     }
   }
 
@@ -2183,6 +2187,27 @@ buffer::list buffer::list::static_from_string(string& s) {
   // const makes me generally sad.
 }
 
+bool buffer::ptr_node::dispose_if_hypercombined(
+  buffer::ptr_node* const delete_this)
+{
+  const bool is_hypercombined = static_cast<void*>(delete_this) == \
+    static_cast<void*>(&delete_this->get_raw()->bptr_storage);
+  if (is_hypercombined) {
+    delete_this->~ptr_node();
+  }
+  return is_hypercombined;
+}
+
+buffer::ptr_node& buffer::ptr_node::create_hypercombined(
+  buffer::raw* const r)
+{
+  if (likely(r->nref == 0)) {
+    return *new (&r->bptr_storage) ptr_node(r);
+  } else {
+    return *new ptr_node(r);
+  }
+}
+
 std::ostream& buffer::operator<<(std::ostream& out, const buffer::raw &r) {
   return out << "buffer::raw(" << (void*)r.data << " len " << r.len << " nref " << r.nref.load() << ")";
 }
diff --git a/src/include/buffer.h b/src/include/buffer.h
index b592255667494..27eba3ccd0474 100644
--- a/src/include/buffer.h
+++ b/src/include/buffer.h
@@ -273,6 +273,8 @@ namespace buffer CEPH_BUFFER_API {
     ptr& operator= (const ptr& p);
     ptr& operator= (ptr&& p) noexcept;
     ~ptr() {
+      // BE CAREFUL: this destructor is called also for hypercombined ptr_node.
+      // After freeing underlying raw, `*this` can become inaccessible as well!
       release();
     }
 
@@ -385,12 +387,27 @@ namespace buffer CEPH_BUFFER_API {
     template <class... Args>
     ptr_node(Args&&... args) : ptr(std::forward<Args>(args)...) {
     }
-
     ptr_node(const ptr_node&) = default;
-  public:
 
+    ptr& operator= (const ptr& p) = delete;
+    ptr& operator= (ptr&& p) noexcept = delete;
+    ptr_node& operator= (const ptr_node& p) = delete;
+    ptr_node& operator= (ptr_node&& p) noexcept = delete;
+    void swap(ptr& other) noexcept = delete;
+    void swap(ptr_node& other) noexcept = delete;
+
+  public:
     ~ptr_node() = default;
 
+    static bool dispose_if_hypercombined(ptr_node* delete_this);
+    static ptr_node& create_hypercombined(raw* r);
+
+    static ptr_node& create(raw* const r) {
+      return create_hypercombined(r);
+    }
+    static ptr_node& create(const unsigned l) {
+      return create_hypercombined(buffer::create(l));
+    }
     template <class... Args>
     static ptr_node& create(Args&&... args) {
       return *new ptr_node(std::forward<Args>(args)...);
@@ -403,7 +420,9 @@ namespace buffer CEPH_BUFFER_API {
     };
     struct disposer {
       void operator()(ptr_node* const delete_this) {
-	delete delete_this;
+	if (!dispose_if_hypercombined(delete_this)) {
+	  delete delete_this;
+	}
       }
     };
   };
@@ -834,7 +853,7 @@ namespace buffer CEPH_BUFFER_API {
       _len += bp.length();
       _buffers.push_back(bp);
     }
-    void push_back(raw *r) {
+    void push_back(raw* const r) {
       _buffers.push_back(ptr_node::create(r));
       _len += _buffers.back().length();
     }
diff --git a/src/include/buffer_raw.h b/src/include/buffer_raw.h
index 2303fc5631e15..301e2ca99c1aa 100644
--- a/src/include/buffer_raw.h
+++ b/src/include/buffer_raw.h
@@ -18,6 +18,7 @@
 #include <atomic>
 #include <map>
 #include <utility>
+#include <type_traits>
 #include "include/buffer.h"
 #include "include/mempool.h"
 #include "include/spinlock.h"
@@ -25,6 +26,9 @@
 namespace ceph::buffer {
   class raw {
   public:
+    // In the future we might want to have a slab allocator here with few
+    // embedded slots. This would allow to avoid the "if" in dtor of ptr_node.
+    std::aligned_storage_t<sizeof(ptr_node), alignof(ptr_node)> bptr_storage;
     char *data;
     unsigned len;
     std::atomic<unsigned> nref { 0 };
@@ -36,7 +40,7 @@ namespace ceph::buffer {
     mutable ceph::spinlock crc_spinlock;
 
     explicit raw(unsigned l, int mempool=mempool::mempool_buffer_anon)
-      : data(NULL), len(l), nref(0), mempool(mempool) {
+      : data(nullptr), len(l), nref(0), mempool(mempool) {
       mempool::get_pool(mempool::pool_index_t(mempool)).adjust_count(1, len);
     }
     raw(char *c, unsigned l, int mempool=mempool::mempool_buffer_anon)
-- 
2.39.5