From 65d72a4e48d379dc1f8ae1b545a5ea20f160ecd4 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kchai@redhat.com>
Date: Tue, 16 Jan 2018 13:54:39 +0800
Subject: [PATCH] denc: genenalize denc interface for seastar::temporary_buffer

Signed-off-by: Kefu Chai <kchai@redhat.com>
---
 src/common/buffer_seastar.cc | 35 +++++++++++++--
 src/common/buffer_seastar.h  | 61 +++++++++++++++++++++++++
 src/include/denc.h           | 87 ++++++++++++++++++++++--------------
 3 files changed, 146 insertions(+), 37 deletions(-)
 create mode 100644 src/common/buffer_seastar.h

diff --git a/src/common/buffer_seastar.cc b/src/common/buffer_seastar.cc
index 2c6052ab12b..a2bc91df9d6 100644
--- a/src/common/buffer_seastar.cc
+++ b/src/common/buffer_seastar.cc
@@ -15,11 +15,11 @@
 #include <core/sharded.hh>
 
 #include "include/buffer_raw.h"
+#include "buffer_seastar.h"
 
 using temporary_buffer = seastar::temporary_buffer<char>;
 
-namespace ceph {
-namespace buffer {
+namespace ceph::buffer {
 
 class raw_seastar_foreign_ptr : public raw {
   seastar::foreign_ptr<temporary_buffer> ptr;
@@ -77,5 +77,32 @@ list::operator seastar::net::packet() &&
   return p;
 }
 
-} // namespace buffer
-} // namespace ceph
+} // namespace ceph::buffer
+
+namespace {
+
+using ceph::buffer::raw;
+class raw_seastar_local_shared_ptr : public raw {
+  temporary_buffer buf;
+public:
+  raw_seastar_local_shared_ptr(temporary_buffer& buf)
+    : raw(buf.get_write(), buf.size()), buf(buf.share()) {}
+  raw* clone_empty() override {
+    return ceph::buffer::create(len);
+  }
+};
+}
+
+buffer::ptr seastar_buffer_iterator::get_ptr(size_t len)
+{
+  buffer::raw* r = new raw_seastar_local_shared_ptr{buf};
+  buffer::ptr p{r};
+  p.set_length(len);
+  return p;
+}
+
+buffer::ptr const_seastar_buffer_iterator::get_ptr(size_t len)
+{
+  buffer::raw* r = buffer::copy(get_pos_add(len), len);
+  return buffer::ptr{r};
+}
diff --git a/src/common/buffer_seastar.h b/src/common/buffer_seastar.h
new file mode 100644
index 00000000000..5316b7d5980
--- /dev/null
+++ b/src/common/buffer_seastar.h
@@ -0,0 +1,61 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include <core/temporary_buffer.hh>
+#include "include/buffer.h"
+
+namespace details {
+
+template<bool is_const>
+class buffer_iterator_impl {
+public:
+  using pointer = std::conditional_t<is_const, const char*, char *>;
+  buffer_iterator_impl(pointer first, const char* last)
+    : pos(first), end_ptr(last)
+  {}
+  pointer get_pos_add(size_t n) {
+    auto r = pos;
+    pos += n;
+    if (pos > end_ptr) {
+      throw buffer::end_of_buffer{};
+    }
+    return r;
+  }
+  pointer get() const {
+    return pos;
+  }
+protected:
+  pointer pos;
+  const char* end_ptr;
+};
+} // namespace details
+
+class seastar_buffer_iterator : details::buffer_iterator_impl<false> {
+  using parent = details::buffer_iterator_impl<false>;
+  using temporary_buffer = seastar::temporary_buffer<char>;
+public:
+  seastar_buffer_iterator(temporary_buffer& b)
+    : parent(b.get_write(), b.end()), buf(b)
+  {}
+  using parent::pointer;
+  using parent::get_pos_add;
+  using parent::get;
+  ceph::buffer::ptr get_ptr(size_t len);
+
+private:
+  // keep the reference to buf around, so it can be "shared" by get_ptr()
+  temporary_buffer& buf;
+};
+
+class const_seastar_buffer_iterator : details::buffer_iterator_impl<true> {
+  using parent = details::buffer_iterator_impl<true>;
+  using temporary_buffer = seastar::temporary_buffer<char>;
+public:
+  const_seastar_buffer_iterator(temporary_buffer& b)
+    : parent(b.get_write(), b.end())
+  {}
+  using parent::pointer;
+  using parent::get_pos_add;
+  using parent::get;
+  ceph::buffer::ptr get_ptr(size_t len);
+};
diff --git a/src/include/denc.h b/src/include/denc.h
index f80f1a0c4c4..bbb830959ce 100644
--- a/src/include/denc.h
+++ b/src/include/denc.h
@@ -286,13 +286,14 @@ struct denc_traits<
   static void bound_encode(const T &o, size_t& p, uint64_t f=0) {
     p += sizeof(T);
   }
