Introduced a lock-free MPSC queue between the user and I/O thread to make sure...

Introduced a lock-free MPSC queue between the user and I/O thread to make sure that all I/O operations happen in the dedicated I/O thread

main.cc

@@ -50,6 +50,12 @@ class MyHandler : public Net::Http::Handler {
         else if (req.resource() == "/timeout") {
             timeout.arm(std::chrono::seconds(5));
         }
+        else if (req.resource() == "/async") {
+            std::thread([](Net::Http::Response response) {
+                std::this_thread::sleep_for(std::chrono::seconds(1));
+                response.send(Net::Http::Code::Ok, "Async response");
+            }, std::move(response)).detach();
+        }
     }
 
     void onTimeout(const Net::Http::Request& req, Net::Http::Response response) {

src/http.cc

@@ -402,7 +402,7 @@ void
 ResponseStream::flush() {
     io_->disarmTimer();
     auto buf = buf_.buffer();
-    peer()->send(buf.data, buf.len);
+    peer()->send(buf);
 
     buf_.clear();
 }
@@ -460,7 +460,7 @@ Response::putOnWire(const char* data, size_t len)
 
 #undef OUT
 
-        return peer()->send(buffer.data, buffer.len);
+        return peer()->send(buffer);
 
     } catch (const std::runtime_error& e) {
         return Async::Promise<ssize_t>::rejected(e);

src/io.cc

@@ -229,7 +229,10 @@ IoWorker::run() {
     if (pins.count() > 0) {
     }
 
+    thisId = std::this_thread::get_id();
+
     mailbox.bind(poller);
+    writesQueue.bind(poller);
 
     auto n = notifier.tag();
     poller.addFd(n.value(), NotifyOn::Read, n, Polling::Mode::Edge);
@@ -257,6 +260,9 @@ IoWorker::run() {
                 else if (event.tag == notifier.tag()) {
                     handleNotify();
                 }
+                else if (event.tag == writesQueue.tag()) {
+                    handleWriteQueue();
+                }
                 else {
                     if (event.flags.hasFlag(NotifyOn::Read)) {
                         auto fd = event.tag.value();
@@ -277,18 +283,10 @@ IoWorker::run() {
                             throw std::runtime_error("Assertion Error: could not find write data");
                         }
 
-                        auto &write = it->second;
-                        ssize_t bytes = ::send(fd, write.buf, write.len, 0);
-                        if (bytes < 0) {
-                            write.reject(Net::Error::system("Could not write data"));
-                        }
+                        poller.rearmFd(fd, NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
 
-                        else if (bytes < write.len) {
-                            write.reject(Net::Error("Failed to write: could not write all bytes"));
-                        }
-                        else {
-                            write.resolve(bytes);
-                        }
+                        const auto& write = it->second;
+                        asyncWriteImpl(fd, write, Retry);
                     }
                 }
             }
@@ -338,8 +336,33 @@ IoWorker::handleNotify() {
     load = None();
 }
 
+void
+IoWorker::handleWriteQueue() {
+    // Let's drain the queue
+    for (;;) {
+        std::unique_ptr<PollableQueue<OnHoldWrite>::Entry> entry(writesQueue.pop());
+        if (!entry) break;
+
+        const auto &write = entry->data();
+        asyncWriteImpl(write.fd, write);
+    }
+}
+
 Async::Promise<ssize_t>
-IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
+IoWorker::asyncWrite(Fd fd, const Buffer& buffer) {
+
+    // If the I/O operation has been initiated from an other thread, we queue it and we'll process
+    // it in our own thread so that we make sure that every I/O operation happens in the right thread
+    if (std::this_thread::get_id() != thisId) {
+        return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
+            auto detached = buffer.detach();
+            OnHoldWrite write(std::move(resolve), std::move(reject), detached);
+            write.fd = fd;
+            auto *e = writesQueue.allocEntry(write);
+            writesQueue.push(e);
+        });
+    }
+
     return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
 
         auto it = toWrite.find(fd);
@@ -348,35 +371,61 @@ IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
             return;
         }
 
-        ssize_t totalWritten = 0;
-        for (;;) {
-            auto *bufPtr = static_cast<const char *>(buf) + totalWritten;
-            auto bufLen = len - totalWritten;
-            ssize_t bytesWritten = ::send(fd, bufPtr, bufLen, 0);
-            if (bytesWritten < 0) {
-                if (errno == EAGAIN || errno == EWOULDBLOCK) {
+        asyncWriteImpl(fd, buffer, resolve, reject);
+
+    });
+}
+
+void
+IoWorker::asyncWriteImpl(Fd fd, const IoWorker::OnHoldWrite& entry, WriteStatus status) {
+    asyncWriteImpl(fd, entry.buffer, entry.resolve, entry.reject, status);
+}
+
+void
+IoWorker::asyncWriteImpl(
+        Fd fd, const Buffer& buffer,
+        Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
+{
+    auto cleanUp = [&]() {
+        if (buffer.isOwned) delete[] buffer.data;
+        if (status == Retry)
+            toWrite.erase(fd);
+    };
+
+    ssize_t totalWritten = 0;
+    for (;;) {
+        auto *ptr = buffer.data + totalWritten;
+        auto len = buffer.len - totalWritten;
+        ssize_t bytesWritten = ::send(fd, ptr, len, 0);
+        if (bytesWritten < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                auto buf = buffer.isOwned ? buffer : buffer.detach(totalWritten);
+                if (status == FirstTry) {
                     toWrite.insert(
-                            std::make_pair(fd, 
-                                OnHoldWrite(std::move(resolve), std::move(reject), bufPtr, bufLen)));
-                    poller.addFdOneShot(fd, NotifyOn::Write, Polling::Tag(fd));
-                }
-                else {
-                    reject(Net::Error::system("Could not write data"));
+                            std::make_pair(fd,
+                                OnHoldWrite(std::move(resolve), std::move(reject), buf)));
                 }
-                break;
+                poller.rearmFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Tag(fd), Polling::Mode::Edge);
             }
             else {
-                totalWritten += bytesWritten;
-                if (totalWritten == len) {
-                    resolve(totalWritten);
-                    break;
-                }
+                cleanUp();
+                reject(Net::Error::system("Could not write data"));
             }
+            break;
         }
-    });
+        else {
+            totalWritten += bytesWritten;
+            if (totalWritten == len) {
+                cleanUp();
+                resolve(totalWritten);
+                break;
+            }
+        }
+    }
 }
 
 
