FIX | bug in gRPC Timer Object

Author: xiadnoring

include/cache/ManapiTL.hpp

@@ -32,16 +32,16 @@ namespace manapi {
         void iput (const K &key, V &&value, std::chrono::milliseconds duration) {
             this->cleanup();
             auto id = std::make_pair(std::chrono::steady_clock::now() + duration, std::forward<decltype(value)>(value));
-            auto it = this->m_data.insert({key, id});
-            if (!it.second) {
-                this->m_sorted.erase(std::make_pair(it.first->second.first, it.first->first));
-                it.first->second = id;
+            auto it = this->m_data.find(key);
+            if (it != this->m_data.end()) {
+                this->m_sorted.erase(std::make_pair(it->second.first, it->first));
             }
+            it = this->m_data.insert_or_assign(std::move(key), std::move(id)).first;
             try {
-                this->m_sorted.insert(std::make_pair(id.first, it.first->first));
+                this->m_sorted.insert(std::make_pair(it->second.first, it->first));
             }
             catch (...) {
-                this->m_data.erase(it.first);
+                this->m_data.erase(it);
                 std::rethrow_exception(std::current_exception());
             }
         }

include/ext/pq/AsyncPostgreImpl.ipp

@@ -378,6 +378,14 @@ manapi::future<manapi::status_or<manapi::ext::pq::item>> manapi::ext::pq::db::ma
     co_return status_not_found("db:not found");
 }
 
+bool manapi::ext::pq::db::has_master() const {
+    return !!this->m_master;
+}
+
+bool manapi::ext::pq::db::has_slaves() const {
+    return !!this->m_slaves.size();
+}
+
 manapi::future<manapi::ext::pq::status_or<manapi::ext::pq::result>> manapi::ext::pq::db::pexec(ktypes type, const char *command, int nParams, const Oid *paramTypes, const char * const *paramValues, const int *paramLengths,const int *paramFormats, ctoken token) {
     while (true) {
         auto wrk_res = type == kMaster ? (co_await this->master()) :(co_await this->slave());

include/ext/pq/AsyncPostgrePool.hpp

@@ -105,6 +105,10 @@ namespace manapi::ext::pq {
 
         future<manapi::status_or<item>> master ();
 
+        bool has_master () const;
+
+        bool has_slaves () const;
+
         template<typename ...Args>
         manapi::future<pq::status_or<pq::result>> exec (ktypes type, const std::string &sql, Args &&...args) {
             std::string buffer;

src/ManapiGrpc.cpp

@@ -16,6 +16,7 @@
 #if MANAPIHTTP_GRPC_DEPENDENCY
 
 #include <grpcpp/grpcpp.h>
+#include <grpc/event_engine/event_engine.h>
 
 enum manapi_grpc_endpoint_flags {
     MANAPI_GRPC_ENDPOINT_WANT_READ = 1,
@@ -1251,10 +1252,11 @@ grpc_event_engine::experimental::EventEngine::TaskHandle manapi::net::wgrpc::eve
 
         auto rhs = ctx->timerpool()->append_timer_sync(ms,
             [td] (manapi::timer timer) -> void {
-                try { td->closure->Run(); }
+                auto closure = std::move(td->closure);
+                task_handle_cancel(manapi::async::current().get(), td->index);
+                try { closure->Run(); }
                 catch (std::exception const &e) { manapi_log_error("%s:%s failed due to %s",
                     "wgrpc", "gRPC send a error", e.what()); }
-                task_handle_cancel(manapi::async::current().get(), td->index);
             });
 
         *timer = rhs.unwrap();
@@ -1349,10 +1351,11 @@ grpc_event_engine::experimental::EventEngine::TaskHandle manapi::net::wgrpc::eve
 
         auto rhs = ctx->timerpool()->append_timer_sync(ms,
             [td] (manapi::timer timer) mutable -> void {
-                try { td->closure (); }
+                auto closure = std::move(td->closure);
+                task_handle_cancel(manapi::async::current().get(), td->index);
+                try { closure (); }
                 catch (std::exception const &e) { manapi_log_error(
                     "%s:%s failed due to %s", "wgrpc", "gRPC send a error", e.what()); }
-                task_handle_cancel(manapi::async::current().get(), td->index);
             });
 
         *timer = rhs.unwrap();

src/worker/ManapiTcp.cpp

@@ -355,9 +355,8 @@ manapi::net::worker::shared_conn manapi::net::worker::TCP::accept (const ev::sha
             connection->wrk.flags |= WRK_INTERFACE_CONN_RETRY;
         }
 
-        auto res = it->second.insert({reinterpret_cast<uintptr_t>(connection.get()),
-            connection});
-        assert((res.second));
+        auto res = it->second.insert_or_assign(reinterpret_cast<uintptr_t>(connection.get()),
+            connection);
     }
     catch (std::exception const &e) {
         manapi_log_error("tcp accept: failed due to %s", e.what());