Refactored the event-loop so that it's not coupled with the Listener

anymore.

The event-loop has now been extracted inside its own Io::Service
component which allows us to reuse it.
A new Io::ServiceGroup class has been introduced for multi-threaded
I/O. a ServiceGroup will spawn multiple Io::Service inside their own
thread. Every instance of Io::Service also has its own Io::Handler, the
handler is not shared between io threads.

include/async.h

@@ -470,6 +470,12 @@ namespace Async {
             : core_(core)
         { }
 
+        Resolver(const Resolver& other) = delete;
+        Resolver& operator=(const Resolver& other) = delete;
+
+        Resolver(Resolver&& other) = default;
+        Resolver& operator=(Resolver&& other) = default; 
+
         template<typename Arg>
         bool operator()(Arg&& arg) const {
             typedef typename std::remove_reference<Arg>::type Type;
@@ -537,6 +543,32 @@ namespace Async {
 
     };
 
+    struct Holder {
+        Holder(Resolver resolver, Rejection rejection)
+            : resolver(std::move(resolver))
+            , rejection(std::move(rejection))
+        { }
+
+        template<typename Arg>
+        void resolve(Arg&& arg) const {
+            resolver(std::forward<Arg>(arg));
+        }
+
+        void resolve() const {
+            resolver();
+        }
+
+        template<typename Exc>
+        void reject(Exc exc) const {
+            rejection(std::move(exc));
+        }
+
+    private:
+        Resolver resolver;
+        Rejection rejection;
+    };
+
+
     static constexpr Private::IgnoreException IgnoreException{};
     static constexpr Private::NoExcept NoExcept{};
     static constexpr Private::Throw Throw{};
@@ -817,8 +849,8 @@ namespace Async {
         template<typename ContinuationPolicy>
         struct When {
             When(Resolver resolver, Rejection rejection)
-                : resolve(resolver)
-                , reject(rejection)
+                : resolve(std::move(resolver))
+                , reject(std::move(rejection))
             { }
 
             template<typename... Args>
@@ -1070,7 +1102,7 @@ namespace Async {
             reject = &rejection;
         });
 
-        Impl::When<Impl::All> impl(*resolve, *reject);
+        Impl::When<Impl::All> impl(std::move(*resolve), std::move(*reject));
         // So we capture everything we need inside the lambda and then call the
         // implementation and expand the parameters pack here
         impl(std::forward<Args>(args)...);
@@ -1089,7 +1121,7 @@ namespace Async {
             reject = &rejection;
         });
 
-        Impl::When<Impl::Any> impl(*resolve, *reject);
+        Impl::When<Impl::Any> impl(std::move(*resolve), std::move(*reject));
         impl(std::forward<Args>(args)...);
         return promise;
     }

include/endpoint.h

@@ -38,11 +38,11 @@ public:
     }
 
     void init(const Options& options);
+    void setHandler(const std::shared_ptr<Handler>& handler);
 
     void bind();
     void bind(const Address& addr);
 
-    void setHandler(const std::shared_ptr<Handler>& handler);
     void serve();
     void serveThreaded();
 
@@ -57,6 +57,7 @@ public:
     static Options options();
 
 private:
+
     template<typename Method>
     void serveImpl(Method method)
     {

include/http.h

@@ -18,7 +18,8 @@
 #include "mime.h"
 #include "async.h"
 #include "peer.h"
-#include "io.h"
+#include "tcp.h"
+#include "transport.h"
 
 namespace Net {
 
@@ -139,7 +140,7 @@ public:
     template<typename Duration>
     void arm(Duration duration) {
         Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) {
-            io->armTimer(duration, resolve, reject);
+            transport->io()->armTimer(duration, std::move(resolve), std::move(reject));
         });
 
         p.then(
@@ -155,7 +156,7 @@ public:
     }
 
     void disarm() {
-        io->disarmTimer();
+        transport->io()->disarmTimer();
         armed = false;
     }
 
@@ -164,11 +165,11 @@ public:
     }
 
 private:
-    Timeout(Tcp::IoWorker* io,
+    Timeout(Tcp::Transport* transport,
             Handler* handler,
             const std::shared_ptr<Tcp::Peer>& peer,
             const Request& request)
-        : io(io)
+        : transport(transport)
         , handler(handler)
         , peer(peer)
         , request(request)
@@ -181,7 +182,7 @@ private:
     std::weak_ptr<Tcp::Peer> peer;
     Request request;
 
-    Tcp::IoWorker *io;
+    Tcp::Transport* transport;
     bool armed;
 };
 
@@ -193,14 +194,14 @@ public:
         : Message(std::move(other))
         , peer_(std::move(other.peer_))
         , buf_(std::move(other.buf_))
