msg/async, v2: use bptr continuation for segment reading.

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

diff --git a/src/msg/async/ProtocolV2.cc b/src/msg/async/ProtocolV2.cc
index 49189c170043f79548f5db08d443f9e399948e33..4f810fd962d96bab73e492994e5aa8b39194c138 100644 (file)
--- a/src/msg/async/ProtocolV2.cc
+++ b/src/msg/async/ProtocolV2.cc
@@ -59,7 +59,7 @@ void ProtocolV2::run_continuation(CtRef continuation) {
 
 #define READ(L, C) read(CONTINUATION(C), L)
 
-#define READB(L, B, C) read(CONTINUATION(C), L, B)
+#define READ_RXBUF(B, C) read(CONTINUATION(C), B)
 
 #ifdef UNIT_TESTS_BUILT
 
@@ -1096,10 +1096,10 @@ CtPtr ProtocolV2::read_frame_segment() {
 
   // description of current segment to read
   const auto& cur_rx_desc = rx_segments_desc.at(rx_segments_data.size());
-  std::unique_ptr<buffer::ptr_node, buffer::ptr_node::disposer> rx_buffer;
+  rx_buffer_t rx_buffer;
   try {
-    rx_buffer = ceph::buffer::ptr_node::create(buffer::create_aligned(
-      get_onwire_size(cur_rx_desc.length), cur_rx_desc.alignment));
+    rx_buffer = 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 "
@@ -1109,13 +1109,10 @@ CtPtr ProtocolV2::read_frame_segment() {
     return _fault();
   }
 
-  rx_segments_data.emplace_back();
-  rx_segments_data.back().push_back(std::move(rx_buffer));
-  return READB(rx_segments_data.back().length(), rx_segments_data.back().c_str(),
-    handle_read_frame_segment);
+  return READ_RXBUF(std::move(rx_buffer), handle_read_frame_segment);
 }
 
-CtPtr ProtocolV2::handle_read_frame_segment(char *buffer, int r) {
+CtPtr ProtocolV2::handle_read_frame_segment(rx_buffer_t &&rx_buffer, int r) {
   ldout(cct, 20) << __func__ << " r=" << r << dendl;
 
   if (r < 0) {
@@ -1124,6 +1121,9 @@ CtPtr ProtocolV2::handle_read_frame_segment(char *buffer, int r) {
     return _fault();
   }
 
+  rx_segments_data.emplace_back();
+  rx_segments_data.back().push_back(std::move(rx_buffer));
+
   // decrypt incoming data
   // FIXME: if (auth_meta->is_mode_secure()) {
   if (session_stream_handlers.rx) {

diff --git a/src/msg/async/ProtocolV2.h b/src/msg/async/ProtocolV2.h
index 242a2bea3557e8c38f833d74ba469fb15980ea4e..592cc2b1a8fc2e00f9cd34aeb1bedc32779c5bc6 100644 (file)
--- a/src/msg/async/ProtocolV2.h
+++ b/src/msg/async/ProtocolV2.h
@@ -154,7 +154,7 @@ private:
   CONTINUATION_DECL(ProtocolV2, read_frame);
   CONTINUATION_DECL(ProtocolV2, finish_auth);
   READ_HANDLER_CONTINUATION_DECL(ProtocolV2, handle_read_frame_preamble_main);
-  READ_HANDLER_CONTINUATION_DECL(ProtocolV2, handle_read_frame_segment);
+  READ_BPTR_HANDLER_CONTINUATION_DECL(ProtocolV2, handle_read_frame_segment);
   READ_HANDLER_CONTINUATION_DECL(ProtocolV2, handle_read_frame_epilogue_main);
   CONTINUATION_DECL(ProtocolV2, throttle_message);
   CONTINUATION_DECL(ProtocolV2, throttle_bytes);
@@ -164,7 +164,7 @@ private:
   Ct<ProtocolV2> *finish_auth();
   Ct<ProtocolV2> *handle_read_frame_preamble_main(char *buffer, int r);
   Ct<ProtocolV2> *read_frame_segment();
-  Ct<ProtocolV2> *handle_read_frame_segment(char *buffer, int r);
+  Ct<ProtocolV2> *handle_read_frame_segment(rx_buffer_t &&rx_buffer, int r);
   Ct<ProtocolV2> *handle_read_frame_epilogue_main(char *buffer, int r);
   Ct<ProtocolV2> *handle_read_frame_dispatch();
   Ct<ProtocolV2> *handle_frame_payload();