buffer, test: bl::append_zero2() deduplicates zeros, introduce bptr::is_zero_fast()

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>
Signed-off-by: Alex Ainscow <aainscow@uk.ibm.com>
(cherry picked from commit efcac634185dd82b31755a06c7fcc26d9baa1e2c)

diff --git a/src/common/buffer.cc b/src/common/buffer.cc
index 8caacd36319024f0220a23c2f3a496173f8fb9e5..ee502e25b204d9de86fddd9efd9ad9462df559e0 100644 (file)
--- a/src/common/buffer.cc
+++ b/src/common/buffer.cc
@@ -96,10 +96,9 @@ static ceph::spinlock debug_lock;
       : raw(dataptr, l, mempool) {
     }
 
-    static ceph::unique_leakable_ptr<buffer::raw>
-    create(unsigned len,
-          unsigned align,
-          int mempool = mempool::mempool_buffer_anon)
+    static std::pair<char*, size_t> alloc_data_n_controlblock(
+      unsigned len,
+      unsigned align)
     {
       // posix_memalign() requires a multiple of sizeof(void *)
       align = std::max<unsigned>(align, sizeof(void *));
@@ -117,7 +116,15 @@ static ceph::spinlock debug_lock;
 #endif /* DARWIN */
       if (!ptr)
        throw bad_alloc();
+      return {ptr, datalen};
+    }
 
+    static ceph::unique_leakable_ptr<buffer::raw>
+    create(unsigned len,
+          unsigned align,
+          int mempool = mempool::mempool_buffer_anon)
+    {
+      const auto [ptr, datalen] = alloc_data_n_controlblock(len, align);
       // actual data first, since it has presumably larger alignment restriction
       // then put the raw_combined at the end
       return ceph::unique_leakable_ptr<buffer::raw>(
@@ -130,6 +137,26 @@ static ceph::spinlock debug_lock;
     }
   };
 
+  class buffer::raw_zeros : public buffer::raw_combined {
+    raw_zeros(char *dataptr, unsigned l, int mempool)
+      : raw_combined(dataptr, l, mempool) {
+      memset(dataptr, 0, l);
+    }
+
+    static constexpr unsigned ZERO_AREA_NUM_PAGES = 4;
+
+  public:
+    static ceph::unique_leakable_ptr<buffer::raw>
+    create(int mempool = mempool::mempool_buffer_anon)
+    {
+      const auto ZERO_AREA_SIZE = ZERO_AREA_NUM_PAGES * CEPH_PAGE_SIZE;
+      const auto [ptr, datalen] = alloc_data_n_controlblock(
+        ZERO_AREA_SIZE, /* align to */CEPH_PAGE_SIZE);
+      return ceph::unique_leakable_ptr<buffer::raw>(
+       new (ptr + datalen) raw_zeros(ptr, ZERO_AREA_SIZE, mempool));
+    }
+  };
+
   class buffer::raw_malloc : public buffer::raw {
   public:
     MEMPOOL_CLASS_HELPERS();
@@ -521,9 +548,14 @@ static ceph::spinlock debug_lock;
     return 0;
   }
 
+  bool buffer::ptr::is_zero_fast() const
+  {
+    return dynamic_cast<const buffer::raw_zeros*>(_raw) != nullptr;
+  }
+
   bool buffer::ptr::is_zero() const
   {
-    return mem_is_zero(c_str(), _len);
+    return /*is_zero_fast() ||*/ mem_is_zero(c_str(), _len);
   }
 
   unsigned buffer::ptr::append(char c)
@@ -1292,6 +1324,14 @@ static ceph::spinlock debug_lock;
     return _buffers.back();
   }
 
+  buffer::ptr buffer::list::always_zeroed_bptr() {
+    // See https://en.cppreference.com/w/cpp/language/storage_duration.html
+    // Section on static block variables states that since C++11 this is
+    // lazily evaluated and thread safe.
+    static ptr always_zeroed_bptr = raw_zeros::create();
+    return always_zeroed_bptr;
+  }
+
   void buffer::list::append(const char *data, unsigned len)
   {
     _len += len;
@@ -1427,6 +1467,18 @@ static ceph::spinlock debug_lock;
     _buffers.push_front(*bp.release());
   }
   
