Http::Endpoint: added a threaded serving mode that will spawn an acceptor...

Http::Endpoint: added a threaded serving mode that will spawn an acceptor thread rather than running in the same thread

main.cc

@@ -93,7 +93,7 @@ struct LoadMonitor {
 
     ~LoadMonitor() {
         shutdown_ = true;
-        thread->join();
+        if (thread) thread->join();
     }
 
 private:
@@ -156,6 +156,9 @@ int main(int argc, char *argv[]) {
     monitor.setInterval(std::chrono::seconds(5));
     monitor.start();
 
-    server->serve();
+    server->serveThreaded();
+    std::this_thread::sleep_for(std::chrono::seconds(10));
+    std::cout << "Shutdowning server" << std::endl;
+    server->shutdown();
     monitor.shutdown();
 }

src/http.cc

@@ -587,16 +587,19 @@ Endpoint::setHandler(const std::shared_ptr<Handler>& handler) {
 void
 Endpoint::serve()
 {
-    if (!handler_)
-        throw std::runtime_error("Must call setHandler() prior to serve()");
+    serveImpl(&Tcp::Listener::run);
+}
 
-    listener.setHandler(handler_);
+void
+Endpoint::serveThreaded()
+{
+    serveImpl(&Tcp::Listener::runThreaded);
+}
 
-    if (listener.bind()) { 
-        const auto& addr = listener.address();
-        cout << "Now listening on " << "http://" + addr.host() << ":" << addr.port() << endl;
-        listener.run();
-    }
+void
+Endpoint::shutdown()
+{
+    listener.shutdown();
 }
 
 Async::Promise<Tcp::Listener::Load>

src/http.h

@@ -530,6 +530,9 @@ public:
 
     void setHandler(const std::shared_ptr<Handler>& handler);
     void serve();
+    void serveThreaded();
+
+    void shutdown();
 
     bool isBound() const {
         return listener.isBound();
@@ -540,6 +543,24 @@ public:
     static Options options();
 
 private:
+    template<typename Method>
+    void serveImpl(Method method)
+    {
+#define CALL_MEMBER_FN(obj, pmf)  ((obj).*(pmf))
+        if (!handler_)
+            throw std::runtime_error("Must call setHandler() prior to serve()");
+
+        listener.setHandler(handler_);
+
+        if (listener.bind()) {
+            const auto& addr = listener.address();
+            std::cout << "Now listening on " << "http://" + addr.host() << ":"
+                      << addr.port() << std::endl;
+            CALL_MEMBER_FN(listener, method)();
+        }
+#undef CALL_MEMBER_FN
+    }
+
     std::shared_ptr<Handler> handler_;
     Net::Tcp::Listener listener;
 };

src/io.cc

@@ -38,6 +38,8 @@ To *message_cast(const std::unique_ptr<Message>& from)
 IoWorker::IoWorker() {
     timerFd = TRY_RET(timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK));
     poller.addFd(timerFd, Polling::NotifyOn::Read, Polling::Tag(timerFd));
+
+    notifier.bind(poller);
 }
 
 IoWorker::~IoWorker() {
@@ -234,9 +236,6 @@ IoWorker::run() {
     mailbox.bind(poller);
     writesQueue.bind(poller);
 
-    auto n = notifier.tag();
-    poller.addFd(n.value(), NotifyOn::Read, n, Polling::Mode::Edge);
-
     std::chrono::milliseconds timeout(-1);
 
     for (;;) {

src/listener.cc

@@ -29,12 +29,17 @@ namespace Tcp {
 namespace {
     volatile sig_atomic_t g_listen_fd = -1;
 
-    void handle_sigint(int) {
+    void closeListener() {
         if (g_listen_fd != -1) {
-            close(g_listen_fd);
+            ::close(g_listen_fd);
             g_listen_fd = -1;
         }
     }
+
+    void handle_sigint(int) {
+        std::cout << "Received C-c, closing listen fd!" << std::endl;
+        closeListener();
+    }
 }
 
 using Polling::NotifyOn;
@@ -76,6 +81,11 @@ Listener::Listener(const Address& address)
 {
 }
 
+Listener::~Listener() {
+    shutdown();
+    if (acceptThread) acceptThread->join();
+}
+
 void
 Listener::init(size_t workers, Flags<Options> options, int backlog)
 {
@@ -168,7 +178,8 @@ Listener::bind(const Address& address) {
         TRY(::listen(fd, backlog_));
     }
 
-
+    make_non_blocking(fd);
+    poller.addFd(fd, Polling::NotifyOn::Write, Polling::Tag(fd), Polling::Mode::Edge);
     listen_fd = fd;
     g_listen_fd = fd;
 
@@ -187,31 +198,43 @@ Listener::isBound() const {
 void
 Listener::run() {
     for (;;) {
-        struct sockaddr_in peer_addr;
-        socklen_t peer_addr_len = sizeof(peer_addr);
-        int client_fd = ::accept(listen_fd, (struct sockaddr *)&peer_addr, &peer_addr_len);
-        if (client_fd < 0) {
-            if (g_listen_fd == -1) {
-                cout << "SIGINT Signal received, shutdowning !" << endl;
-                shutdown();
-                break;
-
-            } else {
-                throw std::runtime_error(strerror(errno));
+        std::vector<Polling::Event> events;
+
+        int ready_fds = poller.poll(events, 128, std::chrono::milliseconds(-1));
+        if (ready_fds == -1) {
+            if (errno == EINTR && g_listen_fd == -1) return;
+            throw Error::system("Polling");
+        }
+        else if (ready_fds > 0) {
+            for (const auto& event: events) {
+                if (event.tag == shutdownFd.tag())
+                    return;
+                else {
+                    if (event.flags.hasFlag(Polling::NotifyOn::Write)) {
+                        auto fd = event.tag.value();
+                        if (fd == listen_fd)
+                            handleNewConnection();
+                    }
+                }
             }
         }
-
-        make_non_blocking(client_fd);
-
-        auto peer = make_shared<Peer>(Address::fromUnix((struct sockaddr *)&peer_addr));
-        peer->associateFd(client_fd);
-
-        dispatchPeer(peer);
     }
 }
 
+void
+Listener::runThreaded() {
+    shutdownFd.bind(poller);
+    acceptThread.reset(new std::thread([=]() { this->run(); }));
+}
+
 void
 Listener::shutdown() {
+    if (shutdownFd.isBound()) shutdownFd.notify();
+    shutdownIo();
+}
+
+void
+Listener::shutdownIo() {
     for (auto &worker: ioGroup) {
         worker->shutdown();
     }
@@ -277,6 +300,23 @@ Listener::options() const {
     return options_;
 }
 
+void
+Listener::handleNewConnection() {
+    struct sockaddr_in peer_addr;
+    socklen_t peer_addr_len = sizeof(peer_addr);
+    int client_fd = ::accept(listen_fd, (struct sockaddr *)&peer_addr, &peer_addr_len);
+    if (client_fd < 0) {
+        throw std::runtime_error(strerror(errno));
+    }
+
+    make_non_blocking(client_fd);
+
+    auto peer = make_shared<Peer>(Address::fromUnix((struct sockaddr *)&peer_addr));
+    peer->associateFd(client_fd);
+
+    dispatchPeer(peer);
+}
+
 void
 Listener::dispatchPeer(const std::shared_ptr<Peer>& peer) {
     const size_t workers = ioGroup.size();

src/listener.h

@@ -36,6 +36,7 @@ public:
     };
 
     Listener();
+    ~Listener();
 
     Listener(const Address& address);
     void init(
@@ -49,6 +50,8 @@ public:
     bool isBound() const;
 
     void run();
+    void runThreaded();
+
     void shutdown();
 
     Async::Promise<Load> requestLoad(const Load& old);
@@ -62,12 +65,18 @@ private:
     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();
+    void handleNewConnection();
     void dispatchPeer(const std::shared_ptr<Peer>& peer);
-    void runLoadThread();
+
 };
 
 } // namespace Tcp

src/net.cc

@@ -119,7 +119,7 @@ Error::system(const char* message) {
     const char *err = strerror(errno);
 
     std::string str(message);
-    str += ':';
+    str += ": ";
     str += err;
 
     return Error(std::move(str));

src/os.cc

@@ -238,8 +238,18 @@ namespace Polling {
 
 } // namespace Poller
 
-NotifyFd::NotifyFd() {
+Polling::Tag
+NotifyFd::bind(Polling::Epoll& poller) {
     event_fd = TRY_RET(eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC));
+    Polling::Tag tag(event_fd);
+
+    poller.addFd(event_fd, Polling::NotifyOn::Read, tag, Polling::Mode::Edge);
+    return Polling::Tag(event_fd);
+}
+
+bool
+NotifyFd::isBound() const {
+    return event_fd != -1;
 }
 
 Polling::Tag
@@ -249,12 +259,16 @@ NotifyFd::tag() const {
 
 void
 NotifyFd::notify() const {
+    if (!isBound())
+        throw std::runtime_error("Can not notify an unbound fd");
     eventfd_t val = 1;
     TRY(eventfd_write(event_fd, val));
 }
 
 void
 NotifyFd::read() const {
+    if (!isBound())
+        throw std::runtime_error("Can not read an unbound fd");
     eventfd_t val;
     TRY(eventfd_read(event_fd, &val));
 }

src/os.h

@@ -115,7 +115,12 @@ private:
 
 class NotifyFd {
 public:
-    NotifyFd();
+    NotifyFd()
+        : event_fd(-1)
+    { }
+
+    Polling::Tag bind(Polling::Epoll& poller);
+    bool isBound() const;
 
     Polling::Tag tag() const;