Experimenting with a timeout API

main.cc

@@ -7,6 +7,16 @@
 
 using namespace std;
 
+struct ExceptionPrinter {
+    void operator()(std::exception_ptr exc) const {
+        try {
+            std::rethrow_exception(exc);
+        } catch (const std::exception& e) {
+            std::cerr << "An exception occured: " << e.what() << std::endl;
+        }
+    }
+};
+
 class MyHandler : public Net::Http::Handler {
     void onRequest(const Net::Http::Request& req, Net::Http::Response response) {
         if (req.resource() == "/ping") {
@@ -37,6 +47,13 @@ class MyHandler : public Net::Http::Handler {
         else if (req.resource() == "/exception") {
             throw std::runtime_error("Exception thrown in the handler");
         }
+        else if (req.resource() == "/timeout") {
+            response
+                .timeoutAfter(std::chrono::seconds(1))
+                .then([=](Net::Http::Response *response) {
+                    response->send(Net::Http::Code::Bad_Request, "Timeout occured");
+            }, Async::NoExcept);
+        }
     }
 };
 

src/http.cc

@@ -348,18 +348,33 @@ Request::query() const {
     return query_;
 }
 
-void
-ResponseWriter::writeStatusLine() {
-}
-
-void
-ResponseWriter::writeHeaders() {
-    std::ostream os(&stream_);
-
+Async::Promise<Response *>
+Response::timeoutAfter(std::chrono::milliseconds timeout)
+{
+#if 0
+    Async::Promise<uint64_t> promise([=](Async::Resolver& resolve, Async::Rejection& reject) {
+       peer()->io()->setTimeout(timeout, resolve, reject);
+   }); 
+
+    promise.then([=](uint64_t) {
+        send(Code::Bad_Request, "A timeout occured");
+    }, Async::NoExcept);
+
+    return promise;
+#endif
+    Async::Promise<uint64_t> promise([=](Async::Resolver& resolve, Async::Rejection& reject) {
+       peer()->io()->setTimeout(timeout, resolve, reject);
+   }); 
+
+    return promise.then([=](uint64_t) {
+        auto p = Async::Promise<Response *>::resolved(this);
+
+        return p;
+    }, Async::NoExcept);
 }
 
 Async::Promise<ssize_t>
-ResponseWriter::send()
+ResponseWriter::send() const
 {
     auto body = stream_.buffer();
 

src/http.h

@@ -131,7 +131,7 @@ public:
         return &stream_;
     }
 
-    Async::Promise<ssize_t> send();
+    Async::Promise<ssize_t> send() const;
 
 private:
     ResponseWriter(Message&& other, size_t size, std::weak_ptr<Tcp::Peer> peer)
@@ -156,9 +156,6 @@ private:
         return peer_.lock();
     }
 
-    void writeStatusLine();
-    void writeHeaders();
-
     NetworkStream stream_;
     std::weak_ptr<Tcp::Peer> peer_;
 };
@@ -182,6 +179,8 @@ public:
         return *this;
     }
 
+    Response(const Response& other) = default;
+
     const Header::Collection& headers() const {
         return headers_;
     }
@@ -235,6 +234,8 @@ public:
         return ResponseWriter(std::move(*this), size, peer_);
     }
 
+    Async::Promise<Response *> timeoutAfter(std::chrono::milliseconds timeout);
+
 private:
     Response()
         : Message()

src/io.cc

@@ -9,6 +9,7 @@
 #include "listener.h"
 #include "peer.h"
 #include "os.h"
+#include <sys/timerfd.h>
 
 namespace Net {
 
@@ -35,6 +36,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));
 }
 
 IoWorker::~IoWorker() {
@@ -70,6 +73,36 @@ IoWorker::pin(const CpuSet& set) {
     }
 }
 
+void
+IoWorker::setTimeoutMs(
+        std::chrono::milliseconds value,
+        Async::Resolver resolve, Async::Rejection reject)
+{
+    itimerspec spec;
+    spec.it_interval.tv_sec = 0;
+    spec.it_interval.tv_nsec = 0;
+
+    if (value.count() < 1000) {
+        spec.it_value.tv_sec = 0;
+        spec.it_value.tv_nsec
+            = std::chrono::duration_cast<std::chrono::nanoseconds>(value).count();
+    } else {
+        spec.it_value.tv_sec
+            = std::chrono::duration_cast<std::chrono::seconds>(value).count();
+        spec.it_value.tv_nsec = 0;
+    }
+
+
+    int res = timerfd_settime(timerFd, 0, &spec, 0);
+    if (res == -1) {
+        reject(Error::system("Could not set timer time"));
+        return;
+    }
+
+    timeout = Some(Timeout(value, std::move(resolve), std::move(reject)));
+
+}
+
 void
 IoWorker::shutdown() {
     mailbox.post(new ShutdownMessage());
@@ -149,7 +182,7 @@ IoWorker::handleNewPeer(const std::shared_ptr<Peer>& peer)
         peers.insert(std::make_pair(fd, peer));
     }
 
