From 77fbfc29d15e3b9b05f7434bea8792077a18aa42 Mon Sep 17 00:00:00 2001
From: Kefu Chai <kchai@redhat.com>
Date: Tue, 23 May 2017 15:28:23 +0800
Subject: [PATCH] denc: add need_contiguous to denc_traits

so the decode() will not try to copy the bufferlist for a continous one
if the bufferlist is segmented *and* its length is greater than
CEPH_PAGE_SIZE (4K) if the decoded object does not "need_contiguous" per
its denc_traits<>.

copying a memory chunk could be expensive if the decoded bufferlist is
huge, so we should try to avoid this. this could happen when we read the
buffer from bluestore.

and drop the partial specialization for denc() which tries to differentiate
traits::featured and !traits::featured, it does not matter to
decode() if the type supports feature or not. the encode() does.

also use denc() in interval_set<>::decode(). unlike encode(), decode(),
encode_nohead() or decode_nohead(), denc() is not part of our legacy
dencoder, so i think it's fine and encouraged to use it.

Signed-off-by: Kefu Chai <kchai@redhat.com>
---
 src/include/denc.h                 | 202 ++++++++++++++++++++++++++---
 src/include/fs_types.h             |   1 +
 src/include/interval_set.h         |  31 +++--
 src/include/object.h               |   1 +
 src/os/bluestore/bluestore_types.h |   1 +
 src/test/test_denc.cc              | 118 +++++++++++++++++
 6 files changed, 329 insertions(+), 25 deletions(-)

diff --git a/src/include/denc.h b/src/include/denc.h
index e3ff107fb1e8..4febc74a75f8 100644
--- a/src/include/denc.h
+++ b/src/include/denc.h
@@ -44,11 +44,12 @@
 #include "buffer.h"
 #include "byteorder.h"
 
-template<typename T, typename VVV=void>
+template<typename T, typename=void>
 struct denc_traits {
   static constexpr bool supported = false;
   static constexpr bool featured = false;
   static constexpr bool bounded = false;
+  static constexpr bool need_contiguous = true;
 };
 
 
@@ -138,6 +139,7 @@ struct denc_traits {
       static constexpr bool supported = true;
       static constexpr bool bounded = false;
       static constexpr bool featured = false;
+      static constexpr bool need_contiguous = true;
       static void bound_encode(const T &o, size_t& p, uint64_t f=0);
       static void encode(const T &o, buffer::list::contiguous_appender& p,
 		         uint64_t f=0);
@@ -151,6 +153,7 @@ struct denc_traits {
       static constexpr bool supported = true;
       static constexpr bool bounded = false;
       static constexpr bool featured = true;
+      static constexpr bool need_contiguous = true;
       static void bound_encode(const T &o, size_t& p, uint64_t f);
       static void encode(const T &o, buffer::list::contiguous_appender& p,
 		         uint64_t f);
@@ -190,6 +193,12 @@ struct denc_traits {
   declared via WRITE_CLASS_DENC, although you can also invoke them explicitly
   in your code.
 
+  - These methods are optimised for contiguous buffer, but denc() will try
+    rebuild a contigous one if the decoded bufferlist is segmented. If you are
+    concerned about the cost, you might want to define yet another method:
+
+    void decode(buffer::list::iterator &p);
+
   - These can be defined either explicitly (as above), or can be "magically"
   defined all in one go using the DENC macro and DENC_{START,FINISH} helpers
   (which work like the legacy {ENCODE,DECODE}_{START,FINISH} macros):
@@ -236,6 +245,7 @@ struct denc_traits<
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = true;
+  static constexpr bool need_contiguous = false;
   static void bound_encode(const T &o, size_t& p, uint64_t f=0) {
     p += sizeof(T);
   }
@@ -248,6 +258,9 @@ struct denc_traits<
 		     uint64_t f=0) {
     o = *(T *)p.get_pos_add(sizeof(o));
   }
+  static void decode(T& o, buffer::list::iterator &p) {
+    p.copy(sizeof(T), reinterpret_cast<char*>(&o));
+  }
 };
 
 
@@ -302,6 +315,7 @@ struct denc_traits<
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = true;
+  static constexpr bool need_contiguous = false;
   using etype = typename _denc::ExtType<T>::type;
   static void bound_encode(const T &o, size_t& p, uint64_t f=0) {
     p += sizeof(etype);
@@ -314,6 +328,11 @@ struct denc_traits<
 		     uint64_t f=0) {
     o = *(etype*)p.get_pos_add(sizeof(etype));
   }
+  static void decode(T& o, buffer::list::iterator &p) {
+    etype e;
+    p.copy(sizeof(etype), reinterpret_cast<char*>(&e));
+    o = e;
+  }
 };
 
 // varint