-        , io_(other.io_)
+        , transport_(other.transport_)
     { }
 
     ResponseStream& operator=(ResponseStream&& other) {
         Message::operator=(std::move(other));
         peer_ = std::move(other.peer_);
         buf_ = std::move(other.buf_);
-        io_ = other.io_;
+        transport_ = other.transport_;
 
         return *this;
     }
@@ -228,7 +229,7 @@ private:
     ResponseStream(
             Message&& other,
             std::weak_ptr<Tcp::Peer> peer,
-            Tcp::IoWorker* io,
+            Tcp::Transport* transport,
             size_t streamSize);
 
     std::shared_ptr<Tcp::Peer> peer() const {
@@ -240,7 +241,7 @@ private:
 
     std::weak_ptr<Tcp::Peer> peer_;
     DynamicStreamBuf buf_;
-    Tcp::IoWorker* io_;
+    Tcp::Transport* transport_;
 };
 
 inline ResponseStream& ends(ResponseStream &stream) {
@@ -286,12 +287,12 @@ public:
         : Message(std::move(other))
         , peer_(other.peer_)
         , buf_(std::move(other.buf_))
-        , io_(other.io_)
+        , transport_(other.transport_)
     { }
     Response& operator=(Response&& other) {
         Message::operator=(std::move(other));
         peer_ = std::move(other.peer_);
-        io_ = other.io_;
+        transport_ = other.transport_;
         buf_ = std::move(other.buf_);
         return *this;
     }
@@ -369,7 +370,7 @@ public:
     ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) {
         code_ = code;
 
-        return ResponseStream(std::move(*this), peer_, io_, streamSize);
+        return ResponseStream(std::move(*this), peer_, transport_, streamSize);
     }
 
     // Unsafe API
@@ -383,10 +384,10 @@ public:
     }
 
 private:
-    Response(Tcp::IoWorker* io)
+    Response(Tcp::Transport* transport)
         : Message()
         , buf_(DefaultStreamSize)
-        , io_(io)
+        , transport_(transport)
     { }
 
     std::shared_ptr<Tcp::Peer> peer() const {
@@ -408,7 +409,7 @@ private:
 
     std::weak_ptr<Tcp::Peer> peer_;
     DynamicStreamBuf buf_;
-    Tcp::IoWorker *io_;
+    Tcp::Transport *transport_;
 };
 
 Async::Promise<ssize_t> serveFile(
@@ -522,7 +523,6 @@ private:
     Private::Parser& getParser(const std::shared_ptr<Tcp::Peer>& peer) const;
 };
 
-
 } // namespace Http
 
 } // namespace Net

include/listener.h

@@ -10,16 +10,19 @@
 #include "net.h"
 #include "os.h"
 #include "flags.h"
+#include "async.h"
 #include "io.h"
 #include <vector>
 #include <memory>
