msg/async, v2: rx continuations use buffer::ptr_node.

Signed-off-by: Radoslaw Zarzynski <rzarzyns@redhat.com>

diff --git a/src/msg/async/Protocol.h b/src/msg/async/Protocol.h
index 556c9ef1470f92fadeb96f2fa7a7ea6618e0af2b..cccba1835676da7ff09de6a30c79c5308ae2c6b1 100644 (file)
--- a/src/msg/async/Protocol.h
+++ b/src/msg/async/Protocol.h
@@ -46,24 +46,18 @@ public:
   }
 };
 
-using rx_buffer_t = ceph::bufferptr;
-// FIXME: std::function in AsyncConnection requires us to be copy-
-// constructible just in the case - even if nobody actually copies
-// the std::functions there. This inhibits usage of unique_ptr of
-// ptr_node inside. Reworking the callback mechanism might help but
-// for now we can go with regular bptr.
-//
-//    std::unique_ptr<buffer::ptr_node, buffer::ptr_node::disposer>;
+using rx_buffer_t =
+    std::unique_ptr<buffer::ptr_node, buffer::ptr_node::disposer>;
 
 template <class C>
 class CtRxNode : public Ct<C> {
   using fn_t = Ct<C> *(C::*)(rx_buffer_t&&, int r);
   fn_t _f;
 
+public:
   mutable rx_buffer_t node;
   int r;
 
-public:
   CtRxNode(fn_t f) : _f(f) {}
   void setParams(rx_buffer_t &&node, int r) {
     this->node = std::move(node);

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index 4f810fd962d96bab73e492994e5aa8b39194c138..9d7ce00facd6666e18cd3ec5d1bb042f96984e5a 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -717,16 +717,17 @@ CtPtr ProtocolV2::read(CONTINUATION_RX_TYPE<ProtocolV2> &next,
 
 CtPtr ProtocolV2::read(CONTINUATION_RXBPTR_TYPE<ProtocolV2> &next,
                        rx_buffer_t &&buffer) {
-  const auto len = buffer.length();
-  const auto buf = buffer.c_str();
+  const auto len = buffer->length();
+  const auto buf = buffer->c_str();
+  next.node = std::move(buffer);
   ssize_t r = connection->read(len, buf,
-    [&next, this, bufnode = std::move(buffer)](char *buffer, int r) mutable {
-      next.setParams(std::move(bufnode), r);
+    [&next, this](char *buffer, int r) {
+      next.r = r;
       run_continuation(next);
     });
   if (r <= 0) {
     // error or done synchronously
-    next.setParams(rx_buffer_t(), r);
+    next.r = r;
     return &next;
   }
 
@@ -1098,8 +1099,8 @@ CtPtr ProtocolV2::read_frame_segment() {
   const auto& cur_rx_desc = rx_segments_desc.at(rx_segments_data.size());
   rx_buffer_t rx_buffer;
   try {
-    rx_buffer = buffer::create_aligned(
-      get_onwire_size(cur_rx_desc.length), cur_rx_desc.alignment);
+    rx_buffer = buffer::ptr_node::create(buffer::create_aligned(
+      get_onwire_size(cur_rx_desc.length), cur_rx_desc.alignment));
   } catch (std::bad_alloc&) {
     // Catching because of potential issues with satisfying alignment.
     ldout(cct, 20) << __func__ << " can't allocate aligned rx_buffer "