-  static void encode(const T &o,
-		     buffer::list::contiguous_appender& p,
-		     uint64_t f=0) {
+  template<class It>
+  static std::enable_if_t<!is_const_iterator_v<It>>
+  encode(const T &o, It& p, uint64_t f=0) {
     get_pos_add<T>(p) = o;
   }
-  static void decode(T& o, buffer::ptr::const_iterator& p,
-		     uint64_t f=0) {
+  template<class It>
+  static std::enable_if_t<is_const_iterator_v<It>>
+  decode(T& o, It& p, uint64_t f=0) {
     o = get_pos_add<T>(p);
   }
   static void decode(T& o, buffer::list::const_iterator &p) {
@@ -355,11 +356,14 @@ struct denc_traits<T, std::enable_if_t<!std::is_void_v<_denc::ExtType_t<T>>>>
   static void bound_encode(const T &o, size_t& p, uint64_t f=0) {
     p += sizeof(etype);
   }
-  static void encode(const T &o, buffer::list::contiguous_appender& p,
-		     uint64_t f=0) {
+  template<class It>
+  static std::enable_if_t<!is_const_iterator_v<It>>
+  encode(const T &o, It& p, uint64_t f=0) {
     get_pos_add<etype>(p) = o;
   }
-  static void decode(T& o, bufferptr::const_iterator &p, uint64_t f=0) {
+  template<class It>
+  static std::enable_if_t<is_const_iterator_v<It>>
+  decode(T& o, It &p, uint64_t f=0) {
     o = get_pos_add<etype>(p);
   }
   static void decode(T& o, buffer::list::const_iterator &p) {
@@ -410,7 +414,9 @@ inline void denc_varint(T& v, bufferptr::const_iterator& p) {
 inline void denc_signed_varint(int64_t v, size_t& p) {
   p += sizeof(v) + 2;
 }
-inline void denc_signed_varint(int64_t v, bufferlist::contiguous_appender& p) {
+template<class It>
+inline std::enable_if_t<!is_const_iterator_v<It>>
+denc_signed_varint(int64_t v, It& p) {
   if (v < 0) {
     v = (-v << 1) | 1;
   } else {
@@ -419,8 +425,9 @@ inline void denc_signed_varint(int64_t v, bufferlist::contiguous_appender& p) {
   denc_varint(v, p);
 }
 
-template<typename T>
-inline void denc_signed_varint(T& v, bufferptr::const_iterator& p)
+template<typename T, class It>
+inline std::enable_if_t<is_const_iterator_v<It>>
+denc_signed_varint(T& v, It& p)
 {
   int64_t i = 0;
   denc_varint(i, p);
@@ -469,8 +476,9 @@ inline void denc_varint_lowz(T& v, bufferptr::const_iterator& p)
 inline void denc_signed_varint_lowz(int64_t v, size_t& p) {
   p += sizeof(v) + 2;
 }
-inline void denc_signed_varint_lowz(int64_t v,
-				    bufferlist::contiguous_appender& p) {
+template<class It>
+inline std::enable_if_t<!is_const_iterator_v<It>>
+denc_signed_varint_lowz(int64_t v, It& p) {
   bool negative = false;
   if (v < 0) {
     v = -v;
@@ -486,8 +494,9 @@ inline void denc_signed_varint_lowz(int64_t v,
   denc_varint(v, p);
 }
 
-template<typename T>
-inline void denc_signed_varint_lowz(T& v, bufferptr::const_iterator& p)
+template<typename T, class It>
+inline std::enable_if_t<is_const_iterator_v<It>>
+denc_signed_varint_lowz(T& v, It& p)
 {
   int64_t i = 0;
   denc_varint(i, p);
@@ -518,7 +527,9 @@ inline void denc_lba(uint64_t v, size_t& p) {
   p += sizeof(v) + 2;
 }
 
-inline void denc_lba(uint64_t v, bufferlist::contiguous_appender& p) {
+template<class It>
+inline std::enable_if_t<!is_const_iterator_v<It>>
+denc_lba(uint64_t v, It& p) {
   int low_zero_nibbles = v ? (int)(ctz(v) / 4) : 0;
   int pos;
   uint32_t word;
@@ -554,7 +565,9 @@ inline void denc_lba(uint64_t v, bufferlist::contiguous_appender& p) {
   *(__u8*)p.get_pos_add(1) = byte;
 }
 
-inline void denc_lba(uint64_t& v, bufferptr::const_iterator& p) {
+template<class It>
+inline std::enable_if_t<is_const_iterator_v<It>>
+denc_lba(uint64_t& v, It& p) {
   uint32_t word = *(__le32*)p.get_pos_add(sizeof(uint32_t));
   int shift;
   switch (word & 7) {
@@ -603,10 +616,10 @@ inline std::enable_if_t<traits::supported> denc(
   }
 }
 
-template<typename T, typename traits=denc_traits<T>>
-inline std::enable_if_t<traits::supported>
+template<typename T, class It, typename traits=denc_traits<T>>
+inline std::enable_if_t<traits::supported && !is_const_iterator_v<It>>
 denc(const T& o,
-     buffer::list::contiguous_appender& p,
+     It& p,
      uint64_t features=0)
 {
   if constexpr (traits::featured) {
@@ -616,10 +629,10 @@ denc(const T& o,
   }
 }
 
-template<typename T, typename traits=denc_traits<T>>
-inline std::enable_if_t<traits::supported>
+template<typename T, class It, typename traits=denc_traits<T>>
+inline std::enable_if_t<traits::supported && is_const_iterator_v<It>>
 denc(T& o,
-     buffer::ptr::const_iterator& p,
+     It& p,
      uint64_t features=0)
 {
   if constexpr (traits::featured) {
@@ -682,14 +695,16 @@ public:
   static void bound_encode(const value_type& s, size_t& p, uint64_t f=0) {
     p += sizeof(uint32_t) + s.size();
   }
+  template<class It>
   static void encode(const value_type& s,
-		     buffer::list::contiguous_appender& p,
+		     It& p,
                      uint64_t f=0) {
     denc((uint32_t)s.size(), p);
     memcpy(p.get_pos_add(s.size()), s.data(), s.size());
   }
+  template<class It>
   static void decode(value_type& s,
-		     buffer::ptr::const_iterator& p,
+		     It& p,
 		     uint64_t f=0) {
     uint32_t len;
     denc(len, p);
@@ -702,16 +717,19 @@ public:
     s.clear();
     p.copy(len, s);
   }
-  static void decode_nohead(size_t len, value_type& s,
-                            buffer::ptr::const_iterator& p) {
+  template<class It>
+  static std::enable_if_t<is_const_iterator_v<It>>
+  decode_nohead(size_t len, value_type& s, It& p) {
     s.clear();
     if (len) {
       s.append(p.get_pos_add(len), len);
     }
   }
-  static void encode_nohead(const value_type& s,
-			    buffer::list::contiguous_appender& p) {
-    p.append(s.data(), s.length());
+  template<class It>
+  static std::enable_if_t<!is_const_iterator_v<It>>
+  encode_nohead(const value_type& s, It& p) {
+    auto len = s.length();
+    maybe_inline_memcpy(p.get_pos_add(len), s.data(), len, 16);
   }
 };
 
@@ -727,12 +745,15 @@ struct denc_traits<bufferptr> {
   static void bound_encode(const bufferptr& v, size_t& p, uint64_t f=0) {
     p += sizeof(uint32_t) + v.length();
   }
-  static void encode(const bufferptr& v, buffer::list::contiguous_appender& p,
-	      uint64_t f=0) {
+  template <class It>
+  static std::enable_if_t<!is_const_iterator_v<It>>
+  encode(const bufferptr& v, It& p, uint64_t f=0) {
     denc((uint32_t)v.length(), p);
     p.append(v);
   }
-  static void decode(bufferptr& v, buffer::ptr::const_iterator& p, uint64_t f=0) {
+  template <class It>
+  static std::enable_if_t<is_const_iterator_v<It>>
+  decode(bufferptr& v, It& p, uint64_t f=0) {
     uint32_t len;
     denc(len, p);
     v = p.get_ptr(len);
-- 
2.39.5