@@ -594,6 +613,40 @@ inline typename std::enable_if<traits::supported &&
   traits::decode(o, p, features);
 }
 
+namespace _denc {
+  template<typename T, typename=void>
+  struct has_legacy_denc : std::false_type
+  {};
+  template<typename T>
+  struct has_legacy_denc<T,
+    decltype(std::declval<T&>()
+	     .decode(std::declval<bufferlist::iterator&>()))> : std::true_type
+  {
+    static void decode(T& v, bufferlist::iterator& p) {
+      v.decode(p);
+    }
+  };
+  template<typename T>
+  struct has_legacy_denc<T,
+    typename std::enable_if<
+      !denc_traits<T>::need_contiguous>::type> : std::true_type
+  {
+    static void decode(T& v, bufferlist::iterator& p) {
+      denc_traits<T>::decode(v, p);
+    }
+  };
+}
+
+template<typename T,
+	 typename traits=denc_traits<T>,
+	 typename has_legacy_denc=_denc::has_legacy_denc<T>>
+inline typename std::enable_if<traits::supported &&
+			       has_legacy_denc::value>::type denc(
+  T& o,
+  buffer::list::iterator& p)
+{
+  has_legacy_denc::decode(o, p);
+}
 
 // ---------------------------------------------------------------------
 // base types and containers
@@ -610,6 +663,7 @@ public:
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = false;
+  static constexpr bool need_contiguous = false;
 
   static void bound_encode(const value_type& s, size_t& p, uint64_t f=0) {
     p += sizeof(uint32_t) + s.size();
@@ -627,6 +681,13 @@ public:
     ::denc(len, p);
     decode_nohead(len, s, p);
   }
