test/crimson: add timeout feature for blockers to fail

Signed-off-by: Yingxin Cheng <yingxin.cheng@intel.com>

diff --git a/src/crimson/net/Interceptor.h b/src/crimson/net/Interceptor.h
index 5e61e1c0261fc2c89cdf96f3f326649c219ce4d5..7bc07cc9b841ca453652b4aaf8684580214cc514 100644 (file)
--- a/src/crimson/net/Interceptor.h
+++ b/src/crimson/net/Interceptor.h
@@ -5,6 +5,7 @@
 
 #include <variant>
 #include <seastar/core/sharded.hh>
+#include <seastar/core/sleep.hh>
 
 #include "Fwd.h"
 #include "msg/async/frames_v2.h"
@@ -50,8 +51,8 @@ inline std::ostream& operator<<(std::ostream& out, const bp_action_t& action) {
 }
 
 class socket_blocker {
-  std::optional<seastar::promise<>> p_blocked;
-  std::optional<seastar::promise<>> p_unblocked;
+  std::optional<seastar::abort_source> p_blocked;
+  std::optional<seastar::abort_source> p_unblocked;
 
  public:
   seastar::future<> wait_blocked() {
@@ -59,25 +60,34 @@ class socket_blocker {
     if (p_unblocked) {
       return seastar::now();
     } else {
-      p_blocked = seastar::promise<>();
-      return p_blocked->get_future();
+      p_blocked = seastar::abort_source();
+      return seastar::sleep_abortable(10s, *p_blocked).then([this] {
+        throw std::runtime_error(
+            "Timeout (10s) in socket_blocker::wait_blocked()");
+      }).handle_exception_type([] (const seastar::sleep_aborted& e) {
+        // wait done!
+      });
     }
   }
 
   seastar::future<> block() {
     if (p_blocked) {
-      p_blocked->set_value();
+      p_blocked->request_abort();
       p_blocked = std::nullopt;
     }
     ceph_assert(!p_unblocked);
-    p_unblocked = seastar::promise<>();
-    return p_unblocked->get_future();
+    p_unblocked = seastar::abort_source();
+    return seastar::sleep_abortable(10s, *p_unblocked).then([this] {
+      ceph_abort("Timeout (10s) in socket_blocker::block()");
+    }).handle_exception_type([] (const seastar::sleep_aborted& e) {
+      // wait done!
+    });
   }
 
   void unblock() {
     ceph_assert(!p_blocked);
     ceph_assert(p_unblocked);
-    p_unblocked->set_value();
+    p_unblocked->request_abort();
     p_unblocked = std::nullopt;
   }
 };

diff --git a/src/test/crimson/test_messenger.cc b/src/test/crimson/test_messenger.cc
index 4b16be518ba788abe73452bc4ceb56d8d144ecf3..f3bdb106dd93d2d5d4a10fb7363f68660c60f43b 100644 (file)
--- a/src/test/crimson/test_messenger.cc
+++ b/src/test/crimson/test_messenger.cc
@@ -730,7 +730,7 @@ struct TestInterceptor : public Interceptor {
   std::map<Breakpoint, counter_t> breakpoints_counter;
   std::map<ConnectionRef, unsigned> conns;
   ConnResults results;
-  std::optional<seastar::promise<>> signal;
+  std::optional<seastar::abort_source> signal;
 
   TestInterceptor() = default;
   // only used for copy breakpoint configurations
@@ -768,13 +768,17 @@ struct TestInterceptor : public Interceptor {
 
   seastar::future<> wait() {
     assert(!signal);
-    signal = seastar::promise<>();
-    return signal->get_future();
+    signal = seastar::abort_source();
+    return seastar::sleep_abortable(10s, *signal).then([this] {
+      throw std::runtime_error("Timeout (10s) in TestInterceptor::wait()");
+    }).handle_exception_type([] (const seastar::sleep_aborted& e) {
+      // wait done!
+    });
   }
 
   void notify() {
     if (signal) {
-      signal->set_value();
+      signal->request_abort();
       signal = std::nullopt;
     }
   }