From 7387f7f6b615717dd863bcb62ee7f65ace29879b Mon Sep 17 00:00:00 2001 From: Determinant Date: Mon, 12 Nov 2018 15:52:38 -0500 Subject: update PeerNetwork to work with multiloops --- include/salticidae/conn.h | 75 +++++------- include/salticidae/network.h | 278 +++++++++++++++++++++++++------------------ include/salticidae/type.h | 3 + 3 files changed, 194 insertions(+), 162 deletions(-) (limited to 'include') diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h index 26d19fe..1364d4d 100644 --- a/include/salticidae/conn.h +++ b/include/salticidae/conn.h @@ -61,7 +61,7 @@ class ConnPool { /** The handle to a bi-directional connection. */ using conn_t = ArcObj; /** The type of callback invoked when connection status is changed. */ - using conn_callback_t = std::function; + using conn_callback_t = std::function; /** Abstraction for a bi-directional connection. */ class Conn { @@ -72,7 +72,7 @@ class ConnPool { PASSIVE, /**< the connection is established by accept() */ }; - private: + protected: size_t seg_buff_size; conn_t self_ref; int fd; @@ -109,14 +109,10 @@ class ConnPool { /** Get the handle to itself. */ conn_t self() { return self_ref; } operator std::string() const; - int get_fd() const { return fd; } const NetAddr &get_addr() const { return addr; } ConnMode get_mode() const { return mode; } ConnPool *get_pool() const { return cpool; } - SegBuffer &get_recv_buffer() { return recv_buffer; } MPSCWriteBuffer &get_send_buffer() { return send_buffer; } - /** Set the buffer size used for send/receive data. */ - void set_seg_buff_size(size_t size) { seg_buff_size = size; } /** Write data to the connection (non-blocking). The data will be sent * whenever I/O is available. */ @@ -124,11 +120,6 @@ class ConnPool { send_buffer.push(std::move(data)); } - ///** Move the send buffer from the other (old) connection. */ - //void move_send_buffer(conn_t other) { - // send_buffer = std::move(other->send_buffer); - //} - protected: /** Close the IO and clear all on-going or planned events. */ virtual void on_close() { @@ -143,13 +134,9 @@ class ConnPool { /** Called when new data is available. */ virtual void on_read() {} /** Called when the underlying connection is established. */ - virtual void on_setup() { - cpool->update_conn(self()); - } + virtual void on_setup() {} /** Called when the underlying connection breaks. */ - virtual void on_teardown() { - cpool->update_conn(self()); - } + virtual void on_teardown() {} }; private: @@ -165,14 +152,13 @@ class ConnPool { /* owned by the dispatcher */ std::unordered_map pool; int listen_fd; /**< for accepting new network connections */ - int dlisten_fd[2]; /**< for control command sent to the dispatcher */ Event ev_listen; Event ev_dlisten; std::mutex cp_mlock; - void update_conn(const conn_t &conn) { - auto ptr = new conn_t(conn); - write(mlisten_fd[1], &ptr, sizeof(ptr)); + void update_conn(const conn_t &conn, bool connected) { + auto dcmd = new UserConn(conn, connected); + write(mlisten_fd[1], &dcmd, sizeof(dcmd)); } struct Worker; @@ -269,31 +255,22 @@ class ConnPool { void accept_client(evutil_socket_t, short); conn_t add_conn(const conn_t &conn); conn_t _connect(const NetAddr &addr); - void _listen(NetAddr listen_addr); void _post_terminate(int fd); + protected: class DispatchCmd { public: virtual ~DispatchCmd() = default; virtual void exec(ConnPool *cpool) = 0; }; - // TODO: the following two are untested - class DspListen: public DispatchCmd { - const NetAddr addr; - public: - DspListen(const NetAddr &addr): addr(addr) {} - void exec(ConnPool *cpool) override { - cpool->_listen(addr); - } - }; - + private: class DspConnect: public DispatchCmd { const NetAddr addr; public: DspConnect(const NetAddr &addr): addr(addr) {} void exec(ConnPool *cpool) override { - cpool->update_conn(cpool->_connect(addr)); + cpool->update_conn(cpool->_connect(addr), true); } }; @@ -318,6 +295,18 @@ class ConnPool { } }; + class UserConn: public DispatchCmd { + conn_t conn; + bool connected; + public: + UserConn(const conn_t &conn, bool connected): + conn(conn), connected(connected) {} + void exec(ConnPool *cpool) override { + if (cpool->conn_cb) + cpool->conn_cb(*conn, connected); + } + }; + void post_terminate(int fd) { auto dcmd = new DspPostTerm(fd); write(dlisten_fd[1], &dcmd, sizeof(dcmd)); @@ -330,6 +319,7 @@ class ConnPool { protected: EventContext ec; EventContext dispatcher_ec; + int dlisten_fd[2]; /**< for control command sent to the dispatcher */ std::mutex dsp_ec_mlock; /** Should be implemented by derived class to return a new Conn object. */ virtual Conn *create_conn() = 0; @@ -352,11 +342,10 @@ class ConnPool { throw ConnPoolError(std::string("failed to create dispatcher pipe")); ev_mlisten = Event(ec, mlisten_fd[0], EV_READ | EV_PERSIST, [this](int fd, short) { - conn_t *conn_ptr; - read(fd, &conn_ptr, sizeof(conn_ptr)); - if (conn_cb) - conn_cb(**conn_ptr); - delete conn_ptr; + DispatchCmd *dcmd; + read(fd, &dcmd, sizeof(dcmd)); + dcmd->exec(this); + delete dcmd; }); ev_mlisten.add(); @@ -414,15 +403,7 @@ class ConnPool { /** Listen for passive connections (connection initiated from remote). * Does not need to be called if do not want to accept any passive * connections. */ - void listen(NetAddr listen_addr, bool blocking = true) { - if (blocking) - _listen(listen_addr); - else - { - auto dcmd = new DspListen(listen_addr); - write(dlisten_fd[1], &dcmd, sizeof(dcmd)); - } - } + void listen(NetAddr listen_addr); template void reg_conn_handler(Func cb) { conn_cb = cb; } diff --git a/include/salticidae/network.h b/include/salticidae/network.h index d82772f..8f8940b 100644 --- a/include/salticidae/network.h +++ b/include/salticidae/network.h @@ -77,8 +77,8 @@ class MsgNetwork: public ConnPool { protected: #ifdef SALTICIDAE_MSG_STAT - mutable size_t nsent; - mutable size_t nrecv; + mutable std::atomic nsent; + mutable std::atomic nrecv; #endif public: @@ -107,7 +107,8 @@ class MsgNetwork: public ConnPool { #ifdef SALTICIDAE_MSG_STAT class msg_stat_by_opcode_t: public std::unordered_map> { + std::pair, + std::atomic>> { public: void add(const Msg &msg) { auto &p = this->operator[](msg.get_opcode()); @@ -259,6 +260,7 @@ class PeerNetwork: public MsgNetwork { friend PeerNetwork; NetAddr peer_id; Event ev_timeout; + void reset_timeout(double timeout); public: Conn() = default; @@ -292,6 +294,7 @@ class PeerNetwork: public MsgNetwork { bool ping_timer_ok; bool pong_msg_ok; bool connected; + std::mutex mlock; Peer() = delete; Peer(NetAddr addr, conn_t conn, const EventContext &ec): @@ -314,7 +317,7 @@ class PeerNetwork: public MsgNetwork { }; std::unordered_map > id2peer; - std::vector peer_list; + std::mutex pn_mlock; IdentityMode id_mode; double retry_conn_delay; @@ -350,7 +353,6 @@ class PeerNetwork: public MsgNetwork { void msg_ping(MsgPing &&msg, Conn &conn); void msg_pong(MsgPong &&msg, Conn &conn); - void reset_conn_timeout(Conn &conn); bool check_new_conn(Conn &conn, uint16_t port); void start_active_conn(const NetAddr &paddr); @@ -380,12 +382,11 @@ class PeerNetwork: public MsgNetwork { void add_peer(const NetAddr &paddr); const conn_t get_peer_conn(const NetAddr &paddr) const; template - void send_msg(const MsgType &msg, const Peer *peer); + void _send_msg(const MsgType &msg, const Peer *peer); template void send_msg(const MsgType &msg, const NetAddr &paddr); void listen(NetAddr listen_addr); bool has_peer(const NetAddr &paddr) const; - const std::vector &all_peers() const; conn_t connect(const NetAddr &addr) { return static_pointer_cast(ConnPool::connect(addr)); } @@ -395,9 +396,9 @@ class PeerNetwork: public MsgNetwork { template void MsgNetwork::Conn::on_read() { ConnPool::Conn::on_read(); - auto &recv_buffer = get_recv_buffer(); + auto &recv_buffer = this->recv_buffer; auto mn = get_net(); - while (get_fd() != -1) + while (fd != -1) { if (msg_state == Conn::HEADER) { @@ -426,23 +427,22 @@ void MsgNetwork::Conn::on_read() { } } -template -void PeerNetwork::Peer::reset_conn(conn_t new_conn) { - if (conn != new_conn) - { - if (conn) - { - //SALTICIDAE_LOG_DEBUG("moving send buffer"); - //new_conn->move_send_buffer(conn); - SALTICIDAE_LOG_INFO("terminating old connection %s", std::string(*conn).c_str()); - conn->terminate(); - } - addr = new_conn->get_addr(); - conn = new_conn; - } - clear_all_events(); +template +template +void MsgNetwork::send_msg(const MsgType &_msg, Conn &conn) { + Msg msg(_msg); + bytearray_t msg_data = msg.serialize(); + SALTICIDAE_LOG_DEBUG("wrote message %s to %s", + std::string(msg).c_str(), + std::string(conn).c_str()); + conn.write(std::move(msg_data)); +#ifdef SALTICIDAE_MSG_STAT + conn.nsent++; + sent_by_opcode.add(msg); +#endif } +/* begin: functions invoked by the dispatcher */ template void PeerNetwork::Conn::on_setup() { MsgNet::Conn::on_setup(); @@ -452,9 +452,14 @@ void PeerNetwork::Conn::on_setup() { SALTICIDAE_LOG_INFO("peer ping-pong timeout"); this->terminate(); }); + if (this->get_mode() == Conn::ConnMode::ACTIVE) + { + peer_id = this->get_addr(); + if (id_mode == IP_BASED) peer_id.port = 0; + } /* the initial ping-pong to set up the connection */ auto &conn = static_cast(*this); - pn->reset_conn_timeout(conn); + conn->reset_timeout(pn->conn_timeout); pn->MsgNet::send_msg(MsgPing(pn->listen_port), conn); } @@ -462,9 +467,11 @@ template void PeerNetwork::Conn::on_teardown() { MsgNet::Conn::on_teardown(); auto pn = get_net(); + mutex_lg_t _pn_lg(pn->pn_mlock); auto it = pn->id2peer.find(peer_id); if (it == pn->id2peer.end()) return; auto p = it->second.get(); + mutex_lg_t _p_lg(p->mlock); if (this != p->conn.get()) return; p->ev_ping_timer.del(); p->connected = false; @@ -472,13 +479,68 @@ void PeerNetwork::Conn::on_teardown() { SALTICIDAE_LOG_INFO("connection lost %s for %s", std::string(*this).c_str(), std::string(peer_id).c_str()); - p->ev_retry_timer = Event(pn->ec, -1, 0, + // try to reconnect + p->ev_retry_timer = Event(pn->dispatcher_ec, -1, 0, [pn, peer_id = this->peer_id](evutil_socket_t, short) { + mutex_lg_t _pn_lg(pn->pn_mlock); pn->start_active_conn(peer_id); }); p->ev_retry_timer.add_with_timeout(pn->gen_conn_timeout()); } +template +void PeerNetwork::Peer::reset_conn(conn_t new_conn) { + if (conn != new_conn) + { + if (conn) + { + //SALTICIDAE_LOG_DEBUG("moving send buffer"); + //new_conn->move_send_buffer(conn); + SALTICIDAE_LOG_INFO("terminating old connection %s", std::string(*conn).c_str()); + conn->terminate(); + } + addr = new_conn->get_addr(); + conn = new_conn; + } + clear_all_events(); +} + +template +void PeerNetwork::Conn::reset_timeout(double timeout) { + assert(ev_timeout); + ev_timeout.del(); + ev_timeout.add_with_timeout(timeout); + SALTICIDAE_LOG_INFO("reset timeout %.2f", timeout); +} + +template +void PeerNetwork::Peer::reset_ping_timer() { + assert(ev_ping_timer); + ev_ping_timer.del(); + ev_ping_timer.add_with_timeout( + gen_rand_timeout(conn->get_net()->ping_period)); +} + +template +void PeerNetwork::Peer::send_ping() { + auto pn = conn->get_net(); + ping_timer_ok = false; + pong_msg_ok = false; + conn->reset_timeout(pn->conn_timeout); + pn->_send_msg(MsgPing(pn->listen_port), this); +} + +template +void PeerNetwork::Peer::ping_timer(evutil_socket_t, short) { + mutex_lg_t _p_lg(mlock); + ping_timer_ok = true; + if (pong_msg_ok) + { + reset_ping_timer(); + send_ping(); + } +} + template bool PeerNetwork::check_new_conn(Conn &conn, uint16_t port) { if (conn.peer_id.is_null()) @@ -489,6 +551,7 @@ bool PeerNetwork::check_new_conn(Conn &conn, uint16_t port) { conn.peer_id.port = port; } auto p = id2peer.find(conn.peer_id)->second.get(); + mutex_lg_t _p_lg(p->mlock); if (p->connected) { if (conn.self() != p->conn) @@ -508,32 +571,78 @@ bool PeerNetwork::check_new_conn(Conn &conn, uint16_t port) { return false; } +template +class PeerNetworkPing: public ConnPool::DispatchCmd { + using conn_t = typename PeerNetwork::conn_t; + conn_t conn; + uint16_t port; + public: + PeerNetworkPing(const conn_t &conn, uint16_t port): + conn(conn), port(port) {} + void exec(ConnPool *cpool) override { + auto pn = static_cast *>(cpool); + mutex_lg_t _pn_lg(pn->pn_mlock); + if (pn->check_new_conn(conn, port)) return; + auto p = pn->id2peer.find(conn.peer_id)->second.get(); + mutex_lg_t _p_lg(p->mlock); + pn->_send_msg(MsgPong(this->listen_port), p); + } +}; + +template +class PeerNetworkPong: public ConnPool::DispatchCmd { + using conn_t = typename PeerNetwork::conn_t; + conn_t conn; + uint16_t port; + public: + PeerNetworkPong(const conn_t &conn, uint16_t port): + conn(conn), port(port) {} + void exec(ConnPool *cpool) override { + auto pn = static_cast *>(cpool); + mutex_lg_t _pn_lg(pn->pn_mlock); + auto it = pn->id2peer.find(conn->peer_id); + if (it == pn->id2peer.end()) + { + SALTICIDAE_LOG_WARN("pong message discarded"); + return; + } + if (pn->check_new_conn(conn, port)) return; + auto p = it->second.get(); + mutex_lg_t _p_lg(p->mlock); + p->pong_msg_ok = true; + if (p->ping_timer_ok) + { + p->reset_ping_timer(); + p->send_ping(); + } + } +}; +/* end: functions invoked by the dispatcher */ + +/* this function could be both invoked by the dispatcher and the user loop */ +template +void PeerNetwork::start_active_conn(const NetAddr &addr) { + auto p = id2peer.find(addr)->second.get(); + mutex_lg_t _p_lg(p->mlock); + if (p->connected) return; + auto conn = static_pointer_cast(connect(addr)); + assert(p->conn == nullptr); + p->conn = conn; +} + +/* begin: functions invoked by the user loop */ template void PeerNetwork::msg_ping(MsgPing &&msg, Conn &conn) { uint16_t port = msg.port; SALTICIDAE_LOG_INFO("ping from %s, port %u", std::string(conn).c_str(), ntohs(port)); - if (check_new_conn(conn, port)) return; - auto p = id2peer.find(conn.peer_id)->second.get(); - send_msg(MsgPong(this->listen_port), p); + auto dcmd = new PeerNetworkPing(conn.self(), port); + write(this->dlisten_fd[1], &dcmd, sizeof(dcmd)); } template void PeerNetwork::msg_pong(MsgPong &&msg, Conn &conn) { - auto it = id2peer.find(conn.peer_id); - if (it == id2peer.end()) - { - SALTICIDAE_LOG_WARN("pong message discarded"); - return; - } - auto p = it->second.get(); - uint16_t port = msg.port; - if (check_new_conn(conn, port)) return; - p->pong_msg_ok = true; - if (p->ping_timer_ok) - { - p->reset_ping_timer(); - p->send_ping(); - } + auto dcmd = new PeerNetworkPong(conn.self(), msg.port); + write(this->dlisten_fd[1], &dcmd, sizeof(dcmd)); } template @@ -544,31 +653,20 @@ void PeerNetwork::listen(NetAddr listen_addr) { this->reg_handler(generic_bind(&PeerNetwork::msg_pong, this, _1, _2)); } -template -void PeerNetwork::start_active_conn(const NetAddr &addr) { - auto p = id2peer.find(addr)->second.get(); - if (p->connected) return; - auto conn = static_pointer_cast(connect(addr)); - assert(p->conn == nullptr); - p->conn = conn; - conn->peer_id = addr; - if (id_mode == IP_BASED) - conn->peer_id.port = 0; -} - template void PeerNetwork::add_peer(const NetAddr &addr) { + mutex_lg_t _pn_lg(pn_mlock); 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->ec))); - peer_list.push_back(addr); + id2peer.insert(std::make_pair(addr, new Peer(addr, nullptr, this->dispatcher_ec))); start_active_conn(addr); } template const typename PeerNetwork::conn_t PeerNetwork::get_peer_conn(const NetAddr &paddr) const { + mutex_lg_t _pn_lg(pn_mlock); auto it = id2peer.find(paddr); if (it == id2peer.end()) throw PeerNetworkError("peer does not exist"); @@ -577,27 +675,13 @@ PeerNetwork::get_peer_conn(const NetAddr &paddr) const { template bool PeerNetwork::has_peer(const NetAddr &paddr) const { + mutex_lg_t _pn_lg(pn_mlock); return id2peer.count(paddr); } -template -template -void MsgNetwork::send_msg(const MsgType &_msg, Conn &conn) { - Msg msg(_msg); - bytearray_t msg_data = msg.serialize(); - SALTICIDAE_LOG_DEBUG("wrote message %s to %s", - std::string(msg).c_str(), - std::string(conn).c_str()); - conn.write(std::move(msg_data)); -#ifdef SALTICIDAE_MSG_STAT - conn.nsent++; - sent_by_opcode.add(msg); -#endif -} - template template -void PeerNetwork::send_msg(const MsgType &msg, const Peer *peer) { +void PeerNetwork::_send_msg(const MsgType &msg, const Peer *peer) { if (peer->connected) MsgNet::send_msg(msg, *(peer->conn)); else @@ -607,6 +691,7 @@ void PeerNetwork::send_msg(const MsgType &msg, const Peer *peer) { template template void PeerNetwork::send_msg(const MsgType &msg, const NetAddr &addr) { + mutex_lg_t _pn_lg(pn_mlock); auto it = id2peer.find(addr); if (it == id2peer.end()) { @@ -614,48 +699,11 @@ void PeerNetwork::send_msg(const MsgType &msg, const NetAddr &addr) { std::string(addr).c_str()); throw PeerNetworkError("peer does not exist"); } - send_msg(msg, it->second.get()); -} - -template -void PeerNetwork::Peer::reset_ping_timer() { - assert(ev_ping_timer); - ev_ping_timer.del(); - ev_ping_timer.add_with_timeout( - gen_rand_timeout(conn->get_net()->ping_period)); -} - -template -void PeerNetwork::reset_conn_timeout(Conn &conn) { - assert(conn.ev_timeout); - conn.ev_timeout.del(); - conn.ev_timeout.add_with_timeout(conn_timeout); - SALTICIDAE_LOG_INFO("reset timeout %.2f", conn_timeout); -} - -template -void PeerNetwork::Peer::send_ping() { - auto pn = conn->get_net(); - ping_timer_ok = false; - pong_msg_ok = false; - pn->reset_conn_timeout(*conn); - pn->send_msg(MsgPing(pn->listen_port), this); -} - -template -void PeerNetwork::Peer::ping_timer(evutil_socket_t, short) { - ping_timer_ok = true; - if (pong_msg_ok) - { - reset_ping_timer(); - send_ping(); - } -} - -template -const std::vector &PeerNetwork::all_peers() const { - return peer_list; + auto p = it->second.get(); + mutex_lg_t _p_lg(p->mlock); + _send_msg(msg, p); } +/* end: functions invoked by the user loop */ template void ClientNetwork::Conn::on_setup() { diff --git a/include/salticidae/type.h b/include/salticidae/type.h index c454265..68deeb0 100644 --- a/include/salticidae/type.h +++ b/include/salticidae/type.h @@ -33,6 +33,7 @@ #include #include #include +#include namespace salticidae { @@ -40,6 +41,8 @@ const auto _1 = std::placeholders::_1; const auto _2 = std::placeholders::_2; using bytearray_t = std::vector; +using mutex_lg_t = std::lock_guard; +using mutex_ul_t = std::unique_lock; template T htole(T) = delete; template<> inline uint16_t htole(uint16_t x) { return htole16(x); } -- cgit v1.2.3-70-g09d2