+  static void decode(value_type& s, buffer::list::iterator& p)
+  {
+    uint32_t len;
+    ::denc(len, p);
+    s.clear();
+    p.copy(len, s);
+  }
   static void decode_nohead(size_t len, value_type& s,
 			    buffer::ptr::iterator& p) {
     s.clear();
@@ -648,6 +709,7 @@ struct denc_traits<bufferptr> {
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = false;
+  static constexpr bool need_contiguous = false;
   static void bound_encode(const bufferptr& v, size_t& p, uint64_t f=0) {
     p += sizeof(uint32_t) + v.length();
   }
@@ -661,6 +723,18 @@ struct denc_traits<bufferptr> {
     ::denc(len, p);
     v = p.get_ptr(len);
   }
+  static void decode(bufferptr& v, buffer::list::iterator& p) {
+    uint32_t len;
+    ::denc(len, p);
+    bufferlist s;
+    p.copy(len, s);
+    if (len) {
+      if (s.get_num_buffers() == 1)
+	v = s.front();
+      else
+	v = buffer::copy(s.c_str(), s.length());
+    }
+  }
 };
 
 //
@@ -671,6 +745,7 @@ struct denc_traits<bufferlist> {
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = false;
+  static constexpr bool need_contiguous = false;
   static void bound_encode(const bufferlist& v, size_t& p, uint64_t f=0) {
     p += sizeof(uint32_t) + v.length();
   }
@@ -685,6 +760,12 @@ struct denc_traits<bufferlist> {
     v.clear();
     v.push_back(p.get_ptr(len));
   }
+  static void decode(bufferlist& v, buffer::list::iterator& p) {
+    uint32_t len;
+    ::denc(len, p);
+    v.clear();
+    p.copy(len, v);
+  }
   static void encode_nohead(const bufferlist& v,
 			    buffer::list::contiguous_appender& p) {
     p.append(v);
@@ -712,6 +793,8 @@ struct denc_traits<
   static constexpr bool supported = true;
   static constexpr bool featured = a_traits::featured || b_traits::featured ;
   static constexpr bool bounded = a_traits::bounded && b_traits::bounded;
+  static constexpr bool need_contiguous = (a_traits::need_contiguous ||
+					   b_traits::need_contiguous);
 
   template<typename AA=A>
   static typename std::enable_if<sizeof(AA) &&
@@ -748,6 +831,13 @@ struct denc_traits<
     denc(v.first, p, f);
     denc(v.second, p, f);
   }
+  template<typename AA=A>
+  static typename std::enable_if<sizeof(AA) && !need_contiguous>::type
+    decode(std::pair<A,B>& v, buffer::list::iterator& p,
+	    uint64_t f = 0) {
+    denc(v.first, p);
+    denc(v.second, p);
+  }
 };
 
 namespace _denc {
@@ -763,6 +853,7 @@ namespace _denc {
     static constexpr bool supported = true;
     static constexpr bool featured = traits::featured;
     static constexpr bool bounded = false;
+    static constexpr bool need_contiguous = traits::need_contiguous;
 
     template<typename U=T>
     static typename std::enable_if<sizeof(U) &&
@@ -833,6 +924,13 @@ namespace _denc {
       denc(num, p);
       decode_nohead(num, s, p, f);
     }
+    template<typename U=T>
+    static typename std::enable_if<sizeof(U) && !need_contiguous>::type
+    decode(container& s, buffer::list::iterator& p) {
+      uint32_t num;
+      denc(num, p);
+      decode_nohead(num, s, p);
+    }
 
     // nohead
     template<typename U=T>
@@ -862,6 +960,18 @@ namespace _denc {
 	Details::insert(s, std::move(t));
       }
     }
+    template<typename U=T>
+    static typename std::enable_if<sizeof(U) && !need_contiguous>::type
+    decode_nohead(size_t num, container& s,
+		  buffer::list::iterator& p) {
+      s.clear();
+      Details::reserve(s, num);
+      while (num--) {
+	T t;
+	denc(t, p);
+	Details::insert(s, std::move(t));
+      }
+    }
   };
 
   template<typename T>
@@ -998,6 +1108,7 @@ public:
   static constexpr bool supported = true;
   static constexpr bool featured = traits::featured;
   static constexpr bool bounded = traits::bounded;
+  static constexpr bool need_contiguous = traits::need_contiguous;
 
   template<typename U=T>
   static typename std::enable_if<sizeof(U) &&
@@ -1056,6 +1167,14 @@ public:
     for (auto& e : s)
       denc(e, p, f);
   }
+  template<typename U=T>
+  static typename std::enable_if<sizeof(U) &&
+                                 !need_contiguous>::type
+  decode(container& s, buffer::list::iterator& p) {
+    for (auto& e : s) {
+      denc(e, p);
+    }
+  }
 };
 
 namespace _denc {
@@ -1099,12 +1218,16 @@ namespace _denc {
 				     tuple_traits<Ts...>::bounded);
     static constexpr bool featured = (denc_traits<T>::featured ||
 				    tuple_traits<Ts...>::featured);
+    static constexpr bool need_contiguous =
+      (denc_traits<T>::need_contiguous ||
+       tuple_traits<Ts...>::need_contiguous);
   };
   template<>
   struct tuple_traits<> {
     static constexpr bool supported = true;
     static constexpr bool bounded = true;
     static constexpr bool featured = false;
+    static constexpr bool need_contiguous = false;
   };
 }
 
@@ -1200,6 +1323,17 @@ private:
 		     _denc::indices<I>) {
     denc(std::get<I>(s), p);
   }
