aboutsummaryrefslogtreecommitdiff
path: root/include
diff options
context:
space:
mode:
authorDeterminant <[email protected]>2019-06-12 19:14:40 -0400
committerDeterminant <[email protected]>2019-06-12 19:14:40 -0400
commite27e529e589ef89fbe010ebf7c5635ec2873f64f (patch)
tree76e7d20589d11f3aabd255ca70201f1288d1d1bc /include
parent9f6460c7ab774d900f391345bbf3fac8617a3aa3 (diff)
WIP: error handling
Diffstat (limited to 'include')
-rw-r--r--include/salticidae/conn.h171
-rw-r--r--include/salticidae/event.h15
-rw-r--r--include/salticidae/netaddr.h10
-rw-r--r--include/salticidae/network.h195
-rw-r--r--include/salticidae/queue.h1
-rw-r--r--include/salticidae/util.h48
6 files changed, 302 insertions, 138 deletions
diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h
index cb637cb..42e87aa 100644
--- a/include/salticidae/conn.h
+++ b/include/salticidae/conn.h
@@ -50,10 +50,6 @@
namespace salticidae {
-struct ConnPoolError: public SalticidaeError {
- using SalticidaeError::SalticidaeError;
-};
-
/** Abstraction for connection management. */
class ConnPool {
class Worker;
@@ -63,6 +59,7 @@ class ConnPool {
using conn_t = ArcObj<Conn>;
/** The type of callback invoked when connection status is changed. */
using conn_callback_t = std::function<void(const conn_t &, bool)>;
+ using error_callback_t = std::function<void(const std::exception &, bool)>;
/** Abstraction for a bi-directional connection. */
class Conn {
friend ConnPool;
@@ -151,6 +148,10 @@ class ConnPool {
ThreadCall* disp_tcall;
/** Should be implemented by derived class to return a new Conn object. */
virtual Conn *create_conn() = 0;
+ using _error_callback_t = std::function<void(const std::exception_ptr err)>;
+ _error_callback_t disp_error_cb;
+ _error_callback_t worker_error_cb;
+
private:
const int max_listen_backlog;
@@ -161,6 +162,7 @@ class ConnPool {
/* owned by user loop */
BoxObj<ThreadCall> user_tcall;
conn_callback_t conn_cb;
+ error_callback_t error_cb;
/* owned by the dispatcher */
FdEvent ev_listen;
@@ -174,15 +176,27 @@ class ConnPool {
}
class Worker {
+ public:
+
+ private:
EventContext ec;
ThreadCall tcall;
std::thread handle;
bool disp_flag;
std::atomic<size_t> nconn;
+ ConnPool::_error_callback_t on_fatal_error;
public:
Worker(): tcall(ec), disp_flag(false), nconn(0) {}
+ void set_error_callback(ConnPool::_error_callback_t _on_error) {
+ on_fatal_error = std::move(_on_error);
+ }
+
+ void error_callback(const std::exception_ptr err) const {
+ on_fatal_error(err);
+ }
+
/* the following functions are called by the dispatcher */
void start() {
handle = std::thread([this]() { ec.dispatch(); });
@@ -191,36 +205,40 @@ class ConnPool {
void feed(const conn_t &conn, int client_fd) {
/* the caller should finalize all the preparation */
tcall.async_call([this, conn, client_fd](ThreadCall::Handle &) {
- if (conn->mode == Conn::ConnMode::DEAD)
- {
- SALTICIDAE_LOG_INFO("worker %x discarding dead connection",
- std::this_thread::get_id());
- return;
- }
- assert(conn->fd != -1);
- SALTICIDAE_LOG_INFO("worker %x got %s",
- std::this_thread::get_id(),
- std::string(*conn).c_str());
- conn->get_send_buffer()
- .get_queue()
- .reg_handler(this->ec, [conn, client_fd]
- (MPSCWriteBuffer::queue_t &) {
- if (conn->ready_send)
+ try {
+ if (conn->mode == Conn::ConnMode::DEAD)
{
- conn->ev_socket.del();
- conn->ev_socket.add(FdEvent::READ | FdEvent::WRITE);
- conn->send_data(client_fd, FdEvent::WRITE);
+ SALTICIDAE_LOG_INFO("worker %x discarding dead connection",
+ std::this_thread::get_id());
+ return;
}
- return false;
- });
- conn->ev_socket = FdEvent(ec, client_fd, [conn=conn](int fd, int what) {
- if (what & FdEvent::READ)
- conn->recv_data(fd, what);
- else
- conn->send_data(fd, what);
- });
- conn->ev_socket.add(FdEvent::READ | FdEvent::WRITE);
- nconn++;
+ assert(conn->fd != -1);
+ SALTICIDAE_LOG_INFO("worker %x got %s",
+ std::this_thread::get_id(),
+ std::string(*conn).c_str());
+ conn->get_send_buffer()
+ .get_queue()
+ .reg_handler(this->ec, [conn, client_fd]
+ (MPSCWriteBuffer::queue_t &) {
+ if (conn->ready_send)
+ {
+ conn->ev_socket.del();
+ conn->ev_socket.add(FdEvent::READ | FdEvent::WRITE);
+ conn->send_data(client_fd, FdEvent::WRITE);
+ }
+ return false;
+ });
+ conn->ev_socket = FdEvent(ec, client_fd, [this, conn=conn](int fd, int what) {
+ try {
+ if (what & FdEvent::READ)
+ conn->recv_data(fd, what);
+ else
+ conn->send_data(fd, what);
+ } catch (...) { on_fatal_error(std::current_exception()); }
+ });
+ conn->ev_socket.add(FdEvent::READ | FdEvent::WRITE);
+ nconn++;
+ } catch (...) { on_fatal_error(std::current_exception()); }
});
}
@@ -253,18 +271,6 @@ class ConnPool {
private:
- //class DspMulticast: public DispatchCmd {
- // std::vector<conn_t> receivers;
- // bytearray_t data;
- // public:
- // DspMulticast(std::vector<conn_t> &&receivers, bytearray_t &&data):
- // receivers(std::move(receivers)),
- // data(std::move(data)) {}
- // void exec(ConnPool *) override {
- // for (auto &r: receivers) r->write(bytearray_t(data));
- // }
- //};
-
Worker &select_worker() {
size_t idx = 0;
size_t best = workers[idx].get_nconn();
@@ -280,6 +286,15 @@ class ConnPool {
return workers[idx];
}
+ void on_fatal_error(const std::exception &error) {
+ stop_workers();
+ if (error_cb) error_cb(error, true);
+ }
+
+ void on_recoverable_error(const std::exception &error) {
+ if (error_cb) error_cb(error, false);
+ }
+
public:
class Config {
@@ -338,6 +353,30 @@ class ConnPool {
disp_ec = workers[0].get_ec();
disp_tcall = workers[0].get_tcall();
workers[0].set_dispatcher();
+ disp_error_cb = [this](const std::exception_ptr _err) {
+ user_tcall->async_call([this, _err](ThreadCall::Handle &) {
+ try {
+ std::rethrow_exception(_err);
+ } catch (const std::exception &err) {
+ on_fatal_error(err);
+ }
+ });
+ };
+
+ worker_error_cb = [this](const std::exception_ptr err) {
+ disp_tcall->async_call([this, err](ThreadCall::Handle &) {
+ // forward to the dispatcher
+ disp_error_cb(err);
+ });
+ };
+ for (size_t i = 0; i < nworker; i++)
+ {
+ auto &worker = workers[i];
+ if (worker.is_dispatcher())
+ worker.set_error_callback(disp_error_cb);
+ else
+ worker.set_error_callback(worker_error_cb);
+ }
}
~ConnPool() { stop(); }
@@ -388,17 +427,28 @@ class ConnPool {
conn_t connect(const NetAddr &addr, bool blocking = true) {
if (blocking)
{
- auto ret = *(static_cast<conn_t *>(disp_tcall->call(
+ auto ret = *(static_cast<std::pair<conn_t, std::exception_ptr> *>(disp_tcall->call(
[this, addr](ThreadCall::Handle &h) {
- auto conn = _connect(addr);
- h.set_result(std::move(conn));
+ 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));
}).get()));
- return std::move(ret);
+ if (ret.second) std::rethrow_exception(ret.second);
+ return std::move(ret.first);
}
else
{
disp_tcall->async_call([this, addr](ThreadCall::Handle &) {
- _connect(addr);
+ try {
+ _connect(addr);
+ } catch (...) {
+ disp_error_cb(std::current_exception());
+ }
});
return nullptr;
}
@@ -408,17 +458,32 @@ class ConnPool {
* Does not need to be called if do not want to accept any passive
* connections. */
void listen(NetAddr listen_addr) {
- disp_tcall->call([this, listen_addr](ThreadCall::Handle &) {
- _listen(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);
}
template<typename Func>
void reg_conn_handler(Func cb) { conn_cb = cb; }
+ template<typename Func>
+ void reg_error_handler(Func cb) { error_cb = cb; }
+
void terminate(const conn_t &conn) {
disp_tcall->async_call([this, conn](ThreadCall::Handle &) {
- conn->disp_terminate();
+ try {
+ conn->disp_terminate();
+ } catch (...) {
+ disp_error_cb(std::current_exception());
+ }
});
}
};
diff --git a/include/salticidae/event.h b/include/salticidae/event.h
index 91637e4..5a315dd 100644
--- a/include/salticidae/event.h
+++ b/include/salticidae/event.h
@@ -467,8 +467,14 @@ class MPSCQueueEventDriven: public MPSCQueue<T> {
// since all enqueue operations are finalized, the dequeue should be able
// to see those enqueued values in func()
wait_sig.exchange(true, std::memory_order_acq_rel);
- if (func(*this))
+ bool again;
+ try {
+ again = func(*this);
+ } catch (SalticidaeError &err) {
write(fd, &dummy, 8);
+ throw err;
+ }
+ if (again) write(fd, &dummy, 8);
});
ev.add(FdEvent::READ);
}
@@ -610,7 +616,12 @@ class ThreadCall {
Handle *h;
while (q.try_dequeue(h))
{
- h->exec();
+ try {
+ h->exec();
+ } catch (SalticidaeError &err) {
+ delete h;
+ throw err;
+ }
delete h;
if (++cnt == burst_size) return true;
}
diff --git a/include/salticidae/netaddr.h b/include/salticidae/netaddr.h
index 909e092..a25bf6f 100644
--- a/include/salticidae/netaddr.h
+++ b/include/salticidae/netaddr.h
@@ -53,7 +53,7 @@ struct NetAddr {
void set_by_ip_port(const std::string &_addr, uint16_t _port) {
struct hostent *h;
if ((h = gethostbyname(_addr.c_str())) == nullptr)
- throw SalticidaeError("gethostbyname failed");
+ throw SalticidaeError(SALTI_ERROR_NETADDR_INVALID, errno);
memmove(&ip, h->h_addr_list[0], sizeof(in_addr_t));
port = htons(_port);
}
@@ -61,19 +61,19 @@ struct NetAddr {
NetAddr(const std::string &ip_port_addr) {
size_t pos = ip_port_addr.find(":");
if (pos == std::string::npos)
- throw SalticidaeError("invalid port format");
+ throw SalticidaeError(SALTI_ERROR_NETADDR_INVALID);
std::string ip_str = ip_port_addr.substr(0, pos);
std::string port_str = ip_port_addr.substr(pos + 1);
long port;
try {
port = std::stol(port_str.c_str());
} catch (std::logic_error &) {
- throw SalticidaeError("invalid port format");
+ throw SalticidaeError(SALTI_ERROR_NETADDR_INVALID);
}
if (port < 0)
- throw SalticidaeError("negative port number");
+ throw SalticidaeError(SALTI_ERROR_NETADDR_INVALID);
if (port > 0xffff)
- throw SalticidaeError("port number greater than 0xffff");
+ throw SalticidaeError(SALTI_ERROR_NETADDR_INVALID);
set_by_ip_port(ip_str, (uint16_t)port);
}
/* construct from unix socket format */
diff --git a/include/salticidae/network.h b/include/salticidae/network.h
index e7e77f5..9c57749 100644
--- a/include/salticidae/network.h
+++ b/include/salticidae/network.h
@@ -199,8 +199,10 @@ class MsgNetwork: public ConnPool {
}
template<typename MsgType>
- bool send_msg(MsgType &&msg, const conn_t &conn);
- inline bool _send_msg(const Msg &msg, const conn_t &conn);
+ void send_msg(MsgType &&msg, const conn_t &conn);
+ inline void _send_msg(Msg &&msg, const conn_t &conn);
+ inline void _send_msg_dispatcher(const Msg &msg, const conn_t &conn);
+
using ConnPool::listen;
conn_t connect(const NetAddr &addr) {
return static_pointer_cast<Conn>(ConnPool::connect(addr));
@@ -248,10 +250,6 @@ class ClientNetwork: public MsgNetwork<OpcodeType> {
void send_msg(MsgType &&msg, const NetAddr &addr);
};
-class PeerNetworkError: public ConnPoolError {
- using ConnPoolError::ConnPoolError;
-};
-
/** Peer-to-peer network where any two nodes could hold a bi-diretional message
* channel, established by either side. */
template<typename OpcodeType = uint8_t,
@@ -434,9 +432,9 @@ class PeerNetwork: public MsgNetwork<OpcodeType> {
template<typename MsgType>
void send_msg(MsgType &&msg, const NetAddr &paddr);
inline void _send_msg(Msg &&msg, const NetAddr &paddr);
-
template<typename MsgType>
void multicast_msg(MsgType &&msg, const std::vector<NetAddr> &paddrs);
+ inline void _multicast_msg(Msg &&msg, const std::vector<NetAddr> &paddrs);
void listen(NetAddr listen_addr);
conn_t connect(const NetAddr &addr) = delete;
@@ -480,13 +478,23 @@ void MsgNetwork<OpcodeType>::Conn::on_read() {
template<typename OpcodeType>
template<typename MsgType>
-bool MsgNetwork<OpcodeType>::send_msg(MsgType &&_msg, const conn_t &conn) {
- Msg msg(std::forward<MsgType>(_msg));
- return _send_msg(msg, conn);
+void MsgNetwork<OpcodeType>::send_msg(MsgType &&msg, const conn_t &conn) {
+ return _send_msg(MsgType(std::move(msg)), conn);
+}
+
+template<typename OpcodeType>
+inline void MsgNetwork<OpcodeType>::_send_msg(Msg &&msg, const conn_t &conn) {
+ this->disp_tcall->async_call(
+ [this, msg=std::move(msg), conn](ThreadCall::Handle &) {
+ try {
+ this->_send_msg_dispatcher(msg, conn);
+ throw SalticidaeError("wow");
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
+ });
}
template<typename OpcodeType>
-inline bool MsgNetwork<OpcodeType>::_send_msg(const Msg &msg, const conn_t &conn) {
+inline void MsgNetwork<OpcodeType>::_send_msg_dispatcher(const Msg &msg, const conn_t &conn) {
bytearray_t msg_data = msg.serialize();
SALTICIDAE_LOG_DEBUG("wrote message %s to %s",
std::string(msg).c_str(),
@@ -495,17 +503,19 @@ inline bool MsgNetwork<OpcodeType>::_send_msg(const Msg &msg, const conn_t &conn
conn->nsent++;
conn->nsentb += msg.get_length();
#endif
- return conn->write(std::move(msg_data));
+ conn->write(std::move(msg_data));
}
template<typename O, O _, O __>
void PeerNetwork<O, _, __>::tcall_reset_timeout(ConnPool::Worker *worker,
const conn_t &conn, double timeout) {
- worker->get_tcall()->async_call([conn, t=timeout](ThreadCall::Handle &) {
- if (!conn->ev_timeout) return;
- conn->ev_timeout.del();
- conn->ev_timeout.add(t);
- SALTICIDAE_LOG_DEBUG("reset connection timeout %.2f", t);
+ worker->get_tcall()->async_call([worker, conn, t=timeout](ThreadCall::Handle &) {
+ try {
+ if (!conn->ev_timeout) return;
+ conn->ev_timeout.del();
+ conn->ev_timeout.add(t);
+ SALTICIDAE_LOG_DEBUG("reset connection timeout %.2f", t);
+ } catch (...) { worker->error_callback(std::current_exception()); }
});
}
@@ -517,9 +527,11 @@ void PeerNetwork<O, _, __>::Conn::on_setup() {
auto conn = static_pointer_cast<Conn>(this->self());
auto worker = this->worker;
assert(!ev_timeout);
- ev_timeout = TimerEvent(worker->get_ec(), [conn](TimerEvent &) {
- SALTICIDAE_LOG_INFO("peer ping-pong timeout");
- conn->worker_terminate();
+ ev_timeout = TimerEvent(worker->get_ec(), [worker, conn](TimerEvent &) {
+ try {
+ SALTICIDAE_LOG_INFO("peer ping-pong timeout");
+ conn->worker_terminate();
+ } catch (...) { worker->error_callback(std::current_exception()); }
});
/* the initial ping-pong to set up the connection */
tcall_reset_timeout(worker, conn, pn->conn_timeout);
@@ -541,7 +553,9 @@ void PeerNetwork<O, _, __>::Conn::on_teardown() {
// try to reconnect
p->ev_retry_timer = TimerEvent(pn->disp_ec,
[pn, peer_id = this->peer_id](TimerEvent &) {
- pn->start_active_conn(peer_id);
+ try {
+ pn->start_active_conn(peer_id);
+ } catch (...) { pn->disp_error_cb(std::current_exception()); }
});
p->ev_retry_timer.add(pn->gen_conn_timeout());
}
@@ -647,11 +661,13 @@ template<typename O, O _, O __>
void PeerNetwork<O, _, __>::msg_ping(MsgPing &&msg, const conn_t &conn) {
uint16_t port = msg.port;
this->disp_tcall->async_call([this, conn, port](ThreadCall::Handle &) {
- if (conn->get_mode() == ConnPool::Conn::DEAD) return;
- SALTICIDAE_LOG_INFO("ping from %s, port %u",
- std::string(*conn).c_str(), ntohs(port));
- if (check_new_conn(conn, port)) return;
- send_msg(MsgPong(this->listen_port), conn);
+ try {
+ if (conn->get_mode() == ConnPool::Conn::DEAD) return;
+ SALTICIDAE_LOG_INFO("ping from %s, port %u",
+ std::string(*conn).c_str(), ntohs(port));
+ if (check_new_conn(conn, port)) return;
+ send_msg(MsgPong(this->listen_port), conn);
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
@@ -659,54 +675,74 @@ template<typename O, O _, O __>
void PeerNetwork<O, _, __>::msg_pong(MsgPong &&msg, const conn_t &conn) {
uint16_t port = msg.port;
this->disp_tcall->async_call([this, conn, port](ThreadCall::Handle &) {
- if (conn->get_mode() == ConnPool::Conn::DEAD) return;
- auto it = id2peer.find(conn->peer_id);
- if (it == id2peer.end())
- {
- SALTICIDAE_LOG_WARN("pong message discarded");
- return;
- }
- if (check_new_conn(conn, port)) return;
- auto p = it->second.get();
- p->pong_msg_ok = true;
- if (p->ping_timer_ok)
- {
- p->reset_ping_timer();
- p->send_ping();
- }
+ try {
+ if (conn->get_mode() == ConnPool::Conn::DEAD) return;
+ auto it = id2peer.find(conn->peer_id);
+ if (it == id2peer.end())
+ {
+ SALTICIDAE_LOG_WARN("pong message discarded");
+ return;
+ }
+ if (check_new_conn(conn, port)) return;
+ auto p = it->second.get();
+ p->pong_msg_ok = true;
+ if (p->ping_timer_ok)
+ {
+ p->reset_ping_timer();
+ p->send_ping();
+ }
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
template<typename O, O _, O __>
void PeerNetwork<O, _, __>::listen(NetAddr listen_addr) {
- this->disp_tcall->call([this, listen_addr](ThreadCall::Handle &) {
- MsgNet::_listen(listen_addr);
- listen_port = listen_addr.port;
- });
+ auto ret = *(static_cast<SalticidaeError *>(
+ this->disp_tcall->call([this, listen_addr](ThreadCall::Handle &h) {
+ SalticidaeError err;
+ try {
+ MsgNet::_listen(listen_addr);
+ listen_port = listen_addr.port;
+ } catch (SalticidaeError &e) {
+ err = e;
+ }
+ h.set_result(std::move(err));
+ }).get()));
+ if (ret.get_code()) throw ret;
}
template<typename O, O _, O __>
void PeerNetwork<O, _, __>::add_peer(const NetAddr &addr) {
- this->disp_tcall->call([this, addr](ThreadCall::Handle &) {
- auto it = id2peer.find(addr);
- if (it != id2peer.end())
- throw PeerNetworkError("peer already exists");
- id2peer.insert(std::make_pair(addr, new Peer(addr, nullptr, this->disp_ec)));
- start_active_conn(addr);
+ this->disp_tcall->async_call([this, addr](ThreadCall::Handle &) {
+ try {
+ auto it = id2peer.find(addr);
+ if (it != id2peer.end())
+ throw PeerNetworkError(SALTI_ERROR_PEER_ALREADY_EXISTS);
+ id2peer.insert(std::make_pair(addr, new Peer(addr, nullptr, this->disp_ec)));
+ start_active_conn(addr);
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
template<typename O, O _, O __>
const typename PeerNetwork<O, _, __>::conn_t
PeerNetwork<O, _, __>::get_peer_conn(const NetAddr &paddr) const {
- auto ret = *(static_cast<conn_t *>(this->disp_tcall->call(
- [this, paddr](ThreadCall::Handle &h) {
- auto it = id2peer.find(paddr);
- if (it == id2peer.end())
- throw PeerNetworkError("peer does not exist");
- h.set_result(it->second->conn);
+ auto ret = *(static_cast<std::pair<conn_t, SalticidaeError> *>(
+ this->disp_tcall->call([this, paddr](ThreadCall::Handle &h) {
+ conn_t conn;
+ SalticidaeError err;
+ try {
+ auto it = id2peer.find(paddr);
+ if (it == id2peer.end())
+ throw PeerNetworkError(SALTI_ERROR_PEER_NOT_EXISTS);
+ conn = it->second->conn;
+ } catch (SalticidaeError &e) {
+ err = e;
+ }
+ h.set_result(std::make_pair(std::move(conn), std::move(err)));
}).get()));
- return std::move(ret);
+ if (ret.second.get_code()) throw ret.second;
+ return std::move(ret.first);
}
template<typename O, O _, O __>
@@ -720,32 +756,41 @@ bool PeerNetwork<O, _, __>::has_peer(const NetAddr &paddr) const {
template<typename O, O _, O __>
template<typename MsgType>
void PeerNetwork<O, _, __>::send_msg(MsgType &&msg, const NetAddr &paddr) {
- return _send_msg(MsgType(msg), paddr);
+ return _send_msg(MsgType(std::move(msg)), paddr);
}
template<typename O, O _, O __>
void PeerNetwork<O, _, __>::_send_msg(Msg &&msg, const NetAddr &paddr) {
this->disp_tcall->async_call(
[this, msg=std::move(msg), paddr](ThreadCall::Handle &) {
- auto it = id2peer.find(paddr);
- if (it == id2peer.end())
- throw PeerNetworkError("peer does not exist");
- send_msg(std::move(msg), it->second->conn);
+ try {
+ auto it = id2peer.find(paddr);
+ if (it == id2peer.end())
+ throw PeerNetworkError(SALTI_ERROR_PEER_NOT_EXISTS);
+ this->_send_msg_dispatcher(msg, it->second->conn);
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
template<typename O, O _, O __>
template<typename MsgType>
void PeerNetwork<O, _, __>::multicast_msg(MsgType &&msg, const std::vector<NetAddr> &paddrs) {
+ return _multicast_msg(MsgType(std::move(msg)), paddrs);
+}
+
+template<typename O, O _, O __>
+void PeerNetwork<O, _, __>::_multicast_msg(Msg &&msg, const std::vector<NetAddr> &paddrs) {
this->disp_tcall->async_call(
[this, msg=std::move(msg), paddrs](ThreadCall::Handle &) {
- for (auto &addr: paddrs)
- {
- auto it = id2peer.find(addr);
- if (it == id2peer.end())
- throw PeerNetworkError("peer does not exist");
- send_msg(std::move(msg), it->second->conn);
- }
+ try {
+ for (auto &addr: paddrs)
+ {
+ auto it = id2peer.find(addr);
+ if (it == id2peer.end())
+ throw PeerNetworkError(SALTI_ERROR_PEER_NOT_EXISTS);
+ this->_send_msg_dispatcher(msg, it->second->conn);
+ }
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
@@ -774,9 +819,11 @@ template<typename MsgType>
void ClientNetwork<OpcodeType>::send_msg(MsgType &&msg, const NetAddr &addr) {
this->disp_tcall->async_call(
[this, addr, msg=std::forward<MsgType>(msg)](ThreadCall::Handle &) {
- auto it = addr2conn.find(addr);
- if (it != addr2conn.end())
- send_msg(std::move(msg), it->second);
+ try {
+ auto it = addr2conn.find(addr);
+ if (it != addr2conn.end())
+ send_msg(std::move(msg), it->second);
+ } catch (...) { this->disp_error_cb(std::current_exception()); }
});
}
@@ -842,7 +889,7 @@ void msgnetwork_config_queue_capacity(msgnetwork_config_t *self, size_t cap);
msgnetwork_t *msgnetwork_new(const eventcontext_t *ec, const msgnetwork_config_t *config);
void msgnetwork_free(const msgnetwork_t *self);
-bool msgnetwork_send_msg(msgnetwork_t *self, const msg_t *msg, const msgnetwork_conn_t *conn);
+void msgnetwork_send_msg_by_move(msgnetwork_t *self, msg_t *_moved_msg, const msgnetwork_conn_t *conn);
msgnetwork_conn_t *msgnetwork_connect(msgnetwork_t *self, const netaddr_t *addr);
msgnetwork_conn_t *msgnetwork_conn_copy(const msgnetwork_conn_t *self);
void msgnetwork_conn_free(const msgnetwork_conn_t *self);
diff --git a/include/salticidae/queue.h b/include/salticidae/queue.h
index da11e8a..9fd11f1 100644
--- a/include/salticidae/queue.h
+++ b/include/salticidae/queue.h
@@ -17,6 +17,7 @@ class FreeList {
std::atomic<size_t> refcnt;
std::atomic<bool> freed;
Node(): next(nullptr), refcnt(1), freed(false) {}
+ virtual ~Node() {}
};
private:
diff --git a/include/salticidae/util.h b/include/salticidae/util.h
index 19779b0..0ddf5be 100644
--- a/include/salticidae/util.h
+++ b/include/salticidae/util.h
@@ -30,6 +30,7 @@
#include <string>
#include <exception>
#include <cstdarg>
+#include <cstring>
#include <vector>
#include <unordered_map>
#include <functional>
@@ -52,18 +53,57 @@ std::vector<std::string> trim_all(const std::vector<std::string> &ss);
std::string vstringprintf(const char *fmt, va_list ap);
std::string stringprintf(const char *fmt, ...);
+enum SalticidaeErrorCode {
+ SALTI_NORMAL,
+ SALTI_ERROR_GENERIC,
+ SALTI_ERROR_ACCEPT,
+ SALTI_ERROR_LISTEN,
+ SALTI_ERROR_CONNECT,
+ SALTI_ERROR_PEER_ALREADY_EXISTS,
+ SALTI_ERROR_PEER_NOT_EXISTS,
+ SALTI_ERROR_NETADDR_INVALID,
+ SALTI_ERROR_OPTVAL_INVALID,
+ SALTI_ERROR_OPTNAME_ALREADY_EXISTS,
+ SALTI_ERROR_OPT_UNKNOWN_ACTION,
+ SALTI_ERROR_CONFIG_LINE_TOO_LONG,
+ SALTI_ERROR_OPT_INVALID
+};
+
+extern const char *SALTICIDAE_ERROR_STRINGS[];
+
class SalticidaeError: public std::exception {
std::string msg;
+ int code;
+ int oscode;
+
public:
- SalticidaeError();
+ SalticidaeError() : code(SALTI_NORMAL) {}
template<typename... Args>
- SalticidaeError(const std::string &fmt, Args... args) {
+ SalticidaeError(const std::string &fmt, Args... args): code(SALTI_ERROR_GENERIC) {
msg = stringprintf(fmt.c_str(), args...);
}
+ SalticidaeError(int code, int oscode = 0): code(code), oscode(oscode) {
+ if (oscode)
+ msg = stringprintf("%s: %s", SALTICIDAE_ERROR_STRINGS[code], strerror(oscode));
+ else
+ msg = SALTICIDAE_ERROR_STRINGS[code];
+ }
+
operator std::string() const { return msg; }
const char *what() const throw() override { return msg.c_str(); }
+ int get_code() const { return code; }
+ int get_oscode() const { return oscode; }
+};
+
+
+struct ConnPoolError: public SalticidaeError {
+ using SalticidaeError::SalticidaeError;
+};
+
+class PeerNetworkError: public ConnPoolError {
+ using ConnPoolError::ConnPoolError;
};
extern const char *TTY_COLOR_RED;
@@ -217,7 +257,7 @@ class Config {
try {
val = stoi(strval, &idx);
} catch (std::invalid_argument &) {
- throw SalticidaeError("invalid integer");
+ throw SalticidaeError(SALTI_ERROR_OPTVAL_INVALID);
}
}
int &get() { return val; }
@@ -237,7 +277,7 @@ class Config {
try {
val = stod(strval, &idx);
} catch (std::invalid_argument &) {
- throw SalticidaeError("invalid double");
+ throw SalticidaeError(SALTI_ERROR_OPTVAL_INVALID);
}
}
double &get() { return val; }