improve dispatcher shutdown

3 files changed, 67 insertions, 72 deletions

diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h
index 6b9b486..56d2d6a 100644
--- a/include/salticidae/conn.h
+++ b/include/salticidae/conn.h
@@ -210,6 +210,7 @@ class ConnPool {
     class Worker {
         EventContext ec;
         ThreadCall tcall;
+        BoxObj<ThreadCall> exit_tcall; /** only used by the dispatcher thread */
         std::thread handle;
         bool disp_flag;
         std::atomic<size_t> nconn;
@@ -297,10 +298,18 @@ class ConnPool {
             tcall.async_call([this](ThreadCall::Handle &) { ec.stop(); });
         }
 
+        void disp_stop() {
+            assert(disp_flag && exit_tcall);
+            exit_tcall->async_call([this](ThreadCall::Handle &) { ec.stop(); });
+        }
+
         std::thread &get_handle() { return handle; }
         const EventContext &get_ec() { return ec; }
         ThreadCall *get_tcall() { return &tcall; }
-        void set_dispatcher() { disp_flag = true; }
+        void set_dispatcher() {
+            disp_flag = true;
+            exit_tcall = new ThreadCall(ec);
+        }
         bool is_dispatcher() const { return disp_flag; }
         size_t get_nconn() { return nconn; }
         void stop_tcall() { tcall.stop(); }
@@ -461,10 +470,9 @@ class ConnPool {
         workers[0].set_dispatcher();
         disp_error_cb = [this](const std::exception_ptr err) {
             workers[0].stop_tcall();
-            disp_ec.stop();
             user_tcall->async_call([this, err](ThreadCall::Handle &) {
-                stop_workers();
-                //std::rethrow_exception(err);
+                for (size_t i = 1; i < nworker; i++)
+                    workers[i].stop();
                 if (error_cb) error_cb(err, true, -1);
             });
         };
@@ -504,7 +512,7 @@ class ConnPool {
         system_state = 2;
         SALTICIDAE_LOG_INFO("stopping all threads...");
         /* stop the dispatcher */
-        workers[0].stop();
+        workers[0].disp_stop();
         workers[0].get_handle().join();
         /* stop all workers */
         for (size_t i = 1; i < nworker; i++)
@@ -532,19 +540,13 @@ class ConnPool {
 
     /** Actively connect to remote addr. */
     conn_t connect_sync(const NetAddr &addr) {
-        auto ret = *(static_cast<std::pair<conn_t, std::exception_ptr> *>(
+        auto ret = *(static_cast<conn_t *>(
                     disp_tcall->call([this, addr](ThreadCall::Handle &h) {
             conn_t conn;
-            std::exception_ptr err = nullptr;
-            try {
-                conn = _connect(addr);
-            } catch (...) {
-                err = std::current_exception();
-            }
-            h.set_result(std::make_pair(std::move(conn), err));
+            conn = _connect(addr);
+            h.set_result(conn);
         }).get()));
-        if (ret.second) std::rethrow_exception(ret.second);
-        return std::move(ret.first);
+        return std::move(ret);
     }
 
     /** Actively connect to remote addr (async). */
@@ -565,17 +567,9 @@ class ConnPool {
      * Does not need to be called if do not want to accept any passive
      * connections. */
     void listen(NetAddr listen_addr) {
-        auto ret = *(static_cast<std::exception_ptr *>(
-                disp_tcall->call([this, listen_addr](ThreadCall::Handle &h) {
-            std::exception_ptr err = nullptr;
-            try {
-                _listen(listen_addr);
-            } catch (...) {
-                err = std::current_exception();
-            }
-            h.set_result(err);
-        }).get()));
-        if (ret) std::rethrow_exception(ret);
+        disp_tcall->call([this, listen_addr](ThreadCall::Handle &) {
+            _listen(listen_addr);
+        }).get();
     }
 
     template<typename Func>
diff --git a/include/salticidae/event.h b/include/salticidae/event.h
index 960588c..420c073 100644
--- a/include/salticidae/event.h
+++ b/include/salticidae/event.h
@@ -656,6 +656,7 @@ class ThreadCall {
     public:
     struct Result {
         void *data;
+        std::exception_ptr error;
         std::function<void(void *)> deleter;
         Result(): data(nullptr) {}
         Result(void *data, std::function<void(void *)> &&deleter):
@@ -663,11 +664,14 @@ class ThreadCall {
         ~Result() { if (data != nullptr) deleter(data); }
         Result(const Result &) = delete;
         Result(Result &&other):
-                data(other.data), deleter(std::move(other.deleter)) {
+                data(other.data),
+                error(std::move(other.error)),
+                deleter(std::move(other.deleter)) {
             other.data = nullptr;
         }
         void swap(Result &other) {
             std::swap(data, other.data);
+            std::swap(error, other.error);
             std::swap(deleter, other.deleter);
         }
         Result &operator=(const Result &other) = delete;
@@ -679,7 +683,10 @@ class ThreadCall {
             }
             return *this;
         }
-        void *get() { return data; }
+        void *get() {
+            if (error) std::rethrow_exception(error);
+            return data;
+        }
     };
     class Handle {
         std::function<void(Handle &)> callback;
@@ -689,15 +696,18 @@ class ThreadCall {
         public:
         Handle(): notifier(nullptr) {}
         Handle(const Handle &) = delete;
-        void exec() {
-            callback(*this);
+        void return_sync() {
             if (notifier)
                 notifier->notify(std::move(result));
         }
+        void exec() {
+            callback(*this);
+            return_sync();
+        }
         template<typename T>
-        void set_result(T &&data) {
+        Result &set_result(T &&data) {
             using _T = std::remove_reference_t<T>;
-            result = Result(new _T(std::forward<T>(data)),
+            return result = Result(new _T(std::forward<T>(data)),
                             [](void *ptr) {delete static_cast<_T *>(ptr);});
         }
     };
@@ -711,7 +721,13 @@ class ThreadCall {
             Handle *h;
             while (q.try_dequeue(h))
             {
-                if (!stopped) h->exec();
+                try {
+                    if (!stopped) h->exec();
+                    else throw SalticidaeError(SALTI_ERROR_NOT_AVAIL);
+                } catch (...) {
+                    h->set_result(0).error = std::current_exception();
+                    h->return_sync();
+                }
                 delete h;
                 if (++cnt == burst_size) return true;
             }
@@ -726,7 +742,6 @@ class ThreadCall {
 
     template<typename Func>
     bool async_call(Func &&callback) {
-        if (stopped) return false;
         auto h = new Handle();
         h->callback = std::forward<Func>(callback);
         q.enqueue(h);
@@ -735,7 +750,6 @@ class ThreadCall {
 
     template<typename Func>
     Result call(Func &&callback) {
-        if (stopped) throw SalticidaeError(SALTI_ERROR_NOT_AVAIL);
         auto h = new Handle();
         h->callback = std::forward<Func>(callback);
         ThreadNotifier<Result> notifier;
@@ -746,6 +760,7 @@ class ThreadCall {
 
     const EventContext &get_ec() const { return ec; }
     void stop() { stopped = true; }
+    bool is_stopped() { return stopped; }
 };
 
 }
diff --git a/include/salticidae/network.h b/include/salticidae/network.h
index 4fc575c..f664493 100644
--- a/include/salticidae/network.h
+++ b/include/salticidae/network.h
@@ -978,22 +978,14 @@ void PeerNetwork<O, _, __>::pong_handler(MsgPong &&msg, const conn_t &conn) {
 
 template<typename O, O _, O __>
 void PeerNetwork<O, _, __>::listen(NetAddr _listen_addr) {
-    auto ret = *(static_cast<std::exception_ptr *>(
-            this->disp_tcall->call([this, _listen_addr](ThreadCall::Handle &h) {
-        std::exception_ptr err = nullptr;
-        try {
-            MsgNet::_listen(_listen_addr);
-            listen_addr = _listen_addr;
-            uint8_t rand_bytes[32];
-            if (!RAND_bytes(rand_bytes, 32))
-                throw PeerNetworkError(SALTI_ERROR_RAND_SOURCE);
-            my_nonce.load(rand_bytes);
-        } catch (...) {
-            err = std::current_exception();
-        }
-        h.set_result(std::move(err));
-    }).get()));
-    if (ret) std::rethrow_exception(ret);
+    this->disp_tcall->call([this, _listen_addr](ThreadCall::Handle &) {
+        MsgNet::_listen(_listen_addr);
+        listen_addr = _listen_addr;
+        uint8_t rand_bytes[32];
+        if (!RAND_bytes(rand_bytes, 32))
+            throw PeerNetworkError(SALTI_ERROR_RAND_SOURCE);
+        my_nonce.load(rand_bytes);
+    }).get();
 }
 
 template<typename O, O _, O __>
@@ -1054,31 +1046,25 @@ int32_t PeerNetwork<O, _, __>::del_peer(const NetAddr &addr) {
 template<typename O, O _, O __>
 typename PeerNetwork<O, _, __>::conn_t
 PeerNetwork<O, _, __>::get_peer_conn(const NetAddr &addr) const {
-    auto ret = *(static_cast<std::pair<conn_t, std::exception_ptr> *>(
+    auto ret = *(static_cast<conn_t *>(
             this->disp_tcall->call([this, addr](ThreadCall::Handle &h) {
         conn_t conn;
-        std::exception_ptr err = nullptr;
-        try {
-            pinfo_slock_t _g(known_peers_lock);
-            pinfo_slock_t __g(pid2peer_lock);
-            auto it = known_peers.find(addr);
-            if (it == known_peers.end())
-                throw PeerNetworkError(SALTI_ERROR_PEER_NOT_EXIST);
-            if (it->second->peer_id.is_null())
-                conn = nullptr;
-            else
-            {
-                auto it2 = pid2peer.find(it->second->peer_id);
-                assert(it2 != pid2peer.end());
-                conn = it2->second->conn;
-            }
-        } catch (...) {
-            err = std::current_exception();
+        pinfo_slock_t _g(known_peers_lock);
+        pinfo_slock_t __g(pid2peer_lock);
+        auto it = known_peers.find(addr);
+        if (it == known_peers.end())
+            throw PeerNetworkError(SALTI_ERROR_PEER_NOT_EXIST);
+        if (it->second->peer_id.is_null())
+            conn = nullptr;
+        else
+        {
+            auto it2 = pid2peer.find(it->second->peer_id);
+            assert(it2 != pid2peer.end());
+            conn = it2->second->conn;
         }
-        h.set_result(std::make_pair(std::move(conn), err));
+        h.set_result(std::move(conn));
     }).get()));
-    if (ret.second) std::rethrow_exception(ret.second);
-    return std::move(ret.first);
+    return std::move(ret);
 }
 
 template<typename O, O _, O __>