+  template<typename T, size_t I, size_t J, size_t ...Is>
+  static void decode_helper(T& s, buffer::list::iterator& p,
+			    _denc::indices<I, J, Is...>) {
+    denc(std::get<I>(s), p);
+    decode_helper(s, p, _denc::indices<J, Is...>{});
+  }
+  template<typename T, size_t I>
+  static void decode_helper(T& s, buffer::list::iterator& p,
+			    _denc::indices<I>) {
+    denc(std::get<I>(s), p);
+  }
 
 public:
   using traits = _denc::tuple_traits<Ts...>;
@@ -1207,6 +1341,7 @@ public:
   static constexpr bool supported = true;
   static constexpr bool featured = traits::featured;
   static constexpr bool bounded = traits::bounded;
+  static constexpr bool need_contiguous = traits::need_contiguous;
 
 
   template<typename U = traits>
@@ -1255,6 +1390,12 @@ public:
   static void decode(container& s, buffer::ptr::iterator& p, uint64_t f = 0) {
     decode_helper(s, p, _denc::build_indices_t<sizeof...(Ts)>{});
   }
+  template<typename U = traits>
+  static typename std::enable_if<U::supported &&
+                                 !need_contiguous>::type
+  decode(container& s, buffer::list::iterator& p, uint64_t f = 0) {
+    decode_helper(s, p, _denc::build_indices_t<sizeof...(Ts)>{});
+  }
 };
 
 //
@@ -1269,6 +1410,7 @@ struct denc_traits<
   static constexpr bool supported = true;
   static constexpr bool featured = traits::featured;
   static constexpr bool bounded = false;
+  static constexpr bool need_contiguous = traits::need_contiguous;
 
   template<typename U = T>
   static typename std::enable_if<sizeof(U) && !featured>::type
@@ -1313,6 +1455,19 @@ struct denc_traits<
     }
   }
 
