[coro_io][ibverbs] fix close failed to notify peer (#980)

Author: poor-circle

include/ylt/coro_io/ibverbs/ib_buffer.hpp

@@ -5,6 +5,7 @@
 
 #include <atomic>
 #include <cerrno>
+#include <cmath>
 #include <cstddef>
 #include <cstdint>
 #include <memory>
@@ -114,9 +115,12 @@ class ib_buffer_pool_t : public std::enable_shared_from_this<ib_buffer_pool_t> {
                    << "}, now ib_buffer count: " << self->free_buffers_.size();
         std::size_t clear_cnt = self->free_buffers_.clear_old(1000);
         self->total_memory_ -= clear_cnt * self->buffer_size();
-        ELOG_WARN << "finish ib_buffer timeout free of pool{" << self.get()
+        ELOG_INFO << "finish ib_buffer timeout free of pool{" << self.get()
                   << "}, now ib_buffer cnt: " << self->free_buffers_.size()
-                  << " mem usage:" << self->total_memory_;
+                  << " mem usage:"
+                  << (int64_t)(std::round(self->total_memory_ /
+                                          (1.0 * 1024 * 1024)))
+                  << " MB";
         if (clear_cnt != 0) {
           try {
             co_await async_simple::coro::Yield{};

include/ylt/coro_io/ibverbs/ib_socket.hpp

@@ -206,7 +206,7 @@ struct ib_socket_shared_state_t
           sr.opcode = IBV_WR_SEND;
           sr.send_flags |= IBV_SEND_SIGNALED;
           std::error_code err;
-          if (self->has_close_) {
+          if (self->fd_ == nullptr) {
             err = std::make_error_code(std::errc::operation_canceled);
           }
           // post the receive request to the RQ
@@ -240,7 +240,7 @@ struct ib_socket_shared_state_t
       ELOG_ERROR << std::make_error_code(std::errc{r}).message();
       return std::make_error_code(std::errc{r});
     }
-    struct ibv_wc wc {};
+    struct ibv_wc wc{};
     int ne = 0;
     std::vector<resume_struct> vec;
     callback_t tmp_callback;

include/ylt/coro_rpc/impl/coro_rpc_server.hpp

@@ -172,7 +172,7 @@ class coro_rpc_server_base {
       }
       errc_ = listen();
       if (!errc_) {
-        if constexpr (requires(typename server_config::executor_pool_t & pool) {
+        if constexpr (requires(typename server_config::executor_pool_t &pool) {
                         pool.run();
                       }) {
           thd_ = std::thread([this] {
@@ -420,7 +420,12 @@ class coro_rpc_server_base {
       }
 #endif
       if (error) {
-        ELOG_ERROR << "accept failed, error: " << error.message();
+        if (error == asio::error::operation_aborted) {
+          ELOG_INFO << "server was canceled:" << error.message();
+        }
+        else {
+          ELOG_ERROR << "server accept failed:" << error.message();
+        }
         if (error == asio::error::operation_aborted ||
             error == asio::error::bad_descriptor) {
           acceptor_close_waiter_.set_value();

include/ylt/thirdparty/async_simple/CommonMacros.h

@@ -43,18 +43,26 @@
 #endif  // __SANITIZE_ADDRESS__
 #endif  // __GNUC__
 
+#if __has_cpp_attribute(clang::coro_only_destroy_when_complete)
+#define CORO_ONLY_DESTROY_WHEN_DONE [[clang::coro_only_destroy_when_complete]]
+#else
 #if defined(__alibaba_clang__) && \
     __has_cpp_attribute(ACC::coro_only_destroy_when_complete)
 #define CORO_ONLY_DESTROY_WHEN_DONE [[ACC::coro_only_destroy_when_complete]]
 #else
 #define CORO_ONLY_DESTROY_WHEN_DONE
 #endif
+#endif
 
+#if __has_cpp_attribute(clang::coro_await_elidable)
+#define ELIDEABLE_AFTER_AWAIT [[clang::coro_await_elidable]]
+#else
 #if defined(__alibaba_clang__) && \
     __has_cpp_attribute(ACC::elideable_after_await)
 #define ELIDEABLE_AFTER_AWAIT [[ACC::elideable_after_await]]
 #else
 #define ELIDEABLE_AFTER_AWAIT
 #endif
+#endif
 
 #endif

include/ylt/thirdparty/async_simple/Signal.h

@@ -17,6 +17,7 @@
 #define ASYNC_SIMPLE_SIGNAL_H
 
 #ifndef ASYNC_SIMPLE_USE_MODULES
+
 #include <assert.h>
 #include <any>
 #include <atomic>
@@ -94,7 +95,7 @@ class Signal : public std::enable_shared_from_this<Signal> {
     // binding slots, then execute the slot callback functions. It will return
     // the signal which success triggered. If no signal success triggger, return
     // SignalType::none.
-    SignalType emit(SignalType state) noexcept;
+    SignalType emits(SignalType state) noexcept;
 
     // Return now signal type.
     SignalType state() const noexcept {
@@ -213,7 +214,8 @@ class Slot {
             (signal()->state() & type)) {
             return false;
         }
-        // if signal triggered later, we will found it by cas failed.
+        // if signal triggered later, we will found it by atomic handler CAS
+        // failed.
         auto oldHandler = oldHandlerPtr->load(std::memory_order_acquire);
         if (oldHandler ==
             &detail::SignalSlotSharedState::HandlerManager::emittedTag) {
@@ -318,7 +320,7 @@ class Slot {
                 [chainedSignal =
                      chainedSignal->weak_from_this()](SignalType type) {
                     if (auto signal = chainedSignal.lock(); signal != nullptr) {
-                        signal->emit(type);
+                        signal->emits(type);
                     }
                 }),
             std::memory_order_release);
@@ -449,7 +451,7 @@ inline detail::SignalSlotSharedState::~SignalSlotSharedState() {
     }
 }
 
-inline SignalType Signal::emit(SignalType state) noexcept {
+inline SignalType Signal::emits(SignalType state) noexcept {
     if (state != SignalType::None) {
         SignalType vaildSignal = UpdateState(_state, state);
         if (vaildSignal) {

include/ylt/thirdparty/async_simple/async_simple.cppm

@@ -15,23 +15,30 @@ module;
 #include <stdio.h>
 #include <cassert>
 #include <climits>
+#include <cstdint>
 #include <version>
 
 #ifdef __linux__
 #include <sched.h>
 #endif
 
+#if __has_include(<libaio.h>) && !defined(ASYNC_SIMPLE_HAS_NOT_AIO)
+#include <libaio.h>
+#endif
+
 export module async_simple;
 import std;
 #define ASYNC_SIMPLE_USE_MODULES
 export extern "C++" {
   #include "util/move_only_function.h"
   #include "coro/Traits.h"
+  #include "CommonMacros.h"
+  #include "Common.h"
   #include "MoveWrapper.h"
   #include "experimental/coroutine.h"
+  #include "async_simple/Signal.h"
   #include "Executor.h"
-  #include "CommonMacros.h"
-  #include "Common.h"
+  #include "async_simple/coro/LazyLocalBase.h"
   #include "Unit.h"
   #include "Try.h"
   #include "FutureState.h"
@@ -41,6 +48,9 @@ export extern "C++" {
   #include "Promise.h"
   #include "coro/DetachedCoroutine.h"
   #include "coro/ViaCoroutine.h"
+#if defined(__clang_major__)  && __clang_major__ >= 17
+  #include "coro/PromiseAllocator.h"
+#endif
   #include "coro/Lazy.h"
   #include "uthread/internal/thread_impl.h"
   #include "uthread/Await.h"
@@ -71,7 +81,6 @@ export extern "C++" {
   #include "coro/Semaphore.h"
   // There are some bugs in clang lower versions.
 #if defined(__clang_major__)  && __clang_major__ >= 17
-  #include "coro/PromiseAllocator.h"
   #include "coro/Generator.h"
 #endif
 }

include/ylt/thirdparty/async_simple/coro/Collect.h

@@ -167,7 +167,7 @@ struct CollectAnyAwaiter {
                            SignalType type, Signal*) mutable {
                     auto count = e->downCount();
                     if (count == size + 1) {
-                      c.resume();
+                        c.resume();
                     }
                 })) {  // has canceled
             return false;
@@ -187,13 +187,13 @@ struct CollectAnyAwaiter {
                     auto count = e->downCount();
                     // n+1: n coro + 1 cancel handler
                     if (count == size + 1) {
-                      _result = std::make_unique<ResultType>();
-                      _result->_idx = i;
-                      _result->_value = std::move(result);
-                      if (auto ptr = local->getSlot(); ptr) {
-                        ptr->signal()->emit(_SignalType);
-                      }
-                      c.resume();
+                        _result = std::make_unique<ResultType>();
+                        _result->_idx = i;
+                        _result->_value = std::move(result);
+                        if (auto ptr = local->getSlot(); ptr) {
+                            ptr->signal()->emits(_SignalType);
+                        }
+                        c.resume();
                     }
                 });
         }  // end for
@@ -269,7 +269,7 @@ struct CollectAnyVariadicAwaiter {
                                                      Signal*) mutable {
                     auto count = e->downCount();
                     if (count == std::tuple_size<InputType>() + 1) {
-                      c.resume();
+                        c.resume();
                     }
                 })) {  // has canceled
             return false;
@@ -291,12 +291,12 @@ struct CollectAnyVariadicAwaiter {
                         auto count = e->downCount();
                         // n+1: n coro + 1 cancel handler
                         if (count == std::tuple_size<InputType>() + 1) {
-                          _result = std::make_unique<ResultType>(
-                              std::in_place_index_t<index>(), std::move(res));
-                          if (auto ptr = local->getSlot(); ptr) {
-                            ptr->signal()->emit(_SignalType);
-                          }
-                          c.resume();
+                            _result = std::make_unique<ResultType>(
+                                std::in_place_index_t<index>(), std::move(res));
+                            if (auto ptr = local->getSlot(); ptr) {
+                                ptr->signal()->emits(_SignalType);
+                            }
+                            c.resume();
                         }
                     });
             }(),
@@ -392,15 +392,14 @@ struct CollectAllAwaiter {
         _event.setAwaitingCoro(continuation);
         auto size = _input.size();
         for (size_t i = 0; i < size; ++i) {
-          auto& exec = _input[i]._coro.promise()._executor;
-          if (exec == nullptr) {
-            exec = executor;
-          }
-          std::unique_ptr<LazyLocalBase> local;
-          local = std::make_unique<LazyLocalBase>(_signal.get());
-          _input[i]._coro.promise()._lazy_local = local.get();
-          auto&& func =
-              [this, i, local = std::move(local)]() mutable {
+            auto& exec = _input[i]._coro.promise()._executor;
+            if (exec == nullptr) {
+                exec = executor;
+            }
+            std::unique_ptr<LazyLocalBase> local;
+            local = std::make_unique<LazyLocalBase>(_signal.get());
+            _input[i]._coro.promise()._lazy_local = local.get();
+            auto&& func = [this, i, local = std::move(local)]() mutable {
                 _input[i].start([this, i, local = std::move(local)](
                                     Try<ValueType>&& result) {
                     _output[i] = std::move(result);
@@ -410,21 +409,21 @@ struct CollectAllAwaiter {
                     auto signalType = _SignalType;
                     auto awaitingCoro = _event.down(oldCount, 1);
                     if (oldCount == size) {
-                        signal->emit(signalType);
+                        signal->emits(signalType);
                     }
                     if (awaitingCoro) {
                         awaitingCoro.resume();
                     }
                 });
-              };
-          if (Para == true && _input.size() > 1) {
-            if (exec != nullptr)
-              AS_LIKELY {
-                exec->schedule_move_only(std::move(func));
-                continue;
-              }
-          }
-          func();
+            };
+            if (Para == true && _input.size() > 1) {
+                if (exec != nullptr)
+                    AS_LIKELY {
+                        exec->schedule_move_only(std::move(func));
+                        continue;
+                    }
+            }
+            func();
         }
     }
     inline auto await_resume() { return std::move(_output); }
@@ -583,7 +582,7 @@ struct CollectAllVariadicAwaiter {
                         auto signalType = _SignalType;
                         auto awaitingCoro = _event.down(oldCount, 1);
                         if (oldCount == sizeof...(Ts)) {
-                            signal->emit(signalType);
+                            signal->emits(signalType);
                         }
                         if (awaitingCoro) {
                             awaitingCoro.resume();

include/ylt/thirdparty/async_simple/coro/CountEvent.h

@@ -34,15 +34,15 @@ namespace detail {
 // The last 'down' will resume the awaiting coroutine on this event.
 class CountEvent {
 public:
- CountEvent(size_t count) : _count(count) {}
- CountEvent(const CountEvent&) = delete;
- CountEvent(CountEvent&& other)
-     : _count(other._count.exchange(0, std::memory_order_relaxed)),
-       _awaitingCoro(std::exchange(other._awaitingCoro, nullptr)) {}
+    CountEvent(size_t count) : _count(count) {}
+    CountEvent(const CountEvent&) = delete;
+    CountEvent(CountEvent&& other)
+        : _count(other._count.exchange(0, std::memory_order_relaxed)),
+          _awaitingCoro(std::exchange(other._awaitingCoro, nullptr)) {}
 
- [[nodiscard]] CoroHandle<> down(size_t n = 1) {
-   std::size_t oldCount;
-   return down(oldCount, n);
+    [[nodiscard]] CoroHandle<> down(size_t n = 1) {
+        std::size_t oldCount;
+        return down(oldCount, n);
     }
     [[nodiscard]] CoroHandle<> down(size_t& oldCount, std::size_t n) {
         // read acquire and write release, _awaitingCoro store can not be