-    peer->io = this;
+    peer->io_ = this;
 
     handler_->onConnection(peer);
     poller.addFd(fd, NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
@@ -184,9 +217,14 @@ IoWorker::run() {
                         return;
                     }
                 } else {
-                    auto& peer = getPeer(event.tag);
                     if (event.flags.hasFlag(NotifyOn::Read)) {
-                        handleIncoming(peer);
+                        auto fd = event.tag.value();
+                        if (fd == timerFd) {
+                            handleTimeout();
+                        } else {
+                            auto& peer = getPeer(event.tag);
+                            handleIncoming(peer);
+                        }
                     }
                     else if (event.flags.hasFlag(NotifyOn::Write)) {
                         auto fd = event.tag.value();
@@ -216,6 +254,29 @@ IoWorker::run() {
     }
 }
 
+void
+IoWorker::handleTimeout() {
+
+    auto& entry = timeout.unsafeGet();
+
+    uint64_t numWakeups;
+    int res = ::read(timerFd, &numWakeups, sizeof numWakeups);
+    if (res == -1) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK)
+            return;
+        else
+            entry.reject(Error::system("Could not read timerfd"));
+    } else {
+        if (res != sizeof(numWakeups)) {
+            entry.reject(Error("Read invalid number of bytes for timer fd: "
+                        + std::to_string(timerFd)));
+        }
+        else {
+            entry.resolve(numWakeups);
+        }
+    }
+}
+
 Async::Promise<ssize_t>
 IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
     return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {

src/io.h

@@ -39,6 +39,13 @@ public:
     void pin(const CpuSet& set);
 
     void shutdown();
+    template<typename Duration>
+    void setTimeout(Duration timeout, Async::Resolver resolve, Async::Rejection reject) {
+        setTimeoutMs(std::chrono::duration_cast<std::chrono::milliseconds>(timeout), 
+                   std::move(resolve),
+                   std::move(reject));
+    }
+
 
 private:
     struct OnHoldWrite {
@@ -57,11 +64,33 @@ private:
         size_t len;   
     };
 
+    void
+    setTimeoutMs(std::chrono::milliseconds value, Async::Resolver, Async::Rejection reject);
+
+    struct Timeout {
+        Timeout(std::chrono::milliseconds value,
+                Async::Resolver resolve,
+                Async::Rejection reject)
+          : value(value)
+          , resolve(std::move(resolve))
+          , reject(std::move(reject))  {
+        } 
+
+        std::chrono::milliseconds value;
+
+        Async::Resolver resolve;
+        Async::Rejection reject;
+    };
+
     Polling::Epoll poller;
     std::unique_ptr<std::thread> thread;
     mutable std::mutex peersMutex;
     std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
     std::unordered_map<Fd, OnHoldWrite> toWrite;
+
+    Optional<Timeout> timeout;
+    Fd timerFd;
+
     std::shared_ptr<Handler> handler_;
     Flags<Options> options_;
 
@@ -73,6 +102,7 @@ private:
     Async::Promise<ssize_t> asyncWrite(Fd fd, const void *buf, size_t len);
 
     void handleIncoming(const std::shared_ptr<Peer>& peer);
+    void handleTimeout();
     void run();
 
 };

src/peer.cc

@@ -79,7 +79,7 @@ Peer::tryGetData(std::string(name)) const {
 
 Async::Promise<ssize_t>
 Peer::send(const void* buf, size_t len) {
-    return io->asyncWrite(fd_, buf, len);
+    return io_->asyncWrite(fd_, buf, len);
 }
 
 

src/peer.h

@@ -52,8 +52,12 @@ public:
 
     Async::Promise<ssize_t> send(const void* buf, size_t len);
 
+    IoWorker *io() {
+        return io_;
+    }
+
 private:
-    IoWorker* io;
+    IoWorker* io_;
 
     Address addr;
     std::string hostname_;