+  template<typename U = T>
+  static typename std::enable_if<sizeof(U) && !need_contiguous>::type
+  decode(boost::optional<T>& v, buffer::list::iterator& p) {
+    bool x;
+    denc(x, p);
+    if (x) {
+      v = T{};
+      denc(*v, p);
+    } else {
+      v = boost::none;
+    }
+  }
+
   template<typename U = T>
   static typename std::enable_if<sizeof(U) && !featured>::type
   encode_nohead(const boost::optional<T>& v,
@@ -1345,6 +1500,7 @@ struct denc_traits<boost::none_t> {
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = true;
+  static constexpr bool need_contiguous = false;
 
   static void bound_encode(const boost::none_t& v, size_t& p) {
     p += sizeof(bool);
@@ -1369,6 +1525,7 @@ struct denc_traits<boost::none_t> {
     static constexpr bool supported = true;				\
     static constexpr bool featured = false;				\
     static constexpr bool bounded = b;					\
+    static constexpr bool need_contiguous = !_denc::has_legacy_denc<T>::value;\
     static void bound_encode(const T& v, size_t& p, uint64_t f=0) {	\
       v.bound_encode(p);						\
     }									\
@@ -1388,6 +1545,7 @@ struct denc_traits<boost::none_t> {
     static constexpr bool supported = true;				\
     static constexpr bool featured = true;				\
     static constexpr bool bounded = b;					\
+    static constexpr bool need_contiguous = !_denc::has_legacy_denc<T>::value;\
     static void bound_encode(const T& v, size_t& p, uint64_t f) {	\
       v.bound_encode(p, f);						\
     }									\
@@ -1432,34 +1590,48 @@ inline typename std::enable_if<traits::supported &&
   traits::encode(o, a, features);
 }
 
-template<typename T, typename traits=denc_traits<T>>
+template<typename T,
+	 typename traits=denc_traits<T>>
 inline typename std::enable_if<traits::supported &&
-			       !traits::featured>::type decode(
+			       !traits::need_contiguous>::type decode(
   T& o,
   bufferlist::iterator& p)
 {
   if (p.end())
     throw buffer::end_of_buffer();
-  // ensure we get a contigous buffer... until the end of the
-  // bufferlist.  we don't really know how much we'll need here,
-  // unfortunately.  hopefully it is already contiguous and we're just
-  // bumping the raw ref and initializing the ptr tmp fields.
-  bufferptr tmp;
-  bufferlist::iterator t = p;
-  t.copy_shallow(p.get_bl().length() - p.get_off(), tmp);
-  auto cp = tmp.begin();
-  traits::decode(o, cp);
-  p.advance((ssize_t)cp.get_offset());
+  const auto& bl = p.get_bl();
+  const auto remaining = bl.length() - p.get_off();
+  // it is expensive to rebuild a contigous buffer and drop it, so avoid this.
+  if (p.get_current_ptr().get_raw() != bl.back().get_raw() &&
+      remaining > CEPH_PAGE_SIZE) {
+    traits::decode(o, p);
+  } else {
+    // ensure we get a contigous buffer... until the end of the
+    // bufferlist.  we don't really know how much we'll need here,
+    // unfortunately.  hopefully it is already contiguous and we're just
+    // bumping the raw ref and initializing the ptr tmp fields.
+    bufferptr tmp;
+    bufferlist::iterator t = p;
+    t.copy_shallow(remaining, tmp);
+    auto cp = tmp.begin();
+    traits::decode(o, cp);
+    p.advance((ssize_t)cp.get_offset());
+  }
 }
 
-template<typename T, typename traits=denc_traits<T>>
+template<typename T,
+	 typename traits=denc_traits<T>>
 inline typename std::enable_if<traits::supported &&
-			       traits::featured>::type decode(
+			       traits::need_contiguous>::type decode(
   T& o,
   bufferlist::iterator& p)
 {
   if (p.end())
     throw buffer::end_of_buffer();
+  // ensure we get a contigous buffer... until the end of the
+  // bufferlist.  we don't really know how much we'll need here,
+  // unfortunately.  hopefully it is already contiguous and we're just
+  // bumping the raw ref and initializing the ptr tmp fields.
   bufferptr tmp;
   bufferlist::iterator t = p;
   t.copy_shallow(p.get_bl().length() - p.get_off(), tmp);
diff --git a/src/include/fs_types.h b/src/include/fs_types.h
index 5513fcefd792..dba085ad13e0 100644
--- a/src/include/fs_types.h
+++ b/src/include/fs_types.h
@@ -32,6 +32,7 @@ struct denc_traits<inodeno_t> {
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = true;
+  static constexpr bool need_contiguous = true;
   static void bound_encode(const inodeno_t &o, size_t& p) {
     denc(o.val, p);
   }
diff --git a/src/include/interval_set.h b/src/include/interval_set.h
index de7cf930894b..b84d8187052d 100644
--- a/src/include/interval_set.h
+++ b/src/include/interval_set.h
@@ -249,15 +249,21 @@ class interval_set {
     denc_traits<std::map<T,T>>::bound_encode(m, p);
   }
   void encode(bufferlist::contiguous_appender& p) const {
-    denc_traits<std::map<T,T>>::encode(m, p);
+    denc(m, p);
   }
   void decode(bufferptr::iterator& p) {
-    denc_traits<std::map<T,T>>::decode(m, p);
+    denc(m, p);
     _size = 0;
-    for (typename std::map<T,T>::const_iterator i = m.begin();
-         i != m.end();
-         i++)
-      _size += i->second;
+    for (const auto& i : m) {
+      _size += i.second;
+    }
+  }
+  void decode(bufferlist::iterator& p) {
+    denc(m, p);
+    _size = 0;
+    for (const auto& i : m) {
+      _size += i.second;
+    }
   }
 
   void encode_nohead(bufferlist::contiguous_appender& p) const {
@@ -266,10 +272,9 @@ class interval_set {
   void decode_nohead(int n, bufferptr::iterator& p) {
     denc_traits<std::map<T,T>>::decode_nohead(n, m, p);
     _size = 0;
-    for (typename std::map<T,T>::const_iterator i = m.begin();
-         i != m.end();
-         i++)
-      _size += i->second;
+    for (const auto& i : m) {
+      _size += i.second;
+    }
   }
 
   void clear() {
@@ -576,6 +581,7 @@ struct denc_traits<interval_set<T>> {
   static constexpr bool supported = true;
   static constexpr bool bounded = false;
   static constexpr bool featured = false;
+  static constexpr bool need_contiguous = denc_traits<T>::need_contiguous;
   static void bound_encode(const interval_set<T>& v, size_t& p) {
     v.bound_encode(p);
   }
@@ -586,6 +592,11 @@ struct denc_traits<interval_set<T>> {
   static void decode(interval_set<T>& v, bufferptr::iterator& p) {
     v.decode(p);
   }
+  template<typename U=T>
+    static typename std::enable_if<sizeof(U) && !need_contiguous>::type
+    decode(interval_set<T>& v, bufferlist::iterator& p) {
+    v.decode(p);
+  }
   static void encode_nohead(const interval_set<T>& v,
 			    bufferlist::contiguous_appender& p) {
     v.encode_nohead(p);
diff --git a/src/include/object.h b/src/include/object.h
index 672fbc4c3f91..1bad9ea24bcf 100644
--- a/src/include/object.h
+++ b/src/include/object.h
@@ -127,6 +127,7 @@ struct denc_traits<snapid_t> {
   static constexpr bool supported = true;
   static constexpr bool featured = false;
   static constexpr bool bounded = true;
+  static constexpr bool need_contiguous = true;
   static void bound_encode(const snapid_t& o, size_t& p) {
     denc(o.val, p);
   }
diff --git a/src/os/bluestore/bluestore_types.h b/src/os/bluestore/bluestore_types.h
index f60b53baf1dd..6841ead46b06 100644
--- a/src/os/bluestore/bluestore_types.h
+++ b/src/os/bluestore/bluestore_types.h
@@ -168,6 +168,7 @@ struct denc_traits<PExtentVector> {
   static constexpr bool supported = true;
   static constexpr bool bounded = false;
   static constexpr bool featured = false;
+  static constexpr bool need_contiguous = true;
   static void bound_encode(const PExtentVector& v, size_t& p) {
     p += sizeof(uint32_t);
     const auto size = v.size();
diff --git a/src/test/test_denc.cc b/src/test/test_denc.cc
index 0ea97a116991..8f702735bc80 100644
--- a/src/test/test_denc.cc
+++ b/src/test/test_denc.cc
@@ -15,6 +15,7 @@
  */
 
 #include <stdio.h>
+#include <numeric>
 
 #include "global/global_init.h"
 #include "common/ceph_argparse.h"
@@ -578,3 +579,120 @@ TEST(denc, optional)
     ASSERT_EQ(bpi.get_pos(), bl.c_str() + bl.length());
   }
 }
+
+// unlike legacy_t, Legacy supports denc() also.
+struct Legacy {
+  static unsigned n_denc;
+  static unsigned n_decode;
+  uint8_t value = 0;
+  DENC(Legacy, v, p) {
+    n_denc++;
+    denc(v.value, p);
+  }
+  void decode(buffer::list::iterator& p) {
+    n_decode++;
+    ::decode(value, p);
+  }
+  static void reset() {
+    n_denc = n_decode = 0;
+  }
+  static bufferlist encode_n(unsigned n, const vector<unsigned>& segments);
+};
+WRITE_CLASS_DENC(Legacy)
+unsigned Legacy::n_denc = 0;
+unsigned Legacy::n_decode = 0;
+
+bufferlist Legacy::encode_n(unsigned n, const vector<unsigned>& segments) {
+  vector<Legacy> v;
+  for (unsigned i = 0; i < n; i++) {
+    v.push_back(Legacy());
+  }
+  bufferlist bl(n * sizeof(uint8_t));
+  ::encode(v, bl);
+  bufferlist segmented;
+  auto p = bl.begin();
+
+  auto sum = std::accumulate(segments.begin(), segments.end(), 0u);
+  for (auto i : segments) {
+    buffer::ptr seg;
+    p.copy_deep(bl.length() * i / sum, seg);
+    segmented.push_back(seg);
+  }
+  p.copy_all(segmented);
+  return segmented;
+}
+
+TEST(denc, no_copy_if_segmented_and_lengthy)
+{
+  static_assert(_denc::has_legacy_denc<Legacy>::value,
+                "Legacy do have legacy denc");
+  {
+    // use denc() which shallow_copy() if the buffer is small
+    constexpr unsigned N_COPIES = 42;
+    const vector<unsigned> segs{50, 50}; // half-half
+    bufferlist segmented = Legacy::encode_n(N_COPIES, segs);
+    ASSERT_GT(segmented.get_num_buffers(), 1u);
+    ASSERT_LT(segmented.length(), CEPH_PAGE_SIZE);
+    auto p = segmented.begin();
+    vector<Legacy> v;
+    // denc() is shared by encode() and decode(), so reset() right before
+    // decode()
+    Legacy::reset();
+    ::decode(v, p);
+    ASSERT_EQ(N_COPIES, v.size());
+    ASSERT_EQ(N_COPIES, Legacy::n_denc);
+    ASSERT_EQ(0u, Legacy::n_decode);
+  }
+  {
+    // use denc() which shallow_copy() if the buffer is not segmented and large
+    const unsigned N_COPIES = CEPH_PAGE_SIZE * 2;
+    const vector<unsigned> segs{100};
+    bufferlist segmented = Legacy::encode_n(N_COPIES, segs);
+    ASSERT_EQ(segmented.get_num_buffers(), 1u);
+    ASSERT_GT(segmented.length(), CEPH_PAGE_SIZE);
+    auto p = segmented.begin();
+    vector<Legacy> v;
+    Legacy::reset();
+    ::decode(v, p);
+    ASSERT_EQ(N_COPIES, v.size());
+    ASSERT_EQ(N_COPIES, Legacy::n_denc);
+    ASSERT_EQ(0u, Legacy::n_decode);
+  }
+  {
+    // use denc() which shallow_copy() if the buffer is segmented and large,
+    // but the total size of the chunks to be decoded is smallish.
+    bufferlist large_bl = Legacy::encode_n(CEPH_PAGE_SIZE * 2, {50, 50});
+    bufferlist small_bl = Legacy::encode_n(100, {50, 50});
+    bufferlist segmented;
+    segmented.append(large_bl);
+    segmented.append(small_bl);
+    ASSERT_GT(segmented.get_num_buffers(), 1);
+    ASSERT_GT(segmented.length(), CEPH_PAGE_SIZE);
+    auto p = segmented.begin();
+    p.advance(large_bl.length());
+    ASSERT_LT(segmented.length() - p.get_off(), CEPH_PAGE_SIZE);
+    vector<Legacy> v;
+    Legacy::reset();
+    ::decode(v, p);
+    ASSERT_EQ(Legacy::n_denc, 100u);
+    ASSERT_EQ(0u, Legacy::n_decode);
+  }
+  {
+    // use decode() which avoids deep copy if the buffer is segmented and large
+    bufferlist small_bl = Legacy::encode_n(100, {50, 50});
+    bufferlist large_bl = Legacy::encode_n(CEPH_PAGE_SIZE * 2, {50, 50});
+    bufferlist segmented;
+    segmented.append(small_bl);
+    segmented.append(large_bl);
+    ASSERT_GT(segmented.get_num_buffers(), 1);
+    ASSERT_GT(segmented.length(), CEPH_PAGE_SIZE);
+    auto p = segmented.begin();
+    p.advance(small_bl.length());
+    ASSERT_GT(segmented.length() - p.get_off(), CEPH_PAGE_SIZE);
+    vector<Legacy> v;
+    Legacy::reset();
+    ::decode(v, p);
+    ASSERT_EQ(0u, Legacy::n_denc);
+    ASSERT_EQ(CEPH_PAGE_SIZE * 2, Legacy::n_decode);
+  }
+}
-- 
2.47.3