+  void buffer::list::append_zero2(unsigned len)
+  {
+    _len += len;
+    while (len > 0) {
+      const auto round_size = std::min(len, always_zeroed_bptr().length());
+      auto bptr = ptr_node::create(always_zeroed_bptr(), 0, round_size);
+      _buffers.push_back(*bptr.release());
+      _num += 1;
+      len -= round_size;
+    }
+  }
+
   void buffer::list::append_zero(unsigned len)
   {
     _len += len;

diff --git a/src/include/buffer.h b/src/include/buffer.h
index 07ae34f6a15baa336eccc3674d70b2e43e1959ea..85e40d1cf2cb32e6f9098bdaa903507c5732c2be 100644 (file)
--- a/src/include/buffer.h
+++ b/src/include/buffer.h
@@ -129,6 +129,7 @@ struct error_code;
   class raw_claimed_char;
   class raw_unshareable; // diagnostic, unshareable char buffer
   class raw_combined;
+  class raw_zeros;
   class raw_claim_buffer;
 
 
@@ -303,6 +304,9 @@ struct error_code;
     unsigned wasted() const;
 
     int cmp(const ptr& o) const;
+    /// is_zero_fast() is a variant aware about deduplicated zeros.
+    /// In Tentacle it shall NOT be used by anybody except ECBackend.
+    bool is_zero_fast() const;
     bool is_zero() const;
 
     // modifiers
@@ -930,6 +934,8 @@ struct error_code;
       return *_carriage;
     }
 
+    ptr always_zeroed_bptr();
+
   public:
     // cons/des
     list()
@@ -1163,6 +1169,11 @@ struct error_code;
     void append(std::istream& in);
     contiguous_filler append_hole(unsigned len);
     void append_zero(unsigned len);
+    /// append_zero2() is a temporary, short-living variant that deduplicates zeros.
+    /// In Tentacle it shall NOT be used by anybody except ECBackend.
+    /// In future release it will likely replace the append_zero() variant but
+    /// other changes at the interface are needed to make the transition safe.
+    void append_zero2(unsigned len);
     void prepend_zero(unsigned len);
 
     reserve_t obtain_contiguous_space(const unsigned len);

diff --git a/src/test/bufferlist.cc b/src/test/bufferlist.cc
index a91ed09883a7cb93a9201e319d691238d71935b2..a083ad75f095c9b83289e70d8f1ae915d05c1a44 100644 (file)
--- a/src/test/bufferlist.cc
+++ b/src/test/bufferlist.cc
@@ -481,15 +481,44 @@ TEST(BufferPtr, cmp) {
   EXPECT_LE(1, af.cmp(acc));
 }
 
+TEST(BufferPtr, is_zero_fast) {
+  // there is no easy way to create `raw_zeros` instances outside
+  // of bufferlist, thus use list::append_zero() to instantiate
+  buffer::list zeroed_bl;
+  zeroed_bl.append_zero2(42);
+  {
+    const auto& zeroed_bptr = zeroed_bl.front();
+    EXPECT_TRUE(zeroed_bptr.is_zero());
+    EXPECT_TRUE(zeroed_bptr.is_zero_fast());
+  }
+  {
+    buffer::ptr sub_zeroed_bptr(zeroed_bl.front(), 0, 42/2);
+    EXPECT_TRUE(sub_zeroed_bptr.is_zero());
+    EXPECT_TRUE(sub_zeroed_bptr.is_zero_fast());
+  }
+  {
+    buffer::ptr zeroed_empty_bptr(zeroed_bl.front(), 0, 0);
+    EXPECT_TRUE(zeroed_empty_bptr.is_zero());
+    EXPECT_TRUE(zeroed_empty_bptr.is_zero_fast());
+  }
+}
+
 TEST(BufferPtr, is_zero) {
   char str[2] = { '\0', 'X' };
   {
     const bufferptr ptr(buffer::create_static(2, str));
     EXPECT_FALSE(ptr.is_zero());
+    EXPECT_FALSE(ptr.is_zero_fast());
   }
   {
     const bufferptr ptr(buffer::create_static(1, str));
     EXPECT_TRUE(ptr.is_zero());
+    EXPECT_FALSE(ptr.is_zero_fast());
+  }
+  {
+    const bufferptr ptr(buffer::create_static(0, str));
+    EXPECT_TRUE(ptr.is_zero());
+    EXPECT_FALSE(ptr.is_zero_fast());
   }
 }