Merge pull request ceph#58179 from cbodley/wip-rgw-assert-async

rgw: test that no asio threads are blocked on null_yield

Reviewed-by: Adam Emerson <[email protected]>

Author: cbodley

src/common/options/rgw.yaml.in

@@ -2238,6 +2238,14 @@ options:
   services:
   - rgw
   with_legacy: true
+- name: rgw_asio_assert_yielding
+  type: bool
+  level: dev
+  desc: Trigger an assertion failure if an operation would block an asio thread
+  default: false
+  services:
+  - rgw
+  with_legacy: true
 - name: rgw_user_quota_bucket_sync_interval
   type: int
   level: advanced

src/rgw/CMakeLists.txt

@@ -63,6 +63,7 @@ set(librgw_common_srcs
   rgw_acl_swift.cc
   rgw_aio.cc
   rgw_aio_throttle.cc
+  rgw_asio_thread.cc
   rgw_auth.cc
   rgw_auth_s3.cc
   rgw_arn.cc

src/rgw/driver/rados/rgw_data_sync.cc

@@ -277,7 +277,7 @@ class RGWReadRemoteDataLogShardInfoCR : public RGWCoroutine {
           return io_block(0);
         }
         yield {
-          op_ret = http_op->wait(shard_info, null_yield);
+          op_ret = http_op->wait(dpp, shard_info, null_yield);
           http_op->put();
         }
 
@@ -377,7 +377,7 @@ class RGWReadRemoteDataLogShardCR : public RGWCoroutine {
         }
         yield {
           timer.reset();
-          op_ret = http_op->wait(&response, null_yield);
+          op_ret = http_op->wait(dpp, &response, null_yield);
           http_op->put();
         }
 
@@ -495,7 +495,7 @@ class RGWListRemoteDataLogShardCR : public RGWSimpleCoroutine {
   }
 
   int request_complete() override {
-    int ret = http_op->wait(result, null_yield);
+    int ret = http_op->wait(sync_env->dpp, result, null_yield);
     http_op->put();
     if (ret < 0 && ret != -ENOENT) {
       ldpp_dout(sync_env->dpp, 5) << "ERROR: failed to list remote datalog shard, ret=" << ret << dendl;

src/rgw/driver/rados/rgw_lc_tier.cc

@@ -269,7 +269,7 @@ static int cloud_tier_get_object(RGWLCCloudTierCtx& tier_ctx, bool head,
   }
 
   /* fetch headers */
-  ret = tier_ctx.conn.complete_request(in_req, nullptr, nullptr, nullptr, nullptr, &headers, null_yield);
+  ret = tier_ctx.conn.complete_request(tier_ctx.dpp, in_req, nullptr, nullptr, nullptr, nullptr, &headers, null_yield);
   if (ret < 0 && ret != -ENOENT) {
     ldpp_dout(tier_ctx.dpp, 20) << "ERROR: " << __func__ << "(): conn.complete_request() returned ret=" << ret << dendl;
     return ret;
@@ -704,8 +704,7 @@ RGWGetDataCB *RGWLCCloudStreamPut::get_cb() {
 }
 
 int RGWLCCloudStreamPut::complete_request() {
-  int ret = conn.complete_request(out_req, etag, &obj_properties.mtime, null_yield);
-  return ret;
+  return conn.complete_request(dpp, out_req, etag, &obj_properties.mtime, null_yield);
 }
 
 /* Read local copy and write to Cloud endpoint */

src/rgw/driver/rados/rgw_notify.cc

@@ -248,7 +248,7 @@ class Manager : public DoutPrefixProvider {
                         << " retry_number: "
                         << entry_persistency_tracker.retires_num
                         << " current time: " << time_now << dendl;
-    const auto ret = push_endpoint->send(event_entry.event, yield);
+    const auto ret = push_endpoint->send(this, event_entry.event, yield);
     if (ret < 0) {
       ldpp_dout(this, 5) << "WARNING: push entry marker: " << entry.marker
                          << " failed. error: " << ret
@@ -1266,7 +1266,7 @@ int publish_commit(rgw::sal::Object* obj,
 	  dpp->get_cct());
         ldpp_dout(res.dpp, 20) << "INFO: push endpoint created: "
 			       << topic.cfg.dest.push_endpoint << dendl;
-        const auto ret = push_endpoint->send(event_entry.event, res.yield);
+        const auto ret = push_endpoint->send(dpp, event_entry.event, res.yield);
         if (ret < 0) {
           ldpp_dout(dpp, 1)
               << "ERROR: failed to push sync notification event with error: "

src/rgw/driver/rados/rgw_pubsub_push.cc

@@ -9,6 +9,7 @@
 #include "common/Formatter.h"
 #include "common/iso_8601.h"
 #include "common/async/completion.h"
+#include "rgw_asio_thread.h"
 #include "rgw_common.h"
 #include "rgw_data_sync.h"
 #include "rgw_pubsub.h"
@@ -88,7 +89,8 @@ class RGWPubSubHTTPEndpoint : public RGWPubSubEndpoint {
     }
   }
 
-  int send(const rgw_pubsub_s3_event& event, optional_yield y) override {
+  int send(const DoutPrefixProvider* dpp, const rgw_pubsub_s3_event& event,
+           optional_yield y) override {
     std::shared_lock lock(s_http_manager_mutex);
     if (!s_http_manager) {
       ldout(cct, 1) << "ERROR: send failed. http endpoint manager not running" << dendl;
@@ -114,7 +116,7 @@ class RGWPubSubHTTPEndpoint : public RGWPubSubEndpoint {
     if (perfcounter) perfcounter->inc(l_rgw_pubsub_push_pending);
     auto rc = s_http_manager->add_request(&request);
     if (rc == 0) {
-      rc = request.wait(y);
+      rc = request.wait(dpp, y);
     }
     if (perfcounter) perfcounter->dec(l_rgw_pubsub_push_pending);
     // TODO: use read_bl to process return code and handle according to ack level
@@ -144,7 +146,7 @@ class Waiter {
   mutable std::condition_variable cond;
 
 public:
-  int wait(optional_yield y) {
+  int wait(const DoutPrefixProvider* dpp, optional_yield y) {
     std::unique_lock l{lock};
     if (done) {
       return ret;
@@ -160,6 +162,8 @@ class Waiter {
           }, token, yield.get_executor());
       return -ec.value();
     }
+    maybe_warn_about_blocking(dpp);
+
     cond.wait(l, [this]{return (done==true);});
     return ret;
   }
@@ -247,7 +251,7 @@ class RGWPubSubAMQPEndpoint : public RGWPubSubEndpoint {
     }
   }
 
-  int send(const rgw_pubsub_s3_event& event, optional_yield y) override {
+  int send(const DoutPrefixProvider* dpp, const rgw_pubsub_s3_event& event, optional_yield y) override {
     if (ack_level == ack_level_t::None) {
       return amqp::publish(conn_id, topic, json_format_pubsub_event(event));
     } else {
@@ -262,7 +266,7 @@ class RGWPubSubAMQPEndpoint : public RGWPubSubEndpoint {
         // failed to publish, does not wait for reply
         return rc;
       }
-      return w->wait(y);
+      return w->wait(dpp, y);
     }
   }
 
@@ -320,7 +324,8 @@ class RGWPubSubKafkaEndpoint : public RGWPubSubEndpoint {
    }
  }
 
-  int send(const rgw_pubsub_s3_event& event, optional_yield y) override {
+  int send(const DoutPrefixProvider* dpp, const rgw_pubsub_s3_event& event,
+           optional_yield y) override {
     if (ack_level == ack_level_t::None) {
       return kafka::publish(conn_id, topic, json_format_pubsub_event(event));
     } else {
@@ -332,7 +337,7 @@ class RGWPubSubKafkaEndpoint : public RGWPubSubEndpoint {
         // failed to publish, does not wait for reply
         return rc;
       }
-      return w->wait(y);
+      return w->wait(dpp, y);
     }
   }
 

src/rgw/driver/rados/rgw_pubsub_push.h

@@ -8,6 +8,7 @@
 #include "include/common_fwd.h"
 #include "common/async/yield_context.h"
 
+class DoutPrefixProvider;
 class RGWHTTPArgs;
 struct rgw_pubsub_s3_event;
 
@@ -28,7 +29,9 @@ class RGWPubSubEndpoint {
 
   // this method is used in order to send notification and wait for completion 
   // in async manner via a coroutine when invoked in the frontend environment
-  virtual int send(const rgw_pubsub_s3_event& event, optional_yield y) = 0;
+  virtual int send(const DoutPrefixProvider* dpp,
+                   const rgw_pubsub_s3_event& event,
+                   optional_yield y) = 0;
 
   // present as string
   virtual std::string to_str() const = 0;

src/rgw/driver/rados/rgw_rados.cc

@@ -4084,7 +4084,7 @@ int RGWRados::stat_remote_obj(const DoutPrefixProvider *dpp,
       return ret;
     }
 
-    ret = conn->complete_request(in_stream_req, nullptr, &set_mtime, psize,
+    ret = conn->complete_request(dpp, in_stream_req, nullptr, &set_mtime, psize,
                                  nullptr, pheaders, y);
     if (ret < 0) {
       if (ret == -EIO && tries < NUM_ENPOINT_IOERROR_RETRIES - 1) {
@@ -4319,7 +4319,7 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& obj_ctx,
       goto set_err_state;
     }
 
-    ret = conn->complete_request(in_stream_req, &etag, &set_mtime,
+    ret = conn->complete_request(rctx.dpp, in_stream_req, &etag, &set_mtime,
                                  &accounted_size, nullptr, nullptr, rctx.y);
     if (ret < 0) {
       if (ret == -EIO && tries < NUM_ENPOINT_IOERROR_RETRIES - 1) {
@@ -4580,7 +4580,7 @@ int RGWRados::copy_obj_to_remote_dest(const DoutPrefixProvider *dpp,
       return ret;
     }
 
-    ret = rest_master_conn->complete_request(out_stream_req, etag, mtime, y);
+    ret = rest_master_conn->complete_request(dpp, out_stream_req, etag, mtime, y);
     if (ret < 0) {
       if (ret == -EIO && tries < NUM_ENPOINT_IOERROR_RETRIES - 1) {
         ldpp_dout(dpp, 20) << __func__  << "(): failed to put_obj_async_init. retries=" << tries << dendl;
@@ -7792,7 +7792,7 @@ int RGWRados::block_while_resharding(RGWRados::BucketShard *bs,
       } // if taking of lock succeeded
     } // block to encapsulate recovery from incomplete reshard
 
-    ret = reshard_wait->wait(y);
+    ret = reshard_wait->wait(dpp, y);
     if (ret < 0) {
       ldpp_dout(dpp, 0) << __func__ <<
 	" ERROR: bucket is still resharding, please retry" << dendl;

src/rgw/driver/rados/rgw_reshard.cc

@@ -7,6 +7,7 @@
 
 #include "rgw_zone.h"
 #include "driver/rados/rgw_bucket.h"
+#include "rgw_asio_thread.h"
 #include "rgw_reshard.h"
 #include "rgw_sal.h"
 #include "rgw_sal_rados.h"
@@ -1261,7 +1262,7 @@ int RGWReshard::clear_bucket_resharding(const DoutPrefixProvider *dpp, const str
   return 0;
 }
 
-int RGWReshardWait::wait(optional_yield y)
+int RGWReshardWait::wait(const DoutPrefixProvider* dpp, optional_yield y)
 {
   std::unique_lock lock(mutex);
 
@@ -1285,6 +1286,7 @@ int RGWReshardWait::wait(optional_yield y)
     waiters.erase(waiters.iterator_to(waiter));
     return -ec.value();
   }
+  maybe_warn_about_blocking(dpp);
 
   cond.wait_for(lock, duration);
 

src/rgw/driver/rados/rgw_reshard.h

@@ -266,7 +266,7 @@ class RGWReshardWait {
   ~RGWReshardWait() {
     ceph_assert(going_down);
   }
-  int wait(optional_yield y);
+  int wait(const DoutPrefixProvider* dpp, optional_yield y);
   // unblock any threads waiting on reshard
   void stop();
 };