[rdma]Fix some rdma bugs (#973)

Author: qicosmos

.clang-format

@@ -11,9 +11,4 @@ AllowShortIfStatementsOnASingleLine: Never
 Language: JavaScript
 DisableFormat: true
 ---
-# Ignore json file
-Language: Json
-DisableFormat: true
----
-Language: Proto
-DisableFormat: true
+

cmake/develop.cmake

@@ -50,7 +50,7 @@ if(ENABLE_SANITIZER AND NOT MSVC)
         endif()
     else()
         ## address santizer
-        if(CMAKE_BUILD_TYPE STREQUAL "Debug")
+        if(CMAKE_BUILD_TYPE STREQUAL "Debug" OR CMAKE_BUILD_TYPE STREQUAL "RelWithDebInfo")
             check_asan(HAS_ASAN)
             if(HAS_ASAN)
                 set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address")

include/ylt/coro_io/client_pool.hpp

@@ -194,7 +194,7 @@ class client_pool : public std::enable_shared_from_this<
         co_await coro_io::sleep_for(wait_time, &client->get_executor());
       self = watcher.lock();
       ++i;
-    } while (i < self->pool_config_.connect_retry_count);
+    } while (self && i < self->pool_config_.connect_retry_count);
     ELOG_WARN << "reconnect client{" << client.get() << "},host:{"
               << client->get_host() << ":" << client->get_port()
               << "} out of max limit, stop retry. connect failed";
@@ -220,7 +220,7 @@ class client_pool : public std::enable_shared_from_this<
         co_return;
       }
       auto executor = self->io_context_pool_.get_executor();
-      auto client = std::make_unique<client_t>(*executor);
+      auto client = std::make_unique<client_t>(executor);
       if (!client->init_config(client_config))
         AS_UNLIKELY {
           ELOG_ERROR << "Init client config failed in host alive detect. That "
@@ -271,7 +271,7 @@ class client_pool : public std::enable_shared_from_this<
     if (client == nullptr) {
       std::unique_ptr<client_t> cli;
       auto executor = io_context_pool_.get_executor();
-      client = std::make_unique<client_t>(*executor);
+      client = std::make_unique<client_t>(executor);
       if (!client->init_config(client_config))
         AS_UNLIKELY {
           ELOG_ERROR << "init client config failed.";
@@ -298,8 +298,10 @@ class client_pool : public std::enable_shared_from_this<
             this->weak_from_this(), clients,
             (std::max)(collect_time, std::chrono::milliseconds{50}),
             pool_config_.idle_queue_per_max_clear_count)
-            .directlyStart([](auto&&) {
-            },coro_io::get_global_executor());
+            .directlyStart(
+                [](auto&&) {
+                },
+                coro_io::get_global_executor());
       }
     }
   }

include/ylt/coro_io/coro_io.hpp

@@ -240,6 +240,18 @@ dispatch(Func func, Executor executor) {
   co_return co_await awaitor.await_resume(helper);
 }
 
+template <typename Executor>
+inline async_simple::coro::Lazy<async_simple::Try<void>> dispatch(
+    Executor executor) {
+  if (executor.running_in_this_thread()) {
+    co_return async_simple::Try<void>{};
+  }
+  co_return co_await dispatch(
+      []() {
+      },
+      executor);
+}
+
 namespace detail {
 
 template <typename T>

include/ylt/coro_io/detail/client_queue.hpp

@@ -63,16 +63,14 @@ class client_queue {
   }
   bool try_dequeue(client_t& c) {
     const int_fast16_t index = selected_index_;
-    if (size_[index ^ 1]) {
-      if (queue_[index ^ 1].try_dequeue(c)) {
-        --size_[index ^ 1];
-        return true;
-      }
-    }
     if (queue_[index].try_dequeue(c)) {
       --size_[index];
       return true;
     }
+    if (queue_[index ^ 1].try_dequeue(c)) {
+      --size_[index ^ 1];
+      return true;
+    }
     return false;
   }
   std::size_t clear_old(std::size_t max_clear_cnt) {