+
 } // namespace Tcp
 
 } // namespace Net

src/io.h

@@ -11,6 +11,7 @@
 #include "os.h"
 #include "tcp.h"
 #include "async.h"
+#include "stream.h"
 
 #include <thread>
 #include <mutex>
@@ -61,18 +62,18 @@ public:
 private:
     struct OnHoldWrite {
         OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
-                    const void *buf, size_t len)
+                    Buffer buffer)
             : resolve(std::move(resolve))
             , reject(std::move(reject))
-            , buf(buf)
-            , len(len)
+            , buffer(std::move(buffer))
+            , fd(-1)
         { }
 
         Async::Resolver resolve;
         Async::Rejection reject;
 
-        const void *buf;
-        size_t len;   
+        Buffer buffer;
+        Fd fd;
     };
 
     void
@@ -104,6 +105,11 @@ private:
         Async::Rejection reject;
     };
 
+    enum WriteStatus {
+        FirstTry,
+        Retry
+    };
+
     Polling::Epoll poller;
     std::unique_ptr<std::thread> thread;
     mutable std::mutex peersMutex;
@@ -121,16 +127,26 @@ private:
 
     CpuSet pins;
 
+    std::thread::id thisId;
+    PollableQueue<OnHoldWrite> writesQueue;
+
     std::shared_ptr<Peer>& getPeer(Fd fd);
     std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
 
-    Async::Promise<ssize_t> asyncWrite(Fd fd, const void *buf, size_t len);
+    Async::Promise<ssize_t> asyncWrite(Fd fd, const Buffer& buffer);
+
+    void asyncWriteImpl(Fd fd, const OnHoldWrite& entry, WriteStatus status = FirstTry);
+    void asyncWriteImpl(
+            Fd fd, const Buffer& buffer,
+            Async::Resolver resolve, Async::Rejection reject,
+            WriteStatus status = FirstTry);
 
     void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);
 
     void handleIncoming(const std::shared_ptr<Peer>& peer);
     void handleTimeout();
     void handleNotify();
+    void handleWriteQueue();
     void run();
 
 };

src/mailbox.h

@@ -132,3 +132,156 @@ public:
 private:
     int event_fd;
 };
