Implemented event loop for async write

main.cc

@@ -19,7 +19,7 @@ class MyHandler : public Net::Http::Handler {
                     .add(std::make_shared<Header::ContentType>(MIME(Text, Plain)));
 
                 response.send(Net::Http::Code::Ok, "PONG").then([](ssize_t  bytes) {
-                    std::cout << "Sent total of " << bytes << " bytes" << std::endl;
+                    //std::cout << "Sent total of " << bytes << " bytes" << std::endl;
                 }, Async::IgnoreException);
 
             }

src/http.cc

@@ -350,6 +350,8 @@ Request::query() const {
 
 Response::Response()
     : Message()
+    , bufSize(Const::MaxBuffer << 1)
+    , buffer_(new char[bufSize])
 { }
 
 void
@@ -369,40 +371,52 @@ Response::send(Code code) {
 Async::Promise<ssize_t>
 Response::send(Code code, const std::string& body, const Mime::MediaType& mime)
 {
-    Io::OutArrayBuf obuf(buffer_, Io::Init::ZeroOut);
+
+    char *beg = buffer_.get();
+    Io::OutArrayBuf obuf(beg, beg + bufSize, Io::Init::ZeroOut);
 
     std::ostream stream(&obuf);
-    stream << "HTTP/1.1 ";
-    stream << static_cast<int>(code);
-    stream << ' ';
-    stream << code;
-    stream << crlf;
+#define OUT(...) \
+    do { \
+        __VA_ARGS__; \
+        if (!stream) { \
+            return Async::Promise<ssize_t>::rejected(Net::Error("Could not write to stream: insufficient space")); \
+        } \
+    } while (0);
+
+    OUT(stream << "HTTP/1.1 ");
+    OUT(stream << static_cast<int>(code));
+    OUT(stream << ' ');
+    OUT(stream << code);
+    OUT(stream << crlf);
 
     if (mime.isValid()) {
         auto contentType = headers_.tryGet<Header::ContentType>();
         if (contentType)
             contentType->setMime(mime);
         else {
-            writeHeader<Header::ContentType>(stream, mime);
+            OUT(writeHeader<Header::ContentType>(stream, mime));
         }
     }
 
     for (const auto& header: headers_.list()) {
-        stream << header->name() << ": ";
-        header->write(stream);
-        stream << crlf;
+        OUT(stream << header->name() << ": ");
+        OUT(header->write(stream));
+        OUT(stream << crlf);
     }
 
     if (!body.empty()) {
-        writeHeader<Header::ContentLength>(stream, body.size());
-        stream << crlf;
-        stream << body;
+        OUT(writeHeader<Header::ContentLength>(stream, body.size()));
+        OUT(stream << crlf);
+        OUT(stream << body);
     }
     else {
-        stream << crlf;
+        OUT(stream << crlf);
     }
 
-    return peer()->send(buffer_, obuf.len());
+#undef OUT
+
+    return peer()->send(buffer_.get(), obuf.len());
 }
 
 void

src/http.h

@@ -107,7 +107,10 @@ private:
 
     void associatePeer(const std::shared_ptr<Tcp::Peer>& peer);
     std::weak_ptr<Tcp::Peer> peer_;
-    char buffer_[Const::MaxBuffer];
+
+    size_t bufSize;
+    std::unique_ptr<char[]> buffer_;
+
 };
 
 namespace Private {

src/io.cc

@@ -189,6 +189,24 @@ IoWorker::run() {
                         handleIncoming(peer);
                     }
                     else if (event.flags.hasFlag(NotifyOn::Write)) {
+                        auto fd = event.tag.value();
+                        auto it = toWrite.find(fd);
+                        if (it == std::end(toWrite)) {
+                            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"));
+                        }
+
+                        else if (bytes < write.len) {
+                            write.reject(Net::Error("Failed to write: could not write all bytes"));
+                        }
+                        else {
+                            write.resolve(bytes);
+                        }
                     }
                 }
             }
@@ -201,21 +219,38 @@ IoWorker::run() {
 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) {
-        ssize_t bytes = ::send(fd, buf, len, 0);
-        if (bytes < 0) {
-            if (errno == EAGAIN || errno == EWOULDBLOCK) {
-                toWrite.insert(
-                        std::make_pair(fd, 
-                            OnHoldWrite(std::move(resolve), std::move(reject), buf, len)));
-                poller.addFdOneShot(fd, NotifyOn::Write, Polling::Tag(fd));
+
+        auto it = toWrite.find(fd);
+        if (it != std::end(toWrite)) {
+            reject(Net::Error("Multiple writes on the same fd"));
+            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) {
+                    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"));
+                }
+                break;
             }
             else {
-                reject(std::runtime_error("send"));
+                totalWritten += bytesWritten;
+                if (totalWritten == len) {
+                    resolve(totalWritten);
+                    break;
+                }
             }
         }
-        else {
-            resolve(bytes);
-        }
     });
 }
 

src/net.cc

@@ -106,5 +106,24 @@ Port Address::port() const {
     return port_;
 }
 
+Error::Error(const char* message)
+    : std::runtime_error(message)
+{ }
+
+Error::Error(std::string message)
+    : std::runtime_error(std::move(message))
+{ }
+
+Error
+Error::system(const char* message) {
+    const char *err = strerror(errno);
+
+    std::string str(message);
+    str += ':';
+    str += err;
+
+    return Error(std::move(str));
+
+}
 
 } // namespace Net

src/net.h

@@ -7,6 +7,7 @@
 #pragma once
 #include <string>
 #include <sys/socket.h>
+#include <stdexcept>
 
 #ifndef _KERNEL_FASTOPEN
 #define _KERNEL_FASTOPEN
@@ -69,4 +70,11 @@ private:
     Port port_;
 };
 
+class Error : public std::runtime_error {
+public:
+    Error(const char* message);
+    Error(std::string message);
+    static Error system(const char* message);
+};
+
 } // namespace Net