1 files changed, 71 insertions, 48 deletions

diff --git a/src/librbd/migration/HttpClient.cc b/src/librbd/migration/HttpClient.cc
index 6a504d3a9ac..d212981a917 100644
--- a/src/librbd/migration/HttpClient.cc
+++ b/src/librbd/migration/HttpClient.cc
@@ -63,14 +63,13 @@ public:
     m_on_shutdown = on_finish;
 
     auto current_state = m_state;
+    m_state = STATE_SHUTTING_DOWN;
+
     if (current_state == STATE_UNINITIALIZED) {
       // never initialized or resolve/connect failed
       on_finish->complete(0);
       return;
-    }
-
-    m_state = STATE_SHUTTING_DOWN;
-    if (current_state != STATE_READY) {
+    } else if (current_state != STATE_READY) {
       // delay shutdown until current state transition completes
       return;
     }
@@ -118,7 +117,7 @@ public:
     ceph_assert(m_http_client->m_strand.running_in_this_thread());
 
     auto cct = m_http_client->m_cct;
-    ldout(cct, 20) << "work=" << work.get() << ", r=" << -ec.value() << dendl;
+    ldout(cct, 20) << "work=" << work.get() << ", ec=" << ec.what() << dendl;
 
     ceph_assert(m_in_flight_requests > 0);
     --m_in_flight_requests;
@@ -187,13 +186,14 @@ protected:
   virtual void connect(boost::asio::ip::tcp::resolver::results_type results,
                        Context* on_finish) = 0;
   virtual void disconnect(Context* on_finish) = 0;
+  virtual void reset_stream() = 0;
 
   void close_socket() {
     auto cct = m_http_client->m_cct;
     ldout(cct, 15) << dendl;
 
     boost::system::error_code ec;
-    boost::beast::get_lowest_layer(derived().stream()).socket().close(ec);
+    derived().stream().lowest_layer().close(ec);
   }
 
 private:
@@ -229,7 +229,6 @@ private:
     auto cct = m_http_client->m_cct;
     ldout(cct, 15) << dendl;
 
