aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--CMakeLists.txt4
-rw-r--r--include/salticidae/buffer.h148
-rw-r--r--include/salticidae/conn.h392
-rw-r--r--include/salticidae/event.h60
-rw-r--r--include/salticidae/network.h64
-rw-r--r--include/salticidae/queue.h28
-rw-r--r--src/conn.cpp108
-rw-r--r--test/bench_network.cpp2
-rw-r--r--test/test_queue.cpp23
9 files changed, 594 insertions, 235 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 614e328..66cd2d4 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -39,11 +39,11 @@ if(BUILD_SHARED)
set_property(TARGET salticidae PROPERTY POSITION_INDEPENDENT_CODE 1)
add_library(salticidae_shared SHARED $<TARGET_OBJECTS:salticidae>)
set_target_properties(salticidae_shared PROPERTIES OUTPUT_NAME "salticidae")
- target_link_libraries(salticidae_shared event crypto)
+ target_link_libraries(salticidae_shared event crypto pthread)
endif()
add_library(salticidae_static STATIC $<TARGET_OBJECTS:salticidae>)
set_target_properties(salticidae_static PROPERTIES OUTPUT_NAME "salticidae")
-target_link_libraries(salticidae_static event crypto)
+target_link_libraries(salticidae_static event crypto pthread)
option(BUILD_TEST "build test binaries." OFF)
if(BUILD_TEST)
diff --git a/include/salticidae/buffer.h b/include/salticidae/buffer.h
new file mode 100644
index 0000000..3f415c9
--- /dev/null
+++ b/include/salticidae/buffer.h
@@ -0,0 +1,148 @@
+#ifndef _SALTICIDAE_BUFFER_H
+#define _SALTICIDAE_BUFFER_H
+
+#include <list>
+
+namespace salticidae {
+
+class SegBuffer {
+ public:
+ struct buffer_entry_t {
+ bytearray_t data;
+ bytearray_t::iterator offset;
+ buffer_entry_t(): offset(data.begin()) {}
+ buffer_entry_t(bytearray_t &&_data):
+ data(std::move(_data)), offset(data.begin()) {}
+
+ buffer_entry_t(buffer_entry_t &&other) {
+ size_t _offset = other.offset - other.data.begin();
+ data = std::move(other.data);
+ offset = data.begin() + _offset;
+ }
+
+ buffer_entry_t &operator=(buffer_entry_t &&other) {
+ size_t _offset = other.offset - other.data.begin();
+ data = std::move(other.data);
+ offset = data.begin() + _offset;
+ return *this;
+ }
+
+ buffer_entry_t(const buffer_entry_t &other): data(other.data) {
+ offset = data.begin() + (other.offset - other.data.begin());
+ }
+
+ size_t length() const { return data.end() - offset; }
+ };
+
+ private:
+ std::list<buffer_entry_t> buffer;
+ size_t _size;
+
+ public:
+ SegBuffer(): _size(0) {}
+ ~SegBuffer() { clear(); }
+
+ void swap(SegBuffer &other) {
+ std::swap(buffer, other.buffer);
+ std::swap(_size, other._size);
+ }
+
+ SegBuffer(const SegBuffer &other):
+ buffer(other.buffer), _size(other._size) {}
+
+ SegBuffer(SegBuffer &&other):
+ buffer(std::move(other.buffer)), _size(other._size) {
+ other._size = 0;
+ }
+
+ SegBuffer &operator=(SegBuffer &&other) {
+ if (this != &other)
+ {
+ SegBuffer tmp(std::move(other));
+ tmp.swap(*this);
+ }
+ return *this;
+ }
+
+ SegBuffer &operator=(const SegBuffer &other) {
+ if (this != &other)
+ {
+ SegBuffer tmp(other);
+ tmp.swap(*this);
+ }
+ return *this;
+ }
+
+ void rewind(bytearray_t &&data) {
+ _size += data.size();
+ buffer.push_front(buffer_entry_t(std::move(data)));
+ }
+
+ void push(bytearray_t &&data) {
+ _size += data.size();
+ buffer.push_back(buffer_entry_t(std::move(data)));
+ }
+
+ bytearray_t move_pop() {
+ auto res = std::move(buffer.front().data);
+ buffer.pop_front();
+ _size -= res.size();
+ return std::move(res);
+ }
+
+ bytearray_t pop(size_t len) {
+ bytearray_t res;
+ auto i = buffer.begin();
+ while (len && i != buffer.end())
+ {
+ size_t copy_len = std::min(i->length(), len);
+ res.insert(res.end(), i->offset, i->offset + copy_len);
+ i->offset += copy_len;
+ len -= copy_len;
+ if (i->offset == i->data.end())
+ i++;
+ }
+ buffer.erase(buffer.begin(), i);
+ _size -= res.size();
+ return std::move(res);
+ }
+
+ size_t size() const { return _size; }
+ bool empty() const { return buffer.empty(); }
+
+ void clear() {
+ buffer.clear();
+ _size = 0;
+ }
+};
+
+struct MPSCWriteBuffer {
+ using buffer_entry_t = SegBuffer::buffer_entry_t;
+ using queue_t = MPSCQueueEventDriven<buffer_entry_t>;
+ queue_t buffer;
+
+ MPSCWriteBuffer() {}
+
+ MPSCWriteBuffer(const SegBuffer &other) = delete;
+ MPSCWriteBuffer(SegBuffer &&other) = delete;
+
+ void rewind(bytearray_t &&data) {
+ buffer.rewind(buffer_entry_t(std::move(data)));
+ }
+
+ void push(bytearray_t &&data) {
+ buffer.enqueue(buffer_entry_t(std::move(data)));
+ }
+
+ bytearray_t move_pop() {
+ buffer_entry_t res;
+ buffer.try_dequeue(res);
+ return std::move(res.data);
+ }
+
+ queue_t &get_queue() { return buffer; }
+};
+
+}
+
+#endif
diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h
index f290e3d..26d19fe 100644
--- a/include/salticidae/conn.h
+++ b/include/salticidae/conn.h
@@ -36,6 +36,9 @@
#include <list>
#include <algorithm>
#include <exception>
+#include <mutex>
+#include <thread>
+#include <fcntl.h>
#include "salticidae/type.h"
#include "salticidae/ref.h"
@@ -43,110 +46,10 @@
#include "salticidae/util.h"
#include "salticidae/netaddr.h"
#include "salticidae/msg.h"
+#include "salticidae/buffer.h"
namespace salticidae {
-class SegBuffer {
- struct buffer_entry_t {
- bytearray_t data;
- bytearray_t::iterator offset;
- buffer_entry_t(bytearray_t &&_data): data(std::move(_data)) {
- offset = data.begin();
- }
-
- buffer_entry_t(buffer_entry_t &&other) {
- size_t _offset = other.offset - other.data.begin();
- data = std::move(other.data);
- offset = data.begin() + _offset;
- }
-
- buffer_entry_t(const buffer_entry_t &other): data(other.data) {
- offset = data.begin() + (other.offset - other.data.begin());
- }
-
- size_t length() const { return data.end() - offset; }
- };
-
- std::list<buffer_entry_t> buffer;
- size_t _size;
-
- public:
- SegBuffer(): _size(0) {}
- ~SegBuffer() { clear(); }
-
- void swap(SegBuffer &other) {
- std::swap(buffer, other.buffer);
- std::swap(_size, other._size);
- }
-
- SegBuffer(const SegBuffer &other):
- buffer(other.buffer), _size(other._size) {}
-
- SegBuffer(SegBuffer &&other):
- buffer(std::move(other.buffer)), _size(other._size) {
- other._size = 0;
- }
-
- SegBuffer &operator=(SegBuffer &&other) {
- if (this != &other)
- {
- SegBuffer tmp(std::move(other));
- tmp.swap(*this);
- }
- return *this;
- }
-
- SegBuffer &operator=(const SegBuffer &other) {
- if (this != &other)
- {
- SegBuffer tmp(other);
- tmp.swap(*this);
- }
- return *this;
- }
-
- void rewind(bytearray_t &&data) {
- _size += data.size();
- buffer.push_front(buffer_entry_t(std::move(data)));
- }
-
- void push(bytearray_t &&data) {
- _size += data.size();
- buffer.push_back(buffer_entry_t(std::move(data)));
- }
-
- bytearray_t move_pop() {
- auto res = std::move(buffer.front().data);
- buffer.pop_front();
- return std::move(res);
- }
-
- bytearray_t pop(size_t len) {
- bytearray_t res;
- auto i = buffer.begin();
- while (len && i != buffer.end())
- {
- size_t copy_len = std::min(i->length(), len);
- res.insert(res.end(), i->offset, i->offset + copy_len);
- i->offset += copy_len;
- len -= copy_len;
- if (i->offset == i->data.end())
- i++;
- }
- buffer.erase(buffer.begin(), i);
- _size -= res.size();
- return std::move(res);
- }
-
- size_t size() const { return _size; }
- bool empty() const { return buffer.empty(); }
-
- void clear() {
- buffer.clear();
- _size = 0;
- }
-};
-
struct ConnPoolError: public SalticidaeError {
using SalticidaeError::SalticidaeError;
};
@@ -156,7 +59,7 @@ class ConnPool {
public:
class Conn;
/** The handle to a bi-directional connection. */
- using conn_t = RcObj<Conn>;
+ using conn_t = ArcObj<Conn>;
/** The type of callback invoked when connection status is changed. */
using conn_callback_t = std::function<void(Conn &)>;
@@ -177,7 +80,8 @@ class ConnPool {
ConnMode mode;
NetAddr addr;
- SegBuffer send_buffer;
+ // TODO: send_buffer should be a thread-safe mpsc queue
+ MPSCWriteBuffer send_buffer;
SegBuffer recv_buffer;
Event ev_read;
@@ -190,6 +94,9 @@ class ConnPool {
void send_data(evutil_socket_t, short);
void conn_server(evutil_socket_t, short);
+ /** Terminate the connection. */
+ void terminate();
+
public:
Conn(): ready_send(false) {}
Conn(const Conn &) = delete;
@@ -206,7 +113,8 @@ class ConnPool {
const NetAddr &get_addr() const { return addr; }
ConnMode get_mode() const { return mode; }
ConnPool *get_pool() const { return cpool; }
- SegBuffer &read() { return recv_buffer; }
+ 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; }
@@ -214,17 +122,12 @@ class ConnPool {
* whenever I/O is available. */
void write(bytearray_t &&data) {
send_buffer.push(std::move(data));
- if (ready_send)
- send_data(fd, EV_WRITE);
- }
-
- /** Move the send buffer from the other (old) connection. */
- void move_send_buffer(conn_t other) {
- send_buffer = std::move(other->send_buffer);
}
- /** Terminate the connection. */
- void terminate();
+ ///** 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. */
@@ -238,35 +141,196 @@ class ConnPool {
}
/** Called when new data is available. */
- virtual void on_read() {
- if (cpool->read_cb) cpool->read_cb(*this);
- }
+ virtual void on_read() {}
/** Called when the underlying connection is established. */
virtual void on_setup() {
- if (cpool->conn_cb) cpool->conn_cb(*this);
+ cpool->update_conn(self());
}
/** Called when the underlying connection breaks. */
virtual void on_teardown() {
- if (cpool->conn_cb) cpool->conn_cb(*this);
+ cpool->update_conn(self());
}
};
-
+
private:
- int max_listen_backlog;
- double conn_server_timeout;
- size_t seg_buff_size;
- conn_callback_t read_cb;
- conn_callback_t conn_cb;
+ const int max_listen_backlog;
+ const double conn_server_timeout;
+ const size_t seg_buff_size;
+
+ /* owned by user loop */
+ int mlisten_fd[2]; /**< for connection events sent to the user loop */
+ Event ev_mlisten;
+ conn_callback_t conn_cb;
+ /* owned by the dispatcher */
std::unordered_map<int, conn_t> pool;
- int listen_fd;
+ 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));
+ }
+
+ struct Worker;
+ class WorkerFeed;
+
+ class WorkerCmd {
+ public:
+ virtual ~WorkerCmd() = default;
+ virtual void exec(Worker *worker) = 0;
+ };
+
+ class Worker {
+ EventContext ec;
+ Event ev_ctl;
+ int ctl_fd[2]; /**< for control messages from dispatcher */
+ std::thread handle;
+
+ public:
+ Worker() {
+ if (pipe2(ctl_fd, O_NONBLOCK))
+ throw ConnPoolError(std::string("failed to create worker pipe"));
+ ev_ctl = Event(ec, ctl_fd[0], EV_READ | EV_PERSIST, [this](int fd, short) {
+ WorkerCmd *dcmd;
+ read(fd, &dcmd, sizeof(dcmd));
+ dcmd->exec(this);
+ delete dcmd;
+ });
+ ev_ctl.add();
+ }
+
+ ~Worker() {
+ close(ctl_fd[0]);
+ close(ctl_fd[1]);
+ }
+
+ /* the following functions are called by the dispatcher */
+ void start() {
+ handle = std::thread([this]() { ec.dispatch(); });
+ }
+
+ void feed(const conn_t &conn, int client_fd) {
+ auto dcmd = new WorkerFeed(conn, client_fd);
+ write(ctl_fd[1], &dcmd, sizeof(dcmd));
+ }
+
+ void stop() {
+ auto dcmd = new WorkerStop();
+ write(ctl_fd[1], &dcmd, sizeof(dcmd));
+ }
+
+ std::thread &get_handle() { return handle; }
+ const EventContext &get_ec() { return ec; }
+ };
+
+ class WorkerFeed: public WorkerCmd {
+ conn_t conn;
+ int client_fd;
+
+ public:
+ WorkerFeed(const conn_t &conn, int client_fd):
+ conn(conn), client_fd(client_fd) {}
+ void exec(Worker *worker) override {
+ SALTICIDAE_LOG_INFO("worker %x got %s",
+ std::this_thread::get_id(),
+ std::string(*conn).c_str());
+ auto &ec = worker->get_ec();
+ conn->get_send_buffer()
+ .get_queue()
+ .reg_handler(ec, [conn=this->conn,
+ client_fd=this->client_fd](MPSCWriteBuffer::queue_t &) {
+ if (conn->ready_send)
+ conn->send_data(client_fd, EV_WRITE);
+ return false;
+ });
+ auto conn_ptr = conn.get();
+ conn->ev_read = Event(ec, client_fd, EV_READ,
+ std::bind(&Conn::recv_data, conn_ptr, _1, _2));
+ conn->ev_write = Event(ec, client_fd, EV_WRITE,
+ std::bind(&Conn::send_data, conn_ptr, _1, _2));
+ conn->ev_read.add();
+ conn->ev_write.add();
+ }
+ };
+
+ class WorkerStop: public WorkerCmd {
+ public:
+ void exec(Worker *worker) override { worker->get_ec().stop(); }
+ };
+
+ /* related to workers */
+ size_t nworker;
+ salticidae::BoxObj<Worker[]> workers;
void accept_client(evutil_socket_t, short);
- conn_t add_conn(conn_t conn);
+ conn_t add_conn(const conn_t &conn);
+ conn_t _connect(const NetAddr &addr);
+ void _listen(NetAddr listen_addr);
+ void _post_terminate(int fd);
+
+ 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);
+ }
+ };
+
+ 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));
+ }
+ };
+
+ class DspPostTerm: public DispatchCmd {
+ int fd;
+ public:
+ DspPostTerm(int fd): fd(fd) {}
+ void exec(ConnPool *cpool) override {
+ cpool->_post_terminate(fd);
+ }
+ };
+
+ 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));
+ }
+ };
+
+ void post_terminate(int fd) {
+ auto dcmd = new DspPostTerm(fd);
+ write(dlisten_fd[1], &dcmd, sizeof(dcmd));
+ }
+
+ Worker &select_worker() {
+ return workers[1];
+ }
protected:
EventContext ec;
+ EventContext dispatcher_ec;
+ std::mutex dsp_ec_mlock;
/** Should be implemented by derived class to return a new Conn object. */
virtual Conn *create_conn() = 0;
@@ -274,29 +338,91 @@ class ConnPool {
ConnPool(const EventContext &ec,
int max_listen_backlog = 10,
double conn_server_timeout = 2,
- size_t seg_buff_size = 4096):
- max_listen_backlog(max_listen_backlog),
- conn_server_timeout(conn_server_timeout),
- seg_buff_size(seg_buff_size),
- ec(ec) {}
+ size_t seg_buff_size = 4096,
+ size_t nworker = 2):
+ max_listen_backlog(max_listen_backlog),
+ conn_server_timeout(conn_server_timeout),
+ seg_buff_size(seg_buff_size),
+ listen_fd(-1),
+ nworker(std::min((size_t)1, nworker)),
+ ec(ec) {
+ if (pipe2(mlisten_fd, O_NONBLOCK))
+ throw ConnPoolError(std::string("failed to create main pipe"));
+ if (pipe2(dlisten_fd, O_NONBLOCK))
+ 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;
+ });
+ ev_mlisten.add();
+
+ workers = new Worker[nworker];
+ dispatcher_ec = workers[0].get_ec();
+
+ ev_dlisten = Event(dispatcher_ec, dlisten_fd[0], EV_READ | EV_PERSIST, [this](int fd, short) {
+ DispatchCmd *dcmd;
+ read(fd, &dcmd, sizeof(dcmd));
+ dcmd->exec(this);
+ delete dcmd;
+ });
+ ev_dlisten.add();
+
+ SALTICIDAE_LOG_INFO("starting all threads...");
+ for (size_t i = 0; i < nworker; i++)
+ workers[i].start();
+ }
~ConnPool() {
+ /* stop all workers */
+ for (size_t i = 0; i < nworker; i++)
+ workers[i].stop();
+ /* join all worker threads */
+ for (size_t i = 0; i < nworker; i++)
+ workers[i].get_handle().join();
for (auto it: pool)
{
conn_t conn = it.second;
conn->on_close();
}
+ if (listen_fd != -1) close(listen_fd);
+ for (int i = 0; i < 2; i++)
+ {
+ close(mlisten_fd[i]);
+ close(dlisten_fd[i]);
+ }
}
ConnPool(const ConnPool &) = delete;
ConnPool(ConnPool &&) = delete;
/** Actively connect to remote addr. */
- conn_t connect(const NetAddr &addr);
+ conn_t connect(const NetAddr &addr, bool blocking = true) {
+ if (blocking)
+ return _connect(addr);
+ else
+ {
+ auto dcmd = new DspConnect(addr);
+ write(dlisten_fd[1], &dcmd, sizeof(dcmd));
+ return nullptr;
+ }
+ }
+
/** 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);
+ 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));
+ }
+ }
template<typename Func>
void reg_conn_handler(Func cb) { conn_cb = cb; }
diff --git a/include/salticidae/event.h b/include/salticidae/event.h
index 857518b..ddb93fc 100644
--- a/include/salticidae/event.h
+++ b/include/salticidae/event.h
@@ -53,8 +53,8 @@ class EventContext: public _event_context_ot {
EventContext(EventContext &&) = default;
EventContext &operator=(const EventContext &) = default;
EventContext &operator=(EventContext &&) = default;
- void dispatch() { event_base_dispatch(get()); }
- void stop() { event_base_loopbreak(get()); }
+ void dispatch() const { event_base_dispatch(get()); }
+ void stop() const { event_base_loopbreak(get()); }
};
class Event {
@@ -122,7 +122,7 @@ class Event {
};
template<typename T>
-class MPSCQueueEventDriven: public MPMCQueue<T> {
+class MPSCQueueEventDriven: public MPSCQueue<T> {
private:
const uint64_t dummy = 1;
std::atomic<bool> wait_sig;
@@ -130,38 +130,39 @@ class MPSCQueueEventDriven: public MPMCQueue<T> {
Event ev;
public:
- template<typename Func>
- MPSCQueueEventDriven(const EventContext &ec, Func &&func,
- size_t burst_size = 128, size_t capacity = 65536):
- MPMCQueue<T>(capacity),
+ MPSCQueueEventDriven(size_t capacity = 65536):
+ MPSCQueue<T>(capacity),
wait_sig(true),
- fd(eventfd(0, EFD_NONBLOCK)),
- ev(Event(ec, fd, EV_READ | EV_PERSIST,
- [this, func=std::forward<Func>(func), burst_size](int, short) {
- uint64_t t;
- read(fd, &t, 8);
- //fprintf(stderr, "%x\n", std::this_thread::get_id());
- T elem;
- size_t cnt = burst_size;
- while (MPMCQueue<T>::try_dequeue(elem))
- {
- func(std::move(elem));
- if (!--cnt)
- {
- write(fd, &dummy, 8);
- return;
- }
- }
- wait_sig.store(true, std::memory_order_relaxed);
- })) { ev.add(); }
+ fd(eventfd(0, EFD_NONBLOCK)) {}
~MPSCQueueEventDriven() { close(fd); }
+ template<typename Func>
+ void reg_handler(const EventContext &ec, Func &&func) {
+ ev = Event(ec, fd, EV_READ | EV_PERSIST,
+ [this, func=std::forward<Func>(func)](int, short) {
+ //fprintf(stderr, "%x\n", std::this_thread::get_id());
+ uint64_t t;
+ read(fd, &t, 8);
+ // the only undesirable case is there are some new items
+ // enqueued before recovering wait_sig to true, so the consumer
+ // won't be notified. In this case, no enqueuing thread will
+ // get to write(fd). Then store(true) must happen after all exchange(false),
+ // since all enqueue operations are finalized, the dequeue should be able
+ // to see those enqueued values in func()
+ wait_sig.store(true, std::memory_order_release);
+ if (func(*this))
+ write(fd, &dummy, 8);
+ });
+ ev.add();
+ }
+
template<typename U>
bool enqueue(U &&e) {
static const uint64_t dummy = 1;
- bool ret = MPMCQueue<T>::enqueue(std::forward<U>(e));
- if (wait_sig.exchange(false, std::memory_order_relaxed))
+ bool ret = MPSCQueue<T>::enqueue(std::forward<U>(e));
+ // memory barrier here, so any load/store in enqueue must be finialized
+ if (wait_sig.exchange(false, std::memory_order_acq_rel))
{
SALTICIDAE_LOG_DEBUG("mpsc notify");
write(fd, &dummy, 8);
@@ -170,6 +171,8 @@ class MPSCQueueEventDriven: public MPMCQueue<T> {
}
};
+// TODO: incorrect MPMCQueueEventDriven impl
+/*
template<typename T>
class MPMCQueueEventDriven: public MPMCQueue<T> {
private:
@@ -223,6 +226,7 @@ class MPMCQueueEventDriven: public MPMCQueue<T> {
return ret;
}
};
+*/
}
diff --git a/include/salticidae/network.h b/include/salticidae/network.h
index 4e966d1..d82772f 100644
--- a/include/salticidae/network.h
+++ b/include/salticidae/network.h
@@ -103,7 +103,7 @@ class MsgNetwork: public ConnPool {
void on_read() override;
};
- using conn_t = RcObj<Conn>;
+ using conn_t = ArcObj<Conn>;
#ifdef SALTICIDAE_MSG_STAT
class msg_stat_by_opcode_t:
public std::unordered_map<typename Msg::opcode_t,
@@ -121,6 +121,8 @@ class MsgNetwork: public ConnPool {
std::unordered_map<
typename Msg::opcode_t,
std::function<void(const Msg &msg, Conn &)>> handler_map;
+ using queue_t = MPSCQueueEventDriven<std::pair<Msg, conn_t>>;
+ queue_t incoming_msgs;
protected:
#ifdef SALTICIDAE_MSG_STAT
@@ -134,10 +136,38 @@ class MsgNetwork: public ConnPool {
MsgNetwork(const EventContext &ec,
int max_listen_backlog,
double conn_server_timeout,
- size_t seg_buff_size):
+ size_t seg_buff_size,
+ size_t burst_size = 1000):
ConnPool(ec, max_listen_backlog,
conn_server_timeout,
- seg_buff_size) {}
+ seg_buff_size) {
+ incoming_msgs.reg_handler(ec, [this, burst_size](queue_t &q) {
+ std::pair<Msg, conn_t> item;
+ size_t cnt = 0;
+ while (q.try_dequeue(item))
+ {
+ auto &msg = item.first;
+ auto &conn = item.second;
+ auto it = handler_map.find(msg.get_opcode());
+ if (it == handler_map.end())
+ SALTICIDAE_LOG_WARN("unknown opcode: %s",
+ get_hex(msg.get_opcode()).c_str());
+ else /* call the handler */
+ {
+ SALTICIDAE_LOG_DEBUG("got message %s from %s",
+ std::string(msg).c_str(),
+ std::string(*conn).c_str());
+ it->second(msg, *conn);
+#ifdef SALTICIDAE_MSG_STAT
+ conn->nrecv++;
+ recv_by_opcode.add(msg);
+#endif
+ }
+ if (++cnt == burst_size) return true;
+ }
+ return false;
+ });
+ }
template<typename Func>
typename std::enable_if<std::is_constructible<
@@ -189,7 +219,7 @@ class ClientNetwork: public MsgNetwork<OpcodeType> {
void on_teardown() override;
};
- using conn_t = RcObj<Conn>;
+ using conn_t = ArcObj<Conn>;
protected:
ConnPool::Conn *create_conn() override { return new Conn(); }
@@ -249,7 +279,7 @@ class PeerNetwork: public MsgNetwork<OpcodeType> {
void on_teardown() override;
};
- using conn_t = RcObj<Conn>;
+ using conn_t = ArcObj<Conn>;
private:
struct Peer {
@@ -361,10 +391,11 @@ class PeerNetwork: public MsgNetwork<OpcodeType> {
}
};
+/* this callback is run by a worker */
template<typename OpcodeType>
void MsgNetwork<OpcodeType>::Conn::on_read() {
ConnPool::Conn::on_read();
- auto &recv_buffer = read();
+ auto &recv_buffer = get_recv_buffer();
auto mn = get_net();
while (get_fd() != -1)
{
@@ -389,21 +420,8 @@ void MsgNetwork<OpcodeType>::Conn::on_read() {
return;
}
#endif
- auto it = mn->handler_map.find(msg.get_opcode());
- if (it == mn->handler_map.end())
- SALTICIDAE_LOG_WARN("unknown opcode: %s",
- get_hex(msg.get_opcode()).c_str());
- else /* call the handler */
- {
- SALTICIDAE_LOG_DEBUG("got message %s from %s",
- std::string(msg).c_str(),
- std::string(*this).c_str());
- it->second(msg, *this);
-#ifdef SALTICIDAE_MSG_STAT
- nrecv++;
- mn->recv_by_opcode.add(msg);
-#endif
- }
+ mn->incoming_msgs.enqueue(
+ std::make_pair(std::move(msg), static_pointer_cast<Conn>(self())));
}
}
}
@@ -414,8 +432,8 @@ void PeerNetwork<O, _, __>::Peer::reset_conn(conn_t new_conn) {
{
if (conn)
{
- SALTICIDAE_LOG_DEBUG("moving send buffer");
- new_conn->move_send_buffer(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();
}
diff --git a/include/salticidae/queue.h b/include/salticidae/queue.h
index 88b3fc3..9045e2b 100644
--- a/include/salticidae/queue.h
+++ b/include/salticidae/queue.h
@@ -90,6 +90,7 @@ class FreeList {
template<typename T>
class MPMCQueue {
+ protected:
struct Block: public FreeList::Node {
T elem;
std::atomic<Block *> next;
@@ -171,6 +172,33 @@ class MPMCQueue {
}
};
+template<typename T>
+struct MPSCQueue: public MPMCQueue<T> {
+ using MPMCQueue<T>::MPMCQueue;
+ bool try_dequeue(T &e) {
+ auto h = this->head.load(std::memory_order_acquire);
+ auto nh = h->next.load(std::memory_order_relaxed);
+ if (nh == nullptr)
+ return false;
+ e = std::move(nh->elem);
+ this->head.store(nh, std::memory_order_release);
+ this->blks.push(h);
+ return true;
+ }
+
+ template<typename U>
+ bool rewind(U &&e) {
+ FreeList::Node * _nblk;
+ if (!this->blks.pop(_nblk)) return false;
+ auto nblk = static_cast<typename MPMCQueue<T>::Block *>(_nblk);
+ auto h = this->head.load(std::memory_order_acquire);
+ nblk->next.store(h, std::memory_order_release);
+ new (&(h->elem)) T(std::forward<U>(e));
+ this->head.store(nblk, std::memory_order_release);
+ return true;
+ }
+};
+
}
#endif
diff --git a/src/conn.cpp b/src/conn.cpp
index 0607ed3..5863f3c 100644
--- a/src/conn.cpp
+++ b/src/conn.cpp
@@ -44,14 +44,17 @@ ConnPool::Conn::operator std::string() const {
return std::move(s);
}
+/* the following two functions are executed by exactly one worker per Conn object */
+
void ConnPool::Conn::send_data(evutil_socket_t fd, short events) {
if (!(events & EV_WRITE)) return;
auto conn = self(); /* pin the connection */
ssize_t ret = seg_buff_size;
- while (!send_buffer.empty())
+ for (;;)
{
bytearray_t buff_seg = send_buffer.move_pop();
ssize_t size = buff_seg.size();
+ if (!size) break;
ret = send(fd, buff_seg.data(), size, MSG_NOSIGNAL);
SALTICIDAE_LOG_DEBUG("socket sent %zd bytes", ret);
size -= ret;
@@ -112,6 +115,12 @@ void ConnPool::Conn::recv_data(evutil_socket_t fd, short events) {
on_read();
}
+void ConnPool::Conn::terminate() {
+ ev_read.clear();
+ ev_write.clear();
+ cpool->post_terminate(fd);
+}
+
void ConnPool::accept_client(evutil_socket_t fd, short) {
int client_fd;
struct sockaddr client_addr;
@@ -135,33 +144,36 @@ void ConnPool::accept_client(evutil_socket_t fd, short) {
conn->mode = Conn::PASSIVE;
conn->addr = addr;
- Conn *conn_ptr = conn.get();
- conn->ev_read = Event(ec, client_fd, EV_READ,
- std::bind(&Conn::recv_data, conn_ptr, _1, _2));
- conn->ev_write = Event(ec, client_fd, EV_WRITE,
- std::bind(&Conn::send_data, conn_ptr, _1, _2));
- conn->ev_read.add();
- conn->ev_write.add();
+ //Conn *conn_ptr = conn.get();
+ // TODO: use worker thread ec
+ //conn->ev_read = Event(ec, client_fd, EV_READ,
+ // std::bind(&Conn::recv_data, conn_ptr, _1, _2));
+ //conn->ev_write = Event(ec, client_fd, EV_WRITE,
+ // std::bind(&Conn::send_data, conn_ptr, _1, _2));
+ //conn->ev_read.add();
+ //conn->ev_write.add();
add_conn(conn);
SALTICIDAE_LOG_INFO("created %s", std::string(*conn).c_str());
conn->on_setup();
+ select_worker().feed(conn, client_fd);
}
- ev_listen.add();
}
void ConnPool::Conn::conn_server(evutil_socket_t fd, short events) {
auto conn = self(); /* pin the connection */
if (send(fd, "", 0, MSG_NOSIGNAL) == 0)
{
- ev_read = Event(cpool->ec, fd, EV_READ,
- std::bind(&Conn::recv_data, this, _1, _2));
- ev_write = Event(cpool->ec, fd, EV_WRITE,
- std::bind(&Conn::send_data, this, _1, _2));
- ev_read.add();
- ev_write.add();
+ // TODO: use worker thread ec
+ //ev_read = Event(cpool->ec, fd, EV_READ,
+ // std::bind(&Conn::recv_data, this, _1, _2));
+ //ev_write = Event(cpool->ec, fd, EV_WRITE,
+ // std::bind(&Conn::send_data, this, _1, _2));
+ //ev_read.add();
+ //ev_write.add();
ev_connect.clear();
SALTICIDAE_LOG_INFO("connected to peer %s", std::string(*this).c_str());
on_setup();
+ cpool->select_worker().feed(self(), fd);
}
else
{
@@ -172,8 +184,14 @@ void ConnPool::Conn::conn_server(evutil_socket_t fd, short events) {
}
}
-void ConnPool::listen(NetAddr listen_addr) {
+void ConnPool::_listen(NetAddr listen_addr) {
+ std::lock_guard<std::mutex> _(cp_mlock);
int one = 1;
+ if (listen_fd != -1)
+ { /* reset the previous listen() */
+ ev_listen.clear();
+ close(listen_fd);
+ }
if ((listen_fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0)
throw ConnPoolError(std::string("cannot create socket for listening"));
if (setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, (const char *)&one, sizeof(one)) < 0 ||
@@ -192,28 +210,16 @@ void ConnPool::listen(NetAddr listen_addr) {
throw ConnPoolError(std::string("binding error"));
if (::listen(listen_fd, max_listen_backlog) < 0)
throw ConnPoolError(std::string("listen error"));
- ev_listen = Event(ec, listen_fd, EV_READ,
- std::bind(&ConnPool::accept_client, this, _1, _2));
- ev_listen.add();
- SALTICIDAE_LOG_INFO("listening to %u", ntohs(listen_addr.port));
-}
-
-void ConnPool::Conn::terminate() {
- auto &pool = cpool->pool;
- auto it = pool.find(fd);
- if (it != pool.end())
{
- /* temporarily pin the conn before it dies */
- auto conn = it->second;
- assert(conn.get() == this);
- pool.erase(it);
- on_close();
- /* inform the upper layer the connection will be destroyed */
- on_teardown();
+ std::lock_guard<std::mutex> _(dsp_ec_mlock);
+ ev_listen = Event(dispatcher_ec, listen_fd, EV_READ | EV_PERSIST,
+ std::bind(&ConnPool::accept_client, this, _1, _2));
+ ev_listen.add();
}
+ SALTICIDAE_LOG_INFO("listening to %u", ntohs(listen_addr.port));
}
-ConnPool::conn_t ConnPool::connect(const NetAddr &addr) {
+ConnPool::conn_t ConnPool::_connect(const NetAddr &addr) {
int fd;
int one = 1;
if ((fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) < 0)
@@ -245,23 +251,41 @@ ConnPool::conn_t ConnPool::connect(const NetAddr &addr) {
}
else
{
- conn->ev_connect = Event(ec, fd, EV_WRITE,
- std::bind(&Conn::conn_server, conn.get(), _1, _2));
- conn->ev_connect.add_with_timeout(conn_server_timeout);
-
+ {
+ std::lock_guard<std::mutex> _(dsp_ec_mlock);
+ conn->ev_connect = Event(dispatcher_ec, fd, EV_WRITE,
+ std::bind(&Conn::conn_server, conn.get(), _1, _2));
+ conn->ev_connect.add_with_timeout(conn_server_timeout);
+ }
add_conn(conn);
SALTICIDAE_LOG_INFO("created %s", std::string(*conn).c_str());
}
return conn;
}
-ConnPool::conn_t ConnPool::add_conn(conn_t conn) {
- auto it = pool.find(conn->fd);
+void ConnPool::_post_terminate(int fd) {
+ std::lock_guard<std::mutex> _(cp_mlock);
+ auto it = pool.find(fd);
if (it != pool.end())
{
- auto old_conn = it->second;
- old_conn->terminate();
+ /* temporarily pin the conn before it dies */
+ auto conn = it->second;
+ assert(conn->fd == fd);
+ pool.erase(it);
+ conn->on_close();
+ /* inform the upper layer the connection will be destroyed */
+ conn->on_teardown();
}
+}
+
+ConnPool::conn_t ConnPool::add_conn(const conn_t &conn) {
+ std::lock_guard<std::mutex> _(cp_mlock);
+ assert(pool.find(conn->fd) == pool.end());
+ //if (it != pool.end())
+ //{
+ // auto old_conn = it->second;
+ // old_conn->terminate();
+ //}
return pool.insert(std::make_pair(conn->fd, conn)).first->second;
}
diff --git a/test/bench_network.cpp b/test/bench_network.cpp
index 1185c0d..b1f1a0f 100644
--- a/test/bench_network.cpp
+++ b/test/bench_network.cpp
@@ -97,7 +97,7 @@ struct MyNet: public MsgNetworkByteOp {
struct Conn: public MsgNetworkByteOp::Conn {
MyNet *get_net() { return static_cast<MyNet *>(get_pool()); }
- salticidae::RcObj<Conn> self() {
+ salticidae::ArcObj<Conn> self() {
return salticidae::static_pointer_cast<Conn>(
MsgNetworkByteOp::Conn::self());
}
diff --git a/test/test_queue.cpp b/test/test_queue.cpp
index bed88c4..a2444d3 100644
--- a/test/test_queue.cpp
+++ b/test/test_queue.cpp
@@ -4,13 +4,22 @@
#include "salticidae/event.h"
-void test_mpsc(int nproducers = 16, int nops = 100000) {
+void test_mpsc(int nproducers = 16, int nops = 100000, size_t burst_size = 128) {
size_t total = nproducers * nops;
salticidae::EventContext ec;
std::atomic<size_t> collected(0);
- salticidae::MPSCQueueEventDriven<int> q(ec, [&collected](int x) {
- printf("%d\n", x);
- collected.fetch_add(1);
+ using queue_t = salticidae::MPSCQueueEventDriven<int>;
+ queue_t q;
+ q.reg_handler(ec, [&collected, burst_size](queue_t &q) {
+ size_t cnt = burst_size;
+ int x;
+ while (q.try_dequeue(x))
+ {
+ printf("%d\n", x);
+ collected.fetch_add(1);
+ if (!--cnt) return true;
+ }
+ return false;
});
std::vector<std::thread> producers;
std::thread consumer([&collected, total, &ec]() {
@@ -39,6 +48,7 @@ void test_mpsc(int nproducers = 16, int nops = 100000) {
fprintf(stderr, "consumers terminate\n");
}
+/*
void test_mpmc(int nproducers = 16, int nconsumers = 4, int nops = 100000) {
size_t total = nproducers * nops;
salticidae::MPMCQueueEventDriven<int> q;
@@ -84,9 +94,10 @@ void test_mpmc(int nproducers = 16, int nconsumers = 4, int nops = 100000) {
for (auto &t: consumers) t.join();
fprintf(stderr, "consumers terminate\n");
}
+*/
int main() {
- //test_mpsc();
- test_mpmc();
+ test_mpsc();
+ //test_mpmc();
return 0;
}
Powered by cgit, Themed by Ted Yin.