From 69a9bed21f18728483320e88530045180796e2ac Mon Sep 17 00:00:00 2001 From: Determinant Date: Thu, 4 Jul 2019 00:28:30 -0400 Subject: improve dispatcher shutdown --- include/salticidae/conn.h | 46 +++++++++++++++------------------ include/salticidae/event.h | 33 +++++++++++++++++------- include/salticidae/network.h | 60 +++++++++++++++++--------------------------- 3 files changed, 67 insertions(+), 72 deletions(-) (limited to 'include') 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 exit_tcall; /** only used by the dispatcher thread */ std::thread handle; bool disp_flag; std::atomic 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 *>( + auto ret = *(static_cast( 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( - 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 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 deleter; Result(): data(nullptr) {} Result(void *data, std::function &&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 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 - void set_result(T &&data) { + Result &set_result(T &&data) { using _T = std::remove_reference_t; - result = Result(new _T(std::forward(data)), + return result = Result(new _T(std::forward(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 bool async_call(Func &&callback) { - if (stopped) return false; auto h = new Handle(); h->callback = std::forward(callback); q.enqueue(h); @@ -735,7 +750,6 @@ class ThreadCall { template Result call(Func &&callback) { - if (stopped) throw SalticidaeError(SALTI_ERROR_NOT_AVAIL); auto h = new Handle(); h->callback = std::forward(callback); ThreadNotifier 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::pong_handler(MsgPong &&msg, const conn_t &conn) { template void PeerNetwork::listen(NetAddr _listen_addr) { - auto ret = *(static_cast( - 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 @@ -1054,31 +1046,25 @@ int32_t PeerNetwork::del_peer(const NetAddr &addr) { template typename PeerNetwork::conn_t PeerNetwork::get_peer_conn(const NetAddr &addr) const { - auto ret = *(static_cast *>( + auto ret = *(static_cast( 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 -- cgit v1.2.3-70-g09d2