+
+/*
+ * An unbounded MPSC lock-free queue. Usefull for efficient cross-thread message passing.
+
+ * push() and pop() are wait-free.
+ * Might replace the Mailbox implementation below
+
+ * Design comes from http://www.1024cores.net/home/lock-free-algorithms/queues/non-intrusive-mpsc-node-based-queue
+*/
+template<typename T>
+class Queue {
+public:
+    struct Entry {
+        friend class Queue;
+
+        T data() const {
+            return *reinterpret_cast<const T*>(&storage);
+        }
+
+        ~Entry() {
+            auto *d = reinterpret_cast<T *>(&storage);
+            d->~T();
+        }
+    private:
+        typedef typename std::aligned_storage<sizeof(T), alignof(T)>::type Storage;
+        Storage storage;
+        std::atomic<Entry *> next;
+    };
+
+    Queue()
+    {
+        auto *sentinel = new Entry;
+        sentinel->next = nullptr;
+        head.store(sentinel, std::memory_order_relaxed);
+        tail = sentinel;
+    }
+
+    virtual ~Queue() {
+        while (auto *e = pop()) delete e;
+    }
+
+    template<typename U>
+    Entry* allocEntry(U&& u) const {
+        auto *e = new Entry;
+        new (&e->storage) T(std::forward<U>(u));
+        return e;
+    }
+
+    void push(Entry *entry) {
+        entry->next = nullptr;
+        // @Note: we're using SC atomics here (exchange will issue a full fence),
+        // but I don't think we should bother relaxing them for now
+        auto *prev = head.exchange(entry);
+        prev->next = entry;
+    }
+
+    Entry* pop() {
+        auto *res = tail;
+        auto *next = res->next.load(std::memory_order_acquire);
+        if (next) {
+            // Since it's Single-Consumer, the store does not need to be atomic
+            tail = next;
+            new (&res->storage) T(next->data());
+            return res;
+        }
+        return nullptr;
+    }
+
+private:
+    std::atomic<Entry *> head;
+    Entry *tail;
+};
+
+template<typename T>
+class PollableQueue : public Queue<T>
+{
+public:
+    typedef typename Queue<T>::Entry Entry;
+
+    PollableQueue()
+     : event_fd(-1) {
+    }
+
+    ~PollableQueue() {
+        if (event_fd != -1) close(event_fd);
+    }
+
+    bool isBound() const {
+        return event_fd != -1;
+    }
+
+    Polling::Tag bind(Polling::Epoll& poller) {
+        using namespace Polling;
+
+        if (isBound()) {
+            throw std::runtime_error("The queue has already been bound");
+        }
+
+        event_fd = TRY_RET(eventfd(0, EFD_NONBLOCK));
+        Tag tag(event_fd);
+        poller.addFd(event_fd, NotifyOn::Read, tag);
+
+        return tag;
+    }
+
+    void push(Entry* entry) {
+        Queue<T>::push(entry);
+
+        if (isBound()) {
+            uint64_t val = 1;
+            TRY(write(event_fd, &val, sizeof val));
+        }
+    }
+
+    Entry *pop() {
+        auto ret = Queue<T>::pop();
+
+        if (isBound()) {
+            uint64_t val;
+            for (;;) {
+                ssize_t bytes = read(event_fd, &val, sizeof val);
+                if (bytes == -1) {
+                    if (errno == EAGAIN || errno == EWOULDBLOCK)
+                        break;
+                    else {
+                        // TODO
+                    }
+                }
+            }
+        }
+
+        return ret;
+
+    }
+
+    Polling::Tag tag() const {
+        if (!isBound())
+            throw std::runtime_error("Can not retrieve tag of an unbound mailbox");
+
+        return Polling::Tag(event_fd);
+    }
+
+    void unbind(Polling::Epoll& poller) {
+        if (event_fd == -1) {
+            throw std::runtime_error("The mailbox is not bound");
+        }
+
+        poller.removeFd(event_fd);
+        close(event_fd), event_fd = -1;
+    }
+private:
+    int event_fd;
+};

src/peer.cc

@@ -78,8 +78,8 @@ Peer::tryGetData(std::string(name)) const {
 }
 
 Async::Promise<ssize_t>
-Peer::send(const void* buf, size_t len) {
-    return io_->asyncWrite(fd_, buf, len);
+Peer::send(const Buffer& buffer) {
+    return io_->asyncWrite(fd_, buffer);
 }
 
 

src/peer.h

@@ -9,6 +9,7 @@
 #include "net.h"
 #include "os.h"
 #include "async.h"
+#include "stream.h"
 #include <string>
 #include <iostream>
 #include <memory>
@@ -50,7 +51,7 @@ public:
         return std::static_pointer_cast<T>(data);
     }
 
-    Async::Promise<ssize_t> send(const void* buf, size_t len);
+    Async::Promise<ssize_t> send(const Buffer& buffer);
 
 private:
     IoWorker* io_;

src/stream.h

@@ -12,6 +12,7 @@
 #include <streambuf>
 #include <vector>
 #include <limits>
+#include <iostream>
 
 static constexpr char CR = 0xD;
 static constexpr char LF = 0xA;
@@ -113,17 +114,30 @@ private:
     size_t size;
 };
 
+struct Buffer {
+    Buffer(const char * const data, size_t len, bool own = false)
+        : data(data)
+        , len(len)
+        , isOwned(own)
+    { }
+
+    Buffer detach(size_t fromIndex = 0) const {
+        if (fromIndex > 0)
+            throw std::invalid_argument("Invalid index (> len)");
+
+        char *newData = new char[len - fromIndex];
+        std::copy(data + fromIndex, data + len - fromIndex, newData);
+
+        return Buffer(newData, len, true);
+    }
+
+    const char* const data;
+    const size_t len;
+    const bool isOwned;
+};
+
 class DynamicStreamBuf : public StreamBuf<char> {
 public:
-    struct Buffer {
-        Buffer(const void *data, size_t len)
-            : data(data)
-            , len(len)
-        { }
-
-        const void* data;
-        const size_t len;
-    };
 
     typedef StreamBuf<char> Base;
     typedef typename Base::traits_type traits_type;