+#include <thread>
+#include <sys/resource.h>
 
 namespace Net {
 
 namespace Tcp {
 
 class Peer;
-class Handler;
+class Transport;
 
 void setSocketOptions(Fd fd, Flags<Options> options);
 
@@ -40,7 +43,8 @@ public:
 
     Listener(const Address& address);
     void init(
-            size_t workers, Flags<Options> options = Options::None,
+            size_t workers,
+            Flags<Options> options = Options::None,
             int backlog = Const::MaxBacklog);
     void setHandler(const std::shared_ptr<Handler>& handler);
 
@@ -62,18 +66,22 @@ public:
     void pinWorker(size_t worker, const CpuSet& set);
 
 private: 
+    struct TransportFactory;
+
     Address addr_; 
     int listen_fd;
     int backlog_;
     NotifyFd shutdownFd;
     Polling::Epoll poller;
 
-    std::vector<std::unique_ptr<IoWorker>> ioGroup;
     Flags<Options> options_;
-    std::shared_ptr<Handler> handler_;
     std::unique_ptr<std::thread> acceptThread;
 
-    void shutdownIo();
+    size_t workers_;
+    Io::ServiceGroup io_;
+    std::shared_ptr<Transport> transport_;
+    std::shared_ptr<Handler> handler_;
+
     void handleNewConnection();
     void dispatchPeer(const std::shared_ptr<Peer>& peer);
 

include/mailbox.h

@@ -14,6 +14,7 @@
 #include <sys/eventfd.h>
 #include <unistd.h>
 
+
 template<typename T>
 class Mailbox {
 public:
@@ -147,10 +148,14 @@ public:
     struct Entry {
         friend class Queue;
 
-        T data() const {
+        const T& data() const {
             return *reinterpret_cast<const T*>(&storage);
         }
 
+        T& data() {
+            return *reinterpret_cast<T*>(&storage);
+        }
+
         ~Entry() {
             auto *d = reinterpret_cast<T *>(&storage);
             d->~T();
@@ -194,7 +199,7 @@ public:
         if (next) {
             // Since it's Single-Consumer, the store does not need to be atomic
             tail = next;
-            new (&res->storage) T(next->data());
+            new (&res->storage) T(std::move(next->data()));
             return res;
         }
         return nullptr;

include/os.h

@@ -6,6 +6,7 @@
 
 #pragma once
 
+#include <memory>
 #include <chrono>
 #include <vector>
 #include <bitset>
@@ -120,6 +121,7 @@ public:
     { }
 
     Polling::Tag bind(Polling::Epoll& poller);
+
     bool isBound() const;
 
     Polling::Tag tag() const;

include/peer.h

@@ -19,11 +19,11 @@ namespace Net {
 
 namespace Tcp {
 
-class IoWorker;
+class Transport;
 
 class Peer {
 public:
-    friend class IoWorker;
+    friend class Transport;
 
     Peer();
     Peer(const Address& addr);
@@ -54,7 +54,10 @@ public:
     Async::Promise<ssize_t> send(const Buffer& buffer, int flags = 0);
 
 private:
-    IoWorker* io_;
+    void associateTransport(Transport* transport);
+    Transport* transport() const;
+
+    Transport* transport_;
 
     Address addr;
     std::string hostname_;

include/prototype.h

+/* 
+   Mathieu Stefani, 28 janvier 2016
+   
+   Simple Prototype design pattern implement
+*/
+
+#pragma once
+
+#include <type_traits>
+#include <memory>
+
+/* In a sense, a Prototype is just a class that provides a clone() method */
+template<typename Class>
+struct Prototype {
+    virtual std::shared_ptr<Class> clone() const = 0;
+};
+
+template<typename Base, typename Class>
+struct Prototypable {
+
+};
+
+#define PROTOTYPE_OF(Base, Class)                                        \
+private:                                                                 \
+    std::shared_ptr<Base> clone() const {                                \
+        return std::make_shared<Class>();                                \
+    }                                                                    \
+public:
+
+#define HTTP_PROTOTYPE(Class) \
+    PROTOTYPE_OF(Net::Tcp::Handler, Class)

include/router.h

@@ -139,6 +139,10 @@ namespace Private {
                 Http::Timeout timeout);
 
     private:
+        std::shared_ptr<Net::Tcp::Handler> clone() const {
+            return std::make_shared<HttpHandler>(routes);
+        }
+
         std::unordered_map<Http::Method, std::vector<Route>> routes;
     };
 }

include/tcp.h

@@ -7,6 +7,7 @@
 #pragma once
 
 #include "flags.h"
+#include "prototype.h"
 #include <memory>
 
 namespace Net {
@@ -14,6 +15,7 @@ namespace Net {
 namespace Tcp {
 
 class Peer;
+class Transport;
 
 enum class Options : uint64_t {
     None                 = 0,
@@ -28,11 +30,9 @@ enum class Options : uint64_t {
 
 DECLARE_FLAGS_OPERATORS(Options)
 
-class IoWorker;
-
-class Handler {
+class Handler : private Prototype<Handler> {
 public:
-    friend class IoWorker;
+    friend class Transport;
 
     Handler();
     ~Handler();
@@ -43,10 +43,15 @@ public:
     virtual void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
 
 protected:
-    IoWorker *io() { return io_; }
+    Transport *transport() {
+        if (!transport_)
+            throw std::logic_error("Orphaned handler");
+        return transport_;
+     }
 
 private:
-    IoWorker *io_;
+    void associateTransport(Transport* transport);
+    Transport *transport_;
 };
 
 } // namespace Tcp

include/transport.h

+/* 
+   Mathieu Stefani, 26 janvier 2016
+   
+   Transport TCP layer
+*/
+
+#pragma once
+
+#include "io.h"
+#include "optional.h"
+#include "async.h"
+#include "stream.h"
+
+namespace Net {
+
+namespace Tcp {
+
+class Peer;
+class Handler;
+
+class Transport : public Io::Handler {
+public:
+    Transport(const std::shared_ptr<Tcp::Handler>& handler);
+
+    void init(const std::shared_ptr<Tcp::Handler>& handler);
+
+    void registerPoller(Polling::Epoll& poller);
+
+    void handleNewPeer(const std::shared_ptr<Peer>& peer);
+    void onReady(const Io::FdSet& fds);
+
+    template<typename Buf>
+    Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) {
+        // 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() != io()->thread()) {
+            return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
+                BufferHolder holder(buffer);
+                auto detached = holder.detach();
+                OnHoldWrite write(std::move(resolve), std::move(reject), detached, flags);
+                write.peerFd = fd;
+                auto *e = writesQueue.allocEntry(std::move(write));
+                writesQueue.push(e);
+            });
+        }
+        return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) {
+
+            auto it = toWrite.find(fd);
+            if (it != std::end(toWrite)) {
+                reject(Net::Error("Multiple writes on the same fd"));
+                return;
+            }
+
+            asyncWriteImpl(fd, flags, BufferHolder(buffer), std::move(resolve), std::move(reject));
+
+        });
+    }
+
+    std::shared_ptr<Io::Handler> clone() const;
+
+private:
+    enum WriteStatus {
+        FirstTry,
+        Retry
+    };
+
+    struct BufferHolder {
+        enum Type { Raw, File };
+
+        explicit BufferHolder(const Buffer& buffer)
+            : type(Raw)
+            , u(buffer)
+        {
+            size_ = buffer.len;
+        }
+
+        explicit BufferHolder(const FileBuffer& buffer)
+            : type(File)
+            , u(buffer.fd())
+        {
+            size_ = buffer.size();
+        }
+
+        bool isFile() const { return type == File; }
+        bool isRaw() const { return type == Raw; }
+        size_t size() const { return size_; }
+
+        Fd fd() const {
+            if (!isFile())
+                throw std::runtime_error("Tried to retrieve fd of a non-filebuffer");
+
+            return u.fd;
+
+        }
+
+        Buffer raw() const {
+            if (!isRaw())
+                throw std::runtime_error("Tried to retrieve raw data of a non-buffer");
+
+            return u.raw;
+        }
+
+        BufferHolder detach(size_t offset = 0) const {
+            if (!isRaw())
+                return BufferHolder(u.fd, size_);
+
+            if (u.raw.isOwned)
+                return BufferHolder(u.raw);
+
+            auto detached = u.raw.detach(offset);
+            return BufferHolder(detached);
+        }
+
+    private:
+        BufferHolder(Fd fd, size_t size)
+         : u(fd)
+         , size_(size)
+         , type(File)
+        { }
+
+        union U {
+            Buffer raw;
+            Fd fd;
+
+            U(Buffer buffer) : raw(buffer) { }
+            U(Fd fd) : fd(fd) { }
+        } u;
+        size_t size_;
+        Type type;
+    };
+
+    struct OnHoldWrite {
+        OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
+                    BufferHolder buffer, int flags = 0)
+            : resolve(std::move(resolve))
+            , reject(std::move(reject))
+            , buffer(std::move(buffer))
+            , flags(flags)
+            , peerFd(-1)
+        { }
+
+        Async::Resolver resolve;
+        Async::Rejection reject;
+        BufferHolder buffer;
+        int flags;
+        Fd peerFd;
+    };
+
+    mutable std::mutex peersMutex;
+    std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
+    /* @Incomplete: this should be a std::dequeue.
+        If an asyncWrite on a particular fd is initiated whereas the fd is not write-ready
+        yet and some writes are still on-hold, writes should queue-up so that when the
+        fd becomes ready again, we can write everything
+    */
+    std::unordered_map<Fd, OnHoldWrite> toWrite;
+    PollableQueue<OnHoldWrite> writesQueue;
+
+    std::shared_ptr<Tcp::Handler> handler_;
+
+    std::shared_ptr<Peer>& getPeer(Fd fd);
+    std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
+
+    void asyncWriteImpl(Fd fd, OnHoldWrite& entry, WriteStatus status = FirstTry);
+    void asyncWriteImpl(
+            Fd fd, int flags, const BufferHolder& 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 handleWriteQueue();
+};
+
+} // namespace Tcp
+
+} // namespace Net

main.cc

@@ -32,6 +32,9 @@ void printCookies(const Net::Http::Request& req) {
 }
 
 class MyHandler : public Net::Http::Handler {
+
+    HTTP_PROTOTYPE(MyHandler)
+
     void onRequest(
             const Net::Http::Request& req,
             Net::Http::Response response,
@@ -60,6 +63,9 @@ class MyHandler : public Net::Http::Handler {
                     stream << "NG";
                     stream << ends;
                 }
+                else {
+                    response.send(Net::Http::Code::Ok, "PONG");
+                }
 
             }
         }
@@ -97,6 +103,7 @@ class MyHandler : public Net::Http::Handler {
             .send(Net::Http::Code::Request_Timeout, "Timeout")
             .then([=](ssize_t) { }, ExceptionPrinter());
     }
+
 };
 
 struct LoadMonitor {
@@ -161,6 +168,7 @@ void handleReady(const Rest::Request&, Http::Response response) {
 
 }
 
+#if 0
 class StatsEndpoint {
 public:
     StatsEndpoint(Net::Address addr)
@@ -168,9 +176,9 @@ public:
         , monitor(httpEndpoint)
     { }
 
-    void init() {
+    void init(size_t thr = 2) {
         auto opts = Net::Http::Endpoint::options()
-            .threads(1)
+            .threads(thr)
             .flags(Net::Tcp::Options::InstallSignalHandler);
         httpEndpoint->init(opts);
         setupRoutes();
@@ -179,7 +187,8 @@ public:
 
     void start() {
         monitor.start();
-        httpEndpoint->setHandler(router.handler());
+        //httpEndpoint->setHandler(router.handler());
+        httpEndpoint->setHandler(std::make_shared<MyHandler>());
         httpEndpoint->serve();
     }
 
@@ -276,6 +285,7 @@ private:
     Rest::Router router;
     LoadMonitor monitor;
 };
+#endif
 
 int main(int argc, char *argv[]) {
     Net::Port port(9080);
@@ -295,11 +305,31 @@ int main(int argc, char *argv[]) {
     cout << "Cores = " << hardware_concurrency() << endl;
     cout << "Using " << thr << " threads" << endl;
 
+#if 0
     StatsEndpoint stats(addr);
 
-    stats.init();
+    stats.init(thr);
     stats.start();
+#endif
+
+    auto server = std::make_shared<Net::Http::Endpoint>(addr);
+
+    auto opts = Net::Http::Endpoint::options()
+        .threads(thr)
+        .flags(Net::Tcp::Options::InstallSignalHandler);
+    server->init(opts);
+    server->setHandler(std::make_shared<MyHandler>());
+
+    LoadMonitor monitor(server);
+    monitor.setInterval(std::chrono::seconds(1));
+    monitor.start();
+
+    server->serve();
 
     std::cout << "Shutdowning server" << std::endl;
+#if 0
     stats.shutdown();
+#endif
+    server->shutdown();
+    monitor.shutdown();
 }

src/common/http.cc

@@ -17,6 +17,7 @@
 #include "http.h"
 #include "net.h"
 #include "peer.h"
+#include "transport.h"
 
 using namespace std;
 
@@ -448,12 +449,12 @@ Request::peer() const {
 ResponseStream::ResponseStream(
         Message&& other,
         std::weak_ptr<Tcp::Peer> peer,
-        Tcp::IoWorker* io,
+        Tcp::Transport* transport,
         size_t streamSize)
     : Message(std::move(other))
     , peer_(std::move(peer))
     , buf_(streamSize)
-    , io_(io)
+    , transport_(transport)
 {
     if (!writeStatusLine(code_, buf_))
         throw Error("Response exceeded buffer size");
@@ -472,11 +473,11 @@ ResponseStream::ResponseStream(
 
 void
 ResponseStream::flush() {
-    io_->disarmTimer();
+    transport_->io()->disarmTimer();
     auto buf = buf_.buffer();
 
     auto fd = peer()->fd();
-    io_->asyncWrite(fd, buf);
+    transport_->asyncWrite(fd, buf);
 
     buf_.clear();
 }
@@ -522,12 +523,12 @@ Response::putOnWire(const char* data, size_t len)
 
         auto buffer = buf_.buffer();
 
-        io_->disarmTimer();
+        transport_->io()->disarmTimer();
 
 #undef OUT
 
         auto fd = peer()->fd();
-        return io_->asyncWrite(fd, buffer);
+        return transport_->asyncWrite(fd, buffer);
 
     } catch (const std::runtime_error& e) {
         return Async::Promise<ssize_t>::rejected(e);
@@ -590,13 +591,13 @@ serveFile(Response& response, const char* fileName, const Mime::MediaType& conte
 
     OUT(os << crlf);
 
-    auto *io = response.io_;
+    auto *transport = response.transport_;
     auto peer = response.peer();
     auto sockFd = peer->fd();
 
     auto buffer = buf->buffer();
-    return io->asyncWrite(sockFd, buffer, MSG_MORE).then([=](ssize_t bytes) {
-        return io->asyncWrite(sockFd, FileBuffer(fileName));
+    return transport->asyncWrite(sockFd, buffer, MSG_MORE).then([=](ssize_t bytes) {
+        return transport->asyncWrite(sockFd, FileBuffer(fileName));
     }, Async::Throw);
 
 #undef OUT
@@ -613,10 +614,10 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer
 
         auto state = parser.parse();
         if (state == Private::State::Done) {
-            Response response(io());
+            Response response(transport());
             response.associatePeer(peer);
 
-            Timeout timeout(io(), this, peer, parser.request);
+            Timeout timeout(transport(), this, peer, parser.request);
 
 #ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
             parser.request.associatePeer(peer);
@@ -625,13 +626,13 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer
             parser.reset();
         }
     } catch (const HttpError &err) {
-        Response response(io());
+        Response response(transport());
         response.associatePeer(peer);
         response.send(static_cast<Code>(err.code()), err.reason());
         getParser(peer).reset();
     }
     catch (const std::exception& e) {
-        Response response(io());
+        Response response(transport());
         response.associatePeer(peer);
         response.send(Code::Internal_Server_Error, e.what());
         getParser(peer).reset();
@@ -655,7 +656,7 @@ void
 Timeout::onTimeout(uint64_t numWakeup) {
     if (!peer.lock()) return;
 
-    Response response(io);
+    Response response(transport);
     response.associatePeer(peer);
 
     handler->onTimeout(request, std::move(response));

src/common/os.cc

@@ -5,6 +5,7 @@
 
 #include "os.h"
 #include "common.h"
+#include "io.h"
 #include <unistd.h>
 #include <fcntl.h>
 #include <fstream>
@@ -244,7 +245,7 @@ NotifyFd::bind(Polling::Epoll& poller) {
     Polling::Tag tag(event_fd);
 
     poller.addFd(event_fd, Polling::NotifyOn::Read, tag, Polling::Mode::Edge);
-    return Polling::Tag(event_fd);
+    return tag;
 }
 
 bool

src/common/peer.cc

@@ -5,7 +5,7 @@
 
 #include "peer.h"
 #include "async.h"
-#include "io.h"
+#include "transport.h"
 #include <iostream>
 #include <stdexcept>
 #include <sys/socket.h>
@@ -18,11 +18,13 @@ using namespace std;
 
 Peer::Peer()
     : fd_(-1)
+    , transport_(nullptr)
 { }
 
 Peer::Peer(const Address& addr)
     : addr(addr)
     , fd_(-1)
+    , transport_(nullptr)
 { }
 
 Address
@@ -79,7 +81,7 @@ Peer::tryGetData(std::string(name)) const {
 
 Async::Promise<ssize_t>
 Peer::send(const Buffer& buffer, int flags) {
-    return io_->asyncWrite(fd_, buffer, flags);
+    return transport()->asyncWrite(fd_, buffer, flags);
 }
 
 std::ostream& operator<<(std::ostream& os, const Peer& peer) {
@@ -88,6 +90,19 @@ std::ostream& operator<<(std::ostream& os, const Peer& peer) {
     return os;
 }
 
+void
+Peer::associateTransport(Transport* transport) {
+    transport_ = transport;
+}
+
+Transport*
+Peer::transport() const {
+    if (!transport_)
+        throw std::logic_error("Orphaned peer");
+
+    return transport_;
+}
+
 } // namespace Tcp
 
 } // namespace Net

src/common/tcp.cc

@@ -12,11 +12,17 @@ namespace Net {
 namespace Tcp {
 
 Handler::Handler()
+    : transport_(nullptr)
 { }
 
 Handler::~Handler()
 { }
 
+void
+Handler::associateTransport(Transport* transport) {
+    transport_ = transport;
+}
+
 void
 Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
 }

src/common/transport.cc

+#include "transport.h"
+#include "peer.h"
+#include "tcp.h"
+#include "os.h"
+#include <sys/sendfile.h>
+
+using namespace Polling;
+
+namespace Net {
+
+namespace Tcp {
+
+Transport::Transport(const std::shared_ptr<Tcp::Handler>& handler) {
+    init(handler);
+}
+
+void
+Transport::init(const std::shared_ptr<Tcp::Handler>& handler) {
+    handler_ = handler;
+    handler_->associateTransport(this);
+}
+
+std::shared_ptr<Io::Handler>
+Transport::clone() const {
+    return std::make_shared<Transport>(handler_->clone());
+}
+
+void
+Transport::registerPoller(Polling::Epoll& poller) {
+    writesQueue.bind(poller);
+}
+
+void
+Transport::handleNewPeer(const std::shared_ptr<Tcp::Peer>& peer) {
+    int fd = peer->fd();
+    {
+        std::unique_lock<std::mutex> guard(peersMutex);
+        peers.insert(std::make_pair(fd, peer));
+    }
+
+    peer->associateTransport(this);
+
+    handler_->onConnection(peer);
+    io()->registerFd(fd, NotifyOn::Read | NotifyOn::Shutdown, Polling::Mode::Edge);
+}
+
+void
+Transport::onReady(const Io::FdSet& fds) {
+    for (const auto& entry: fds) {
+        if (entry.getTag() == writesQueue.tag()) {
+            handleWriteQueue();
+        }
+
+        else if (entry.isReadable()) {
+            auto& peer = getPeer(entry.getTag());
+            handleIncoming(peer);
+        }
+        else if (entry.isWritable()) {
+            auto tag = entry.getTag();
+            auto fd = tag.value();
+
+            auto it = toWrite.find(fd);
+            if (it == std::end(toWrite)) {
+                throw std::runtime_error("Assertion Error: could not find write data");
+            }
+
+            io()->modifyFd(fd, NotifyOn::Read, Polling::Mode::Edge);
+
+            auto& write = it->second;
+            asyncWriteImpl(fd, write, Retry);
+        }
+    }
+}
+
+void
+Transport::handleIncoming(const std::shared_ptr<Peer>& peer) {
+    char buffer[Const::MaxBuffer];
+    memset(buffer, 0, sizeof buffer);
+
+    ssize_t totalBytes = 0;
+    int fd = peer->fd();
+
+    for (;;) {
+
+        ssize_t bytes;
+
+        bytes = recv(fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
+        if (bytes == -1) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                if (totalBytes > 0) {
+                    handler_->onInput(buffer, totalBytes, peer);
+                }
+            } else {
+                if (errno == ECONNRESET) {
+                    handlePeerDisconnection(peer);
+                }
+                else {
+                    throw std::runtime_error(strerror(errno));
+                }
+            }
+            break;
+        }
+        else if (bytes == 0) {
+            handlePeerDisconnection(peer);
+            break;
+        }
+
+        else {
+            totalBytes += bytes;
+            if (totalBytes >= Const::MaxBuffer) {
+                std::cerr << "Too long packet" << std::endl;
+                break;
+            }
+        }
+    }
+}
+
+void
+Transport::handlePeerDisconnection(const std::shared_ptr<Peer>& peer) {
+    handler_->onDisconnection(peer);
+
+    int fd = peer->fd();
+    {
+        std::unique_lock<std::mutex> guard(peersMutex);
+        auto it = peers.find(fd);
+        if (it == std::end(peers))
+            throw std::runtime_error("Could not find peer to erase");
+
+        peers.erase(it);
+    }
+
+    close(fd);
+}
+
+void
+Transport::asyncWriteImpl(Fd fd, Transport::OnHoldWrite& entry, WriteStatus status) {
+    asyncWriteImpl(fd, entry.flags, entry.buffer, std::move(entry.resolve), std::move(entry.reject), status);
+}
+
+void
+Transport::asyncWriteImpl(
+        Fd fd, int flags, const BufferHolder& buffer,
+        Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
+{
+    auto cleanUp = [&]() {
+        if (buffer.isRaw()) {
+            auto raw = buffer.raw();
+            if (raw.isOwned) delete[] raw.data;
+        }
+
+        if (status == Retry)
+            toWrite.erase(fd);
+    };
+
+    ssize_t totalWritten = 0;
+    for (;;) {
+        ssize_t bytesWritten = 0;
+        auto len = buffer.size() - totalWritten;
+        if (buffer.isRaw()) {
+            auto raw = buffer.raw();
+            auto ptr = raw.data + totalWritten;
+            bytesWritten = ::send(fd, ptr, len, flags);
+        } else {
+            auto file = buffer.fd();
+            off_t offset = totalWritten;
+            bytesWritten = ::sendfile(fd, file, &offset, len);
+        }
+        if (bytesWritten < 0) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                if (status == FirstTry) {
+                    toWrite.insert(
+                            std::make_pair(fd,
+                                OnHoldWrite(std::move(resolve), std::move(reject), buffer.detach(totalWritten), flags)));
+                }
+                io()->modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
+            }
+            else {
+                cleanUp();
+                reject(Net::Error::system("Could not write data"));
+            }
+            break;
+        }
+        else {
+            totalWritten += bytesWritten;
+            if (totalWritten == len) {
+                cleanUp();
+                resolve(totalWritten);
+                break;
+            }
+        }
+    }
+}
+
+void
+Transport::handleWriteQueue() {
+    // Let's drain the queue
+    for (;;) {
+        std::unique_ptr<PollableQueue<OnHoldWrite>::Entry> entry(writesQueue.pop());
+        if (!entry) break;
+
+        auto &write = entry->data();
+        asyncWriteImpl(write.peerFd, write);
+    }
+}
+
+std::shared_ptr<Peer>&
+Transport::getPeer(Fd fd)
+{
+    std::unique_lock<std::mutex> guard(peersMutex);
+    auto it = peers.find(fd);
+    if (it == std::end(peers))
+    {
+        throw std::runtime_error("No peer found for fd: " + std::to_string(fd));
+    }
+    return it->second;
+}
+
+std::shared_ptr<Peer>&
+Transport::getPeer(Polling::Tag tag)
+{
+    return getPeer(tag.value());
+}
+
+} // namespace Tcp
+
+} // namespace Net

src/server/endpoint.cc

@@ -13,6 +13,7 @@ namespace Net {
 
 namespace Http {
 
+
 Endpoint::Options::Options()
     : threads_(1)
 { }

src/server/listener.cc

@@ -13,12 +13,15 @@
 #include <sys/epoll.h>
 #include <pthread.h>
 #include <signal.h>
+#include <sys/timerfd.h>
+#include <sys/sendfile.h>
 #include <cassert>
 #include <cstring>
 #include "listener.h"
 #include "peer.h"
 #include "common.h"
 #include "os.h"
+#include "transport.h"
 
 using namespace std;
 
@@ -87,7 +90,9 @@ Listener::~Listener() {
 }
 
 void
-Listener::init(size_t workers, Flags<Options> options, int backlog)
+Listener::init(
+    size_t workers,
+    Flags<Options> options, int backlog)
 {
     if (workers > hardware_concurrency()) {
         // Log::warning() << "More workers than available cores"
@@ -102,21 +107,19 @@ Listener::init(size_t workers, Flags<Options> options, int backlog)
         }
     }
 
-    for (size_t i = 0; i < workers; ++i) {
-        auto wrk = std::unique_ptr<IoWorker>(new IoWorker);
-        ioGroup.push_back(std::move(wrk));
-    }
+    workers_ = workers;
+
 }
 
 void
-Listener::setHandler(const std::shared_ptr<Handler>& handler)
-{
+Listener::setHandler(const std::shared_ptr<Handler>& handler) {
     handler_ = handler;
 }
 
 void
 Listener::pinWorker(size_t worker, const CpuSet& set)
 {
+#if 0
     if (ioGroup.empty()) {
         throw std::domain_error("Invalid operation, did you call init() before ?");
     }
@@ -126,6 +129,7 @@ Listener::pinWorker(size_t worker, const CpuSet& set)
 
     auto &wrk = ioGroup[worker];
     wrk->pin(set);
+#endif
 }
 
 bool
@@ -135,10 +139,8 @@ Listener::bind() {
 
 bool
 Listener::bind(const Address& address) {
-    if (ioGroup.empty()) {
-        throw std::runtime_error("Call init() before calling bind()");
-    }
-
+    if (!handler_)
+        throw std::runtime_error("Call setHandler before calling bind()");
     addr_ = address;
 
     struct addrinfo hints;
@@ -183,9 +185,10 @@ Listener::bind(const Address& address) {
     listen_fd = fd;
     g_listen_fd = fd;
 
-    for (auto& io: ioGroup) {
-        io->start(handler_, options_);
-    }
+    transport_.reset(new Transport(handler_));
+
+    io_.init(workers_, transport_);
+    io_.start();
 
     return true;
 }
@@ -230,24 +233,12 @@ Listener::runThreaded() {
 void
 Listener::shutdown() {
     if (shutdownFd.isBound()) shutdownFd.notify();
-    shutdownIo();
-}
-
-void
-Listener::shutdownIo() {
-    for (auto &worker: ioGroup) {
-        worker->shutdown();
-    }
+    io_.shutdown();
 }
 
 Async::Promise<Listener::Load>
 Listener::requestLoad(const Listener::Load& old) {
-    std::vector<Async::Promise<rusage>> loads;
-    loads.reserve(ioGroup.size());
-
-    for (const auto& io: ioGroup) {
-        loads.push_back(io->getLoad());
-    }
+    auto loads = io_.load();
 
     return Async::whenAll(std::begin(loads), std::end(loads)).then([=](const std::vector<rusage>& usages) {
 
@@ -256,7 +247,7 @@ Listener::requestLoad(const Listener::Load& old) {
 
         if (old.raw.empty()) {
             res.global = 0.0;
-            for (size_t i = 0; i < ioGroup.size(); ++i) res.workers.push_back(0.0);
+            for (size_t i = 0; i < io_.size(); ++i) res.workers.push_back(0.0);
         } else {
 
             auto totalElapsed = [](rusage usage) {
@@ -319,10 +310,10 @@ Listener::handleNewConnection() {
 
 void
 Listener::dispatchPeer(const std::shared_ptr<Peer>& peer) {
-    const size_t workers = ioGroup.size();
-    size_t worker = peer->fd() % workers;
+    auto service = io_.service(peer->fd());
+    auto transport = std::static_pointer_cast<Transport>(service->handler());
 
-    ioGroup[worker]->handleNewPeer(peer);
+    transport->handleNewPeer(peer);
 
 }
 

tests/async_test.cc

@@ -7,10 +7,10 @@ Async::Promise<int> doAsync(int N)
 {
     Async::Promise<int> promise(
         [=](Async::Resolver& resolve, Async::Rejection& reject) {
-            std::thread thr([=]() mutable {
+            std::thread thr([=](Async::Resolver resolve) mutable {
                 std::this_thread::sleep_for(std::chrono::seconds(1));
                 resolve(N * 2);
-            });
+            }, std::move(resolve));
 
             thr.detach();
     });
@@ -23,10 +23,10 @@ Async::Promise<T> doAsyncTimed(std::chrono::seconds time, T val, Func func)
 {
     Async::Promise<T> promise(
         [=](Async::Resolver& resolve, Async::Rejection& reject) {
-            std::thread thr([=]() mutable {
+            std::thread thr([=](Async::Resolver resolve) mutable {
                 std::this_thread::sleep_for(time);
                 resolve(func(val));
-            });
+            }, std::move(resolve));
 
             thr.detach();
     });