-    shutdown_socket();
     m_resolver.async_resolve(
       m_http_client->m_url_spec.host, m_http_client->m_url_spec.port,
       [this, on_finish](boost::system::error_code ec, auto results) {
@@ -358,8 +357,7 @@ private:
   }
 
   int shutdown_socket() {
-    if (!boost::beast::get_lowest_layer(
-          derived().stream()).socket().is_open()) {
+    if (!derived().stream().lowest_layer().is_open()) {
       return 0;
     }
 
@@ -367,7 +365,7 @@ private:
     ldout(cct, 15) << dendl;
 
     boost::system::error_code ec;
-    boost::beast::get_lowest_layer(derived().stream()).socket().shutdown(
+    derived().stream().lowest_layer().shutdown(
       boost::asio::ip::tcp::socket::shutdown_both, ec);
 
     if (ec && ec != boost::beast::errc::not_connected) {
@@ -414,7 +412,7 @@ private:
   void handle_receive(boost::system::error_code ec,
                       std::shared_ptr<Work>&& work) {
     auto cct = m_http_client->m_cct;
-    ldout(cct, 15) << "work=" << work.get() << ", r=" << -ec.value() << dendl;
+    ldout(cct, 15) << "work=" << work.get() << ", ec=" << ec.what() << dendl;
 
     ceph_assert(m_in_flight_requests > 0);
     --m_in_flight_requests;
@@ -445,10 +443,10 @@ private:
         ldout(cct, 5) << "remote peer stream closed, retrying request" << dendl;
         m_receive_queue.push_front(work);
       } else if (ec == boost::beast::error::timeout) {
-        lderr(cct) << "timed-out while issuing request" << dendl;
+        lderr(cct) << "timed-out while receiving response" << dendl;
         work->complete(-ETIMEDOUT, {});
       } else {
-        lderr(cct) << "failed to issue request: " << ec.message() << dendl;
+        lderr(cct) << "failed to receive response: " << ec.message() << dendl;
         work->complete(-ec.value(), {});
       }
 
@@ -473,7 +471,7 @@ private:
       r = -EACCES;
     } else if (boost::beast::http::to_status_class(result) !=
                  boost::beast::http::status_class::successful) {
-      lderr(cct) << "failed to retrieve size: HTTP " << result << dendl;
+      lderr(cct) << "failed to retrieve resource: HTTP " << result << dendl;
       r = -EIO;
     }
 
@@ -501,7 +499,10 @@ private:
                    << "next_state=" << next_state << ", "
                    << "r=" << r << dendl;
 
-    m_state = next_state;
+    if (current_state != STATE_SHUTTING_DOWN) {
+      m_state = next_state;
+    }
+
     if (current_state == STATE_CONNECTING) {
       if (next_state == STATE_UNINITIALIZED) {
         shutdown_socket();
@@ -512,14 +513,17 @@ private:
         return;
       }
     } else if (current_state == STATE_SHUTTING_DOWN) {
+      ceph_assert(m_on_shutdown != nullptr);
       if (next_state == STATE_READY) {
         // shut down requested while connecting/resetting
         disconnect(new LambdaContext([this](int r) { handle_shut_down(r); }));
         return;
       } else if (next_state == STATE_UNINITIALIZED ||
-                 next_state == STATE_SHUTDOWN ||
                  next_state == STATE_RESET_CONNECTING) {
-        ceph_assert(m_on_shutdown != nullptr);
+        shutdown_socket();
+        m_on_shutdown->complete(r);
+        return;
+      } else if (next_state == STATE_SHUTDOWN) {
         m_on_shutdown->complete(r);
         return;
       }
@@ -528,6 +532,7 @@ private:
       ceph_assert(next_state == STATE_RESET_CONNECTING);
       ceph_assert(on_finish == nullptr);
       shutdown_socket();
+      reset_stream();
       resolve_host(nullptr);
       return;
     } else if (current_state == STATE_RESET_CONNECTING) {
@@ -589,7 +594,7 @@ public:
     this->close_socket();
   }
 
-  inline boost::beast::tcp_stream&
+  inline boost::asio::ip::tcp::socket&
   stream() {
     return m_stream;
   }
@@ -601,20 +606,25 @@ protected:
     auto cct = http_client->m_cct;
     ldout(cct, 15) << dendl;
 
-    m_stream.async_connect(
-      results,
-      [on_finish](boost::system::error_code ec, const auto& endpoint) {
-        on_finish->complete(-ec.value());
-      });
+    ceph_assert(!m_stream.is_open());
+    boost::asio::async_connect(m_stream,
+			       results,
+			       [on_finish](boost::system::error_code ec,
+					   const auto& endpoint) {
+				 on_finish->complete(-ec.value());
+			       });
   }
 
   void disconnect(Context* on_finish) override {
     on_finish->complete(0);
   }
 
-private:
-  boost::beast::tcp_stream m_stream;
+  void reset_stream() override {
+    // no-op -- tcp_stream object can be reused after shut down
+  }
 
+private:
+  boost::asio::ip::tcp::socket m_stream;
 };
 
 #undef dout_prefix
@@ -633,7 +643,7 @@ public:
     this->close_socket();
   }
 
-  inline boost::beast::ssl_stream<boost::beast::tcp_stream>&
+  inline boost::asio::ssl::stream<boost::asio::ip::tcp::socket>&
   stream() {
     return m_stream;
   }
@@ -645,7 +655,9 @@ protected:
     auto cct = http_client->m_cct;
     ldout(cct, 15) << dendl;
 
-    boost::beast::get_lowest_layer(m_stream).async_connect(
+    ceph_assert(!m_stream.lowest_layer().is_open());
+    async_connect(
+      m_stream.lowest_layer(),
       results,
       [this, on_finish](boost::system::error_code ec, const auto& endpoint) {
         handle_connect(-ec.value(), on_finish);
@@ -657,19 +669,25 @@ protected:
     auto cct = http_client->m_cct;
     ldout(cct, 15) << dendl;
 
-    if (!m_ssl_enabled) {
-      on_finish->complete(0);
-      return;
-    }
-
     m_stream.async_shutdown(
-      asio::util::get_callback_adapter([this, on_finish](int r) {
-        shutdown(r, on_finish); }));
+      [this, on_finish](boost::system::error_code ec) {
+        handle_disconnect(ec, on_finish);
+      });
+  }
+
+  void reset_stream() override {
+    auto http_client = this->m_http_client;
+    auto cct = http_client->m_cct;
+    ldout(cct, 15) << dendl;
+
+    // ssl_stream object can't be reused after shut down -- move-in
+    // a freshly constructed instance
+    m_stream = boost::asio::ssl::stream<boost::asio::ip::tcp::socket>(
+      http_client->m_strand, http_client->m_ssl_context);
   }
 
 private:
-  boost::beast::ssl_stream<boost::beast::tcp_stream> m_stream;
-  bool m_ssl_enabled = false;
+  boost::asio::ssl::stream<boost::asio::ip::tcp::socket> m_stream;
 
   void handle_connect(int r, Context* on_finish) {
     auto http_client = this->m_http_client;
@@ -728,33 +746,38 @@ private:
     // Perform the SSL/TLS handshake
     m_stream.async_handshake(
       boost::asio::ssl::stream_base::client,
-      asio::util::get_callback_adapter(
-        [this, on_finish](int r) { handle_handshake(r, on_finish); }));
+      [this, on_finish](boost::system::error_code ec) {
+        handle_handshake(ec, on_finish);
+      });
   }
 
-  void handle_handshake(int r, Context* on_finish) {
+  void handle_handshake(boost::system::error_code ec, Context* on_finish) {
     auto http_client = this->m_http_client;
     auto cct = http_client->m_cct;
-    ldout(cct, 15) << "r=" << r << dendl;
+    ldout(cct, 15) << "ec=" << ec.what() << dendl;
 
-    if (r < 0) {
-      lderr(cct) << "failed to complete handshake: " << cpp_strerror(r)
+    if (ec) {
+      lderr(cct) << "failed to complete SSL handshake: " << ec.message()
                  << dendl;
-      disconnect(new LambdaContext([r, on_finish](int) {
-        on_finish->complete(r); }));
+      on_finish->complete(-ec.value());
       return;
     }
 
-    m_ssl_enabled = true;
     on_finish->complete(0);
   }
 
-  void shutdown(int r, Context* on_finish) {
+  void handle_disconnect(boost::system::error_code ec, Context* on_finish) {
     auto http_client = this->m_http_client;
     auto cct = http_client->m_cct;
-    ldout(cct, 15) << "r=" << r << dendl;
+    ldout(cct, 15) << "ec=" << ec.what() << dendl;
 
-    on_finish->complete(r);
+    if (ec && ec != boost::asio::ssl::error::stream_truncated) {
+      lderr(cct) << "failed to shut down SSL: " << ec.message() << dendl;
+      on_finish->complete(-ec.value());
+      return;
+    }
+
+    on_finish->complete(0);
   }
 };