From 82bd1d81e918cc8a1a46b6c36a81cbba43750bb1 Mon Sep 17 00:00:00 2001 From: Determinant Date: Thu, 15 Nov 2018 00:24:38 -0500 Subject: use a simple worker selection policy; add todo-list --- TODO.rst | 2 ++ include/salticidae/conn.h | 64 +++++++++++++++++++++----------------------- include/salticidae/network.h | 6 ++--- src/conn.cpp | 31 ++++++++++++++++----- test/test_p2p.cpp | 7 ++--- 5 files changed, 65 insertions(+), 45 deletions(-) create mode 100644 TODO.rst diff --git a/TODO.rst b/TODO.rst new file mode 100644 index 0000000..b6fb085 --- /dev/null +++ b/TODO.rst @@ -0,0 +1,2 @@ +- use config factory class to configure ConnPool, MsgNetwork, PeerNetwork, etc. +- stress test for PeerNetwork diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h index 480809f..5f5fef4 100644 --- a/include/salticidae/conn.h +++ b/include/salticidae/conn.h @@ -94,8 +94,10 @@ class ConnPool { void send_data(int, int); void conn_server(int, int); - /** Terminate the connection. */ - void terminate(); + /** Terminate the connection (from the worker thread). */ + void worker_terminate(); + /** Terminate the connection (from the dispatcher thread). */ + void disp_terminate(bool blocking = true); public: Conn(): ready_send(false) {} @@ -121,16 +123,9 @@ class ConnPool { } protected: - /** Close the IO and clear all on-going or planned events. */ - virtual void stop() { - if (!self_ref) return; - ev_connect.clear(); - ev_socket.clear(); - send_buffer.get_queue().unreg_handler(); - ::close(fd); - self_ref = nullptr; /* remove the self-cycle */ - } - + /** Close the IO and clear all on-going or planned events. Remove the + * connection from a Worker. */ + virtual void stop(); /** Called when new data is available. */ virtual void on_read() {} /** Called when the underlying connection is established. */ @@ -153,7 +148,7 @@ class ConnPool { /* owned by user loop */ BoxObj user_tcall; - conn_callback_t conn_cb; + conn_callback_t conn_cb; /* owned by the dispatcher */ Event ev_listen; @@ -170,9 +165,11 @@ class ConnPool { EventContext ec; ThreadCall tcall; std::thread handle; + bool disp_flag; + std::atomic nconn; public: - Worker(): tcall(ec) {} + Worker(): tcall(ec), disp_flag(false), nconn(0) {} /* the following functions are called by the dispatcher */ void start() { @@ -203,15 +200,13 @@ class ConnPool { else conn->send_data(fd, what); }); - - // std::bind(&Conn::recv_data, conn_ptr, _1, _2)); - //conn->ev_write = Event(ec, client_fd, Event::WRITE, - // std::bind(&Conn::send_data, conn_ptr, _1, _2)); conn->ev_socket.add(Event::READ | Event::WRITE); - //conn->ev_write.add(); + nconn++; }); } + void unfeed() { nconn--; } + void stop() { tcall.call([this](ThreadCall::Handle &) { ec.stop(); }); } @@ -219,6 +214,9 @@ class ConnPool { std::thread &get_handle() { return handle; } const EventContext &get_ec() { return ec; } ThreadCall *get_tcall() { return &tcall; } + void set_dispatcher() { disp_flag = true; } + bool is_dispatcher() const { return disp_flag; } + size_t get_nconn() { return nconn; } }; /* related to workers */ @@ -248,7 +246,18 @@ class ConnPool { //}; Worker &select_worker() { - return workers[1]; + size_t idx = 0; + size_t best = workers[idx].get_nconn(); + for (size_t i = 0; i < nworker; i++) + { + size_t t = workers[i].get_nconn(); + if (t < best) + { + best = t; + idx = i; + } + } + return workers[idx]; } public: @@ -256,7 +265,7 @@ class ConnPool { int max_listen_backlog = 10, double conn_server_timeout = 2, size_t seg_buff_size = 4096, - size_t nworker = 2): + size_t nworker = 4): ec(ec), max_listen_backlog(max_listen_backlog), conn_server_timeout(conn_server_timeout), @@ -267,6 +276,7 @@ class ConnPool { user_tcall = new ThreadCall(ec); disp_ec = workers[0].get_ec(); disp_tcall = workers[0].get_tcall(); + workers[0].set_dispatcher(); } ~ConnPool() { @@ -336,18 +346,6 @@ class ConnPool { template void reg_conn_handler(Func cb) { conn_cb = cb; } - - void terminate(const conn_t &conn, bool blocking = true) { - int fd = conn->fd; - auto worker = conn->worker; - if (worker) - worker->get_tcall()->call([conn](ThreadCall::Handle &) { - conn->stop(); - }, blocking); - else - conn->stop(); - remove_conn(fd); - } }; } diff --git a/include/salticidae/network.h b/include/salticidae/network.h index 60d8f20..4f934cd 100644 --- a/include/salticidae/network.h +++ b/include/salticidae/network.h @@ -470,7 +470,7 @@ void PeerNetwork::Conn::on_setup() { assert(!ev_timeout); ev_timeout = Event(worker->get_ec(), -1, [conn](int, int) { SALTICIDAE_LOG_INFO("peer ping-pong timeout"); - conn->terminate(); + conn->worker_terminate(); }); /* the initial ping-pong to set up the connection */ tcall_reset_timeout(worker, conn, pn->conn_timeout); @@ -508,7 +508,7 @@ void PeerNetwork::Peer::reset_conn(conn_t new_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->cpool->terminate(conn); + conn->disp_terminate(); } addr = new_conn->get_addr(); conn = new_conn; @@ -557,7 +557,7 @@ bool PeerNetwork::check_new_conn(const conn_t &conn, uint16_t port) { { if (conn != p->conn) { - conn->cpool->terminate(conn); + conn->disp_terminate(); return true; } return false; diff --git a/src/conn.cpp b/src/conn.cpp index f170d0a..841002f 100644 --- a/src/conn.cpp +++ b/src/conn.cpp @@ -67,7 +67,7 @@ void ConnPool::Conn::send_data(int fd, int events) { if (ret < 0 && errno != EWOULDBLOCK) { SALTICIDAE_LOG_INFO("send(%d) failure: %s", fd, strerror(errno)); - terminate(); + worker_terminate(); return; } } @@ -102,13 +102,13 @@ void ConnPool::Conn::recv_data(int fd, int events) { if (errno == EWOULDBLOCK) break; SALTICIDAE_LOG_INFO("recv(%d) failure: %s", fd, strerror(errno)); /* connection err or half-opened connection */ - terminate(); + worker_terminate(); return; } if (ret == 0) { //SALTICIDAE_LOG_INFO("recv(%d) terminates", fd, strerror(errno)); - terminate(); + worker_terminate(); return; } buff_seg.resize(ret); @@ -118,7 +118,17 @@ void ConnPool::Conn::recv_data(int fd, int events) { on_read(); } -void ConnPool::Conn::terminate() { +void ConnPool::Conn::stop() { + if (!self_ref) return; + if (worker) worker->unfeed(); + ev_connect.clear(); + ev_socket.clear(); + send_buffer.get_queue().unreg_handler(); + ::close(fd); + self_ref = nullptr; /* remove the self-cycle */ +} + +void ConnPool::Conn::worker_terminate() { stop(); cpool->disp_tcall->call( [cpool=this->cpool, fd=this->fd](ThreadCall::Handle &) { @@ -126,6 +136,15 @@ void ConnPool::Conn::terminate() { }); } +void ConnPool::Conn::disp_terminate(bool blocking) { + if (worker && !worker->is_dispatcher()) + worker->get_tcall()->call([conn=self()](ThreadCall::Handle &) { + conn->stop(); + }, blocking); + else stop(); + cpool->remove_conn(fd); +} + void ConnPool::accept_client(int fd, int) { int client_fd; struct sockaddr client_addr; @@ -173,7 +192,7 @@ void ConnPool::Conn::conn_server(int fd, int events) { { if (events & Event::TIMEOUT) SALTICIDAE_LOG_INFO("%s connect timeout", std::string(*this).c_str()); - cpool->terminate(conn); + conn->disp_terminate(); return; } } @@ -237,7 +256,7 @@ ConnPool::conn_t ConnPool::_connect(const NetAddr &addr) { sizeof(struct sockaddr_in)) < 0 && errno != EINPROGRESS) { SALTICIDAE_LOG_INFO("cannot connect to %s", std::string(addr).c_str()); - terminate(conn); + conn->disp_terminate(); } else { diff --git a/test/test_p2p.cpp b/test/test_p2p.cpp index f902afa..99c28a0 100644 --- a/test/test_p2p.cpp +++ b/test/test_p2p.cpp @@ -75,10 +75,11 @@ const uint8_t MsgAck::opcode; using MyNet = salticidae::PeerNetwork; -NetAddr addrs[] = { +std::vector addrs = { NetAddr("127.0.0.1:12345"), NetAddr("127.0.0.1:12346"), - NetAddr("127.0.0.1:12347") + NetAddr("127.0.0.1:12347"), + NetAddr("127.0.0.1:12348") }; void signal_handler(int) { @@ -97,7 +98,7 @@ int main(int argc, char **argv) { int i; net.start(); net.listen(addrs[i = atoi(argv[1])]); - for (int j = 0; j < 3; j++) + for (int j = 0; j < addrs.size(); j++) if (i != j) net.add_peer(addrs[j]); ec.dispatch(); } catch (salticidae::SalticidaeError &e) {} -- cgit v1.2.3