improve non-blocking API

8 files changed, 163 insertions, 127 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index b441591..903a66f 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -62,5 +62,5 @@ option(SALTICIDAE_NOCHECKSUM " disable checksum in messages" OFF)
 
 configure_file(src/config.h.in include/salticidae/config.h @ONLY)
 
-set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -W -Wall -Wextra -pedantic -Wsuggest-override")
-set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -W -Wall -Wextra -pedantic -Wsuggest-override")
+set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -faligned-new -W -Wall -Wextra -pedantic -Wsuggest-override")
+set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -faligned-new -W -Wall -Wextra -pedantic -Wsuggest-override")
diff --git a/include/salticidae/buffer.h b/include/salticidae/buffer.h
index 2324d8b..c7b11dd 100644
--- a/include/salticidae/buffer.h
+++ b/include/salticidae/buffer.h
@@ -126,14 +126,14 @@ struct MPSCWriteBuffer {
     MPSCWriteBuffer(const SegBuffer &other) = delete;
     MPSCWriteBuffer(SegBuffer &&other) = delete;
 
+    void set_capacity(size_t capacity) { buffer.set_capacity(capacity); }
+
     void rewind(bytearray_t &&data) {
         buffer.rewind(buffer_entry_t(std::move(data)));
     }
   
-    void push(bytearray_t &&data) {
-        buffer_entry_t d(std::move(data));
-        // TODO: better bounded buffer impl
-        while (!buffer.try_enqueue(d)) {}
+    bool push(bytearray_t &&data, bool unbounded) {
+        return buffer.enqueue(buffer_entry_t(std::move(data)), unbounded);
     }
 
     bytearray_t move_pop() {
diff --git a/include/salticidae/conn.h b/include/salticidae/conn.h
index b4df259..201e574 100644
--- a/include/salticidae/conn.h
+++ b/include/salticidae/conn.h
@@ -129,8 +129,8 @@ class ConnPool {
 
         /** Write data to the connection (non-blocking). The data will be sent
          * whenever I/O is available. */
-        void write(bytearray_t &&data) {
-            send_buffer.push(std::move(data));
+        bool write(bytearray_t &&data) {
+            return send_buffer.push(std::move(data), !cpool->queue_capacity);
         }
 
         protected:
@@ -156,6 +156,7 @@ class ConnPool {
     const int max_listen_backlog;
     const double conn_server_timeout;
     const size_t seg_buff_size;
+    const size_t queue_capacity;
 
     /* owned by user loop */
     BoxObj<ThreadCall> user_tcall;
@@ -287,13 +288,15 @@ class ConnPool {
         double _conn_server_timeout;
         size_t _seg_buff_size;
         size_t _nworker;
+        size_t _queue_capacity;
 
         public:
         Config():
             _max_listen_backlog(10),
             _conn_server_timeout(2),
             _seg_buff_size(4096),
-            _nworker(1) {}
+            _nworker(1),
+            _queue_capacity(0) {}
 
         Config &max_listen_backlog(int x) {
             _max_listen_backlog = x;
@@ -314,6 +317,11 @@ class ConnPool {
             _nworker = std::max((size_t)1, x);
             return *this;
         }
+
+        Config &queue_capacity(size_t x) {
+            _queue_capacity = x;
+            return *this;
+        }
     };
 
     ConnPool(const EventContext &ec, const Config &config):
@@ -321,6 +329,7 @@ class ConnPool {
             max_listen_backlog(config._max_listen_backlog),
             conn_server_timeout(config._conn_server_timeout),
             seg_buff_size(config._seg_buff_size),
+            queue_capacity(config._queue_capacity),
             listen_fd(-1),
             nworker(config._nworker),
             worker_running(false) {
diff --git a/include/salticidae/event.h b/include/salticidae/event.h
index 641bb8e..0600109 100644
--- a/include/salticidae/event.h
+++ b/include/salticidae/event.h
@@ -434,107 +434,6 @@ class ThreadNotifier {
     }
 };
 
-class ThreadCall {
-    int ctl_fd[2];
-    EventContext ec;
-    FdEvent ev_listen;
-
-    public:
-    struct Result {
-        void *data;
-        std::function<void(void *)> deleter;
-        Result(): data(nullptr) {}
-        Result(void *data, std::function<void(void *)> &&deleter):
-            data(data), deleter(std::move(deleter)) {}
-        ~Result() { if (data != nullptr) deleter(data); }
-        Result(const Result &) = delete;
-        Result(Result &&other):
-                data(other.data), deleter(std::move(other.deleter)) {
-            other.data = nullptr;
-        }
-        void swap(Result &other) {
-            std::swap(data, other.data);
-            std::swap(deleter, other.deleter);
-        }
-        Result &operator=(const Result &other) = delete;
-        Result &operator=(Result &&other) {
-            if (this != &other)
-            {
-                Result tmp(std::move(other));
-                tmp.swap(*this);
-            }
-            return *this;
-        }
-        void *get() { return data; }
-    };
-    class Handle {
-        std::function<void(Handle &)> callback;
-        ThreadNotifier<Result> * notifier;
-        Result result;
-        friend ThreadCall;
-        public:
-        Handle(): notifier(nullptr) {}
-        void exec() {
-            callback(*this);
-            if (notifier)
-                notifier->notify(std::move(result));
-        }
-        template<typename T>
-        void set_result(T &&data) {
-            using _T = std::remove_reference_t<T>;
-            result = Result(new _T(std::forward<T>(data)),
-                            [](void *ptr) {delete static_cast<_T *>(ptr);});
-        }
-    };
-
-    ThreadCall() = default;
-    ThreadCall(const ThreadCall &) = delete;
-    ThreadCall(ThreadCall &&) = delete;
-    ThreadCall(EventContext ec): ec(ec) {
-        if (pipe2(ctl_fd, O_NONBLOCK))
-            throw SalticidaeError(std::string("ThreadCall: failed to create pipe"));
-        ev_listen = FdEvent(ec, ctl_fd[0], [this](int fd, int) {
-            Handle *h;
-            read(fd, &h, sizeof(h));
-            std::atomic_thread_fence(std::memory_order_acquire);
-            h->exec();
-            delete h;
-        });
-        ev_listen.add(FdEvent::READ);
-    }
-
-    ~ThreadCall() {
-        ev_listen.clear();
-        Handle *h;
-        while (read(ctl_fd[0], &h, sizeof(h)) == sizeof(h))
-            delete h;
-        close(ctl_fd[0]);
-        close(ctl_fd[1]);
-    }
-
-    template<typename Func>
-    void async_call(Func callback) {
-        auto h = new Handle();
-        h->callback = callback;
-        std::atomic_thread_fence(std::memory_order_release);
-        write(ctl_fd[1], &h, sizeof(h));
-    }
-
-    template<typename Func>
-    Result call(Func callback) {
-        auto h = new Handle();
-        h->callback = callback;
-        ThreadNotifier<Result> notifier;
-        h->notifier = &notifier;
-        std::atomic_thread_fence(std::memory_order_release);
-        write(ctl_fd[1], &h, sizeof(h));
-        return notifier.wait();
-    }
-
-    const EventContext &get_ec() const { return ec; }
-};
-
-
 template<typename T>
 class MPSCQueueEventDriven: public MPSCQueue<T> {
     private:
@@ -544,8 +443,7 @@ class MPSCQueueEventDriven: public MPSCQueue<T> {
     FdEvent ev;
 
     public:
-    MPSCQueueEventDriven(size_t capacity = 65536):
-        MPSCQueue<T>(capacity),
+    MPSCQueueEventDriven():
         wait_sig(true),
         fd(eventfd(0, EFD_NONBLOCK)) {}
 
@@ -577,9 +475,10 @@ class MPSCQueueEventDriven: public MPSCQueue<T> {
     void unreg_handler() { ev.clear(); }
 
     template<typename U>
-    bool enqueue(U &&e) {
+    bool enqueue(U &&e, bool unbounded = true) {
         static const uint64_t dummy = 1;
-        MPSCQueue<T>::enqueue(std::forward<U>(e));
+        if (!MPSCQueue<T>::enqueue(std::forward<U>(e), unbounded))
+            return false;
         // memory barrier here, so any load/store in enqueue must be finialized
         if (wait_sig.exchange(false, std::memory_order_acq_rel))
         {
@@ -613,8 +512,7 @@ class MPMCQueueEventDriven: public MPMCQueue<T> {
     std::vector<FdEvent> evs;
 
     public:
-    MPMCQueueEventDriven(size_t capacity = 65536):
-        MPMCQueue<T>(capacity),
+    MPMCQueueEventDriven():
         wait_sig(true),
         fd(eventfd(0, EFD_NONBLOCK)) {}
 
@@ -642,9 +540,10 @@ class MPMCQueueEventDriven: public MPMCQueue<T> {
     void unreg_handlers() { evs.clear(); }
 
     template<typename U>
-    bool enqueue(U &&e) {
+    bool enqueue(U &&e, bool unbounded = true) {
         static const uint64_t dummy = 1;
-        MPMCQueue<T>::enqueue(std::forward<U>(e));
+        if (!MPMCQueue<T>::enqueue(std::forward<U>(e), unbounded))
+            return false;
         // memory barrier here, so any load/store in enqueue must be finialized
         if (wait_sig.exchange(false, std::memory_order_acq_rel))
         {
@@ -655,6 +554,107 @@ class MPMCQueueEventDriven: public MPMCQueue<T> {
     }
 };
 
+class ThreadCall {
+    public: class Handle;
+    private:
+    int ctl_fd[2];
+    EventContext ec;
+    const size_t burst_size;
+    using queue_t = MPSCQueueEventDriven<Handle *>;
+    queue_t q;
+
+    public:
+    struct Result {
+        void *data;
+        std::function<void(void *)> deleter;
+        Result(): data(nullptr) {}
+        Result(void *data, std::function<void(void *)> &&deleter):
+            data(data), deleter(std::move(deleter)) {}
+        ~Result() { if (data != nullptr) deleter(data); }
+        Result(const Result &) = delete;
+        Result(Result &&other):
+                data(other.data), deleter(std::move(other.deleter)) {
+            other.data = nullptr;
+        }
+        void swap(Result &other) {
+            std::swap(data, other.data);
+            std::swap(deleter, other.deleter);
+        }
+        Result &operator=(const Result &other) = delete;
+        Result &operator=(Result &&other) {
+            if (this != &other)
+            {
+                Result tmp(std::move(other));
+                tmp.swap(*this);
+            }
+            return *this;
+        }
+        void *get() { return data; }
+    };
+    class Handle {
+        std::function<void(Handle &)> callback;
+        ThreadNotifier<Result> * notifier;
+        Result result;
+        friend ThreadCall;
+        public:
+        Handle(): notifier(nullptr) {}
+        void exec() {
+            callback(*this);
+            if (notifier)
+                notifier->notify(std::move(result));
+        }
+        template<typename T>
+        void set_result(T &&data) {
+            using _T = std::remove_reference_t<T>;
+            result = Result(new _T(std::forward<T>(data)),
+                            [](void *ptr) {delete static_cast<_T *>(ptr);});
+        }
+    };
+
+    ThreadCall(size_t burst_size): burst_size(burst_size) {}
+    ThreadCall(const ThreadCall &) = delete;
+    ThreadCall(ThreadCall &&) = delete;
+    ThreadCall(EventContext ec, size_t burst_size = 128): ec(ec), burst_size(burst_size) {
+        q.reg_handler(ec, [this, burst_size=burst_size](queue_t &q) {
+            size_t cnt = 0;
+            Handle *h;
+            while (q.try_dequeue(h))
+            {
+                h->exec();
+                delete h;
+                if (++cnt == burst_size) return true;
+            }
+            return false;
+        });
+    }
+
+    ~ThreadCall() {
+        Handle *h;
+        while (q.try_dequeue(h)) delete h;
+        close(ctl_fd[0]);
+        close(ctl_fd[1]);
+    }
+
+    template<typename Func>
+    void async_call(Func callback) {
+        auto h = new Handle();
+        h->callback = callback;
+        q.enqueue(h);
+    }
+
+    template<typename Func>
+    Result call(Func callback) {
+        auto h = new Handle();
+        h->callback = callback;
+        ThreadNotifier<Result> notifier;
+        h->notifier = &notifier;
+        q.enqueue(h);
+        return notifier.wait();
+    }
+
+    const EventContext &get_ec() const { return ec; }
+};
+
 }
 
 #endif
diff --git a/include/salticidae/network.h b/include/salticidae/network.h
index 7814d56..2cef81d 100644
--- a/include/salticidae/network.h
+++ b/include/salticidae/network.h
@@ -140,6 +140,7 @@ class MsgNetwork: public ConnPool {
 
     MsgNetwork(const EventContext &ec, const Config &config):
             ConnPool(ec, config) {
+        incoming_msgs.set_capacity(65536);
         incoming_msgs.reg_handler(ec, [this, burst_size=config._burst_size](queue_t &q) {
             std::pair<Msg, conn_t> item;
             size_t cnt = 0;
@@ -179,7 +180,7 @@ class MsgNetwork: public ConnPool {
     }
 
     template<typename MsgType>
-    void send_msg(MsgType &&msg, const conn_t &conn);
+    bool send_msg(MsgType &&msg, const conn_t &conn);
     using ConnPool::listen;
 #ifdef SALTICIDAE_MSG_STAT
 #endif
@@ -413,6 +414,8 @@ class PeerNetwork: public MsgNetwork<OpcodeType> {
     using MsgNet::send_msg;
     template<typename MsgType>
     void send_msg(MsgType &&msg, const NetAddr &paddr);
+    template<typename MsgType>
+    void multicast_msg(MsgType &&msg, const std::vector<NetAddr> &paddrs);
     void listen(NetAddr listen_addr);
     bool has_peer(const NetAddr &paddr) const;
     conn_t connect(const NetAddr &addr) = delete;
@@ -455,16 +458,16 @@ void MsgNetwork<OpcodeType>::Conn::on_read() {
 
 template<typename OpcodeType>
 template<typename MsgType>
-void MsgNetwork<OpcodeType>::send_msg(MsgType &&_msg, const conn_t &conn) {
+bool MsgNetwork<OpcodeType>::send_msg(MsgType &&_msg, const conn_t &conn) {
     Msg msg(std::forward<MsgType>(_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++;
 #endif
+    return conn->write(std::move(msg_data));
 }
 
 template<typename O, O _, O __>
@@ -697,6 +700,22 @@ void PeerNetwork<O, _, __>::send_msg(MsgType &&msg, const NetAddr &paddr) {
         send_msg(std::move(msg), it->second->conn);
     });
 }
+
+template<typename O, O _, O __>
+template<typename MsgType>
+void PeerNetwork<O, _, __>::multicast_msg(MsgType &&msg, const std::vector<NetAddr> &paddrs) {
+    this->disp_tcall->async_call(
+                [this, msg=std::forward<MsgType>(msg), paddrs](ThreadCall::Handle &) {
+        for (auto &paddr: paddrs)
+        {
+            auto it = id2peer.find(paddr);
+            if (it == id2peer.end())
+                throw PeerNetworkError("peer does not exist");
+            send_msg(std::move(msg), it->second->conn);
+        }
+    });
+}
+
 /* end: functions invoked by the user loop */
 
 template<typename OpcodeType>
diff --git a/include/salticidae/queue.h b/include/salticidae/queue.h
index a72a28f..0b4dae0 100644
--- a/include/salticidae/queue.h
+++ b/include/salticidae/queue.h
@@ -107,10 +107,8 @@ class MPMCQueue {
     MPMCQueue(const MPMCQueue &) = delete;
     MPMCQueue(MPMCQueue &&) = delete;
 
-    MPMCQueue(size_t capacity = 65536): head(new Block()), tail(head.load()) {
+    MPMCQueue(): head(new Block()), tail(head.load()) {
         head.load()->next = nullptr;
-        while (capacity--)
-            blks.push(new Block());
     }
 
     ~MPMCQueue() {
@@ -122,10 +120,18 @@ class MPMCQueue {
         }
     }
 
+    void set_capacity(size_t capacity = 0) {
+        while (capacity--) blks.push(new Block());
+    }
+
     template<typename U>
-    bool enqueue(U &&e) {
+    bool enqueue(U &&e, bool unbounded = true) {
         FreeList::Node * _nblk;
-        if (!blks.pop(_nblk)) _nblk = new Block();
+        if (!blks.pop(_nblk))
+        {
+            if (unbounded) _nblk = new Block();
+            else return false;
+        }
         _enqueue(static_cast<Block *>(_nblk), std::forward<U>(e));
         return true;
     }
diff --git a/src/conn.cpp b/src/conn.cpp
index 4ab3040..5fc59f3 100644
--- a/src/conn.cpp
+++ b/src/conn.cpp
@@ -185,6 +185,7 @@ void ConnPool::accept_client(int fd, int) {
         NetAddr addr((struct sockaddr_in *)&client_addr);
         conn_t conn = create_conn();
         conn->self_ref = conn;
+        conn->send_buffer.set_capacity(queue_capacity);
         conn->seg_buff_size = seg_buff_size;
         conn->fd = client_fd;
         conn->cpool = this;
@@ -264,6 +265,7 @@ ConnPool::conn_t ConnPool::_connect(const NetAddr &addr) {
         throw ConnPoolError(std::string("unable to set nonblocking socket"));
     conn_t conn = create_conn();
     conn->self_ref = conn;
+    conn->send_buffer.set_capacity(queue_capacity);
     conn->seg_buff_size = seg_buff_size;
     conn->fd = fd;
     conn->cpool = this;
diff --git a/test/bench_network.cpp b/test/bench_network.cpp
index 89289fe..58f4a64 100644
--- a/test/bench_network.cpp
+++ b/test/bench_network.cpp
@@ -79,7 +79,7 @@ struct MyNet: public MsgNetworkByteOp {
             const std::string name,
             const NetAddr &peer,
             double stat_timeout = -1):
-            MsgNetworkByteOp(ec, MsgNetworkByteOp::Config().burst_size(1000)),
+            MsgNetworkByteOp(ec, MsgNetworkByteOp::Config().burst_size(1000).queue_capacity(65536)),
             name(name),
             peer(peer),
             ev_period_stat(ec, [this, stat_timeout](TimerEvent &) {