Http: Split HttpAsyncServer out of HttpServer

Author: Pagghiu

Examples/SCExample/Examples/WebServerExample/WebServerExample.cpp

@@ -17,6 +17,7 @@
 
 #include "Libraries/ContainersReflection/ContainersReflection.h"
 #include "Libraries/ContainersReflection/MemorySerialization.h"
+#include "Libraries/Http/HttpAsyncServer.h"
 #include "Libraries/Http/HttpWebServer.h"
 #include "Libraries/Plugin/PluginMacros.h"
 #include "Libraries/SerializationBinary/SerializationBinary.h"
@@ -63,8 +64,8 @@ struct SC::WebServerExampleModel
 
     AsyncEventLoop* eventLoop = nullptr;
 
-    HttpServer    httpServer;
-    HttpWebServer httpWebServer;
+    HttpAsyncServer httpServer;
+    HttpWebServer   httpWebServer;
 
     Buffer                 headerBuffer;
     GrowableBuffer<Buffer> gb = {headerBuffer};
@@ -77,7 +78,7 @@ struct SC::WebServerExampleModel
         clients = {new HttpServerClient[numClients], numClients};
         SC_TRY(gb.resizeWithoutInitializing(1024 * 8 * numClients));
         HttpServer::Memory memory = {gb, clients};
-        httpServer.onRequest.bind<HttpWebServer, &HttpWebServer::serveFile>(httpWebServer);
+        httpServer.httpServer.onRequest.bind<HttpWebServer, &HttpWebServer::serveFile>(httpWebServer);
         SC_TRY(
             httpServer.start(*eventLoop, modelState.interface.view(), static_cast<uint16_t>(modelState.port), memory));
         SC_TRY(httpWebServer.init(modelState.directory.view()));

Libraries/Http/HttpAsyncServer.cpp

@@ -0,0 +1,152 @@
+// Copyright (c) Stefano Cristiano
+// SPDX-License-Identifier: MIT
+#include "HttpAsyncServer.h"
+#include "../Foundation/Deferred.h"
+
+namespace SC
+{
+Result HttpAsyncServer::start(AsyncEventLoop& loop, StringSpan address, uint16_t port, HttpServer::Memory& memory)
+{
+    SocketIPAddress nativeAddress;
+    SC_TRY(nativeAddress.fromAddressPort(address, port));
+    eventLoop = &loop;
+    SC_TRY(eventLoop->createAsyncTCPSocket(nativeAddress.getAddressFamily(), serverSocket));
+    SocketServer socketServer(serverSocket);
+    SC_TRY(socketServer.bind(nativeAddress));
+    SC_TRY(socketServer.listen(511));
+
+    asyncServerAccept.setDebugName("HttpServer");
+    asyncServerAccept.callback.bind<HttpAsyncServer, &HttpAsyncServer::onNewClient>(*this);
+    SC_TRY(asyncServerAccept.start(*eventLoop, serverSocket));
+    started = true;
+
+    return httpServer.start(memory);
+}
+
+Result HttpAsyncServer::stopAsync()
+{
+    stopping       = true;
+    auto deferStop = MakeDeferred([this]() { stopping = false; });
+    if (not asyncServerAccept.isFree())
+    {
+        SC_TRY(asyncServerAccept.stop(*eventLoop));
+    }
+
+    for (HttpServerClient& it : httpServer.clients)
+    {
+        if (it.state != HttpServerClient::State::Free)
+        {
+            closeAsync(it);
+        }
+    }
+    return Result(true);
+}
+
+Result HttpAsyncServer::stopSync()
+{
+    stopping       = true;
+    auto deferStop = MakeDeferred([this]() { stopping = false; });
+    SC_TRY(stopAsync());
+    while (httpServer.getNumClients() > 0)
+    {
+        SC_TRY(eventLoop->runNoWait());
+    }
+    while (not asyncServerAccept.isFree())
+    {
+        SC_TRY(eventLoop->runNoWait());
+    }
+    started = false;
+    return Result(true);
+}
+
+void HttpAsyncServer::onNewClient(AsyncSocketAccept::Result& result)
+{
+    SocketDescriptor acceptedClient;
+    if (not result.moveTo(acceptedClient))
+    {
+        // TODO: Invoke an error
+        return;
+    }
+    size_t idx = 0;
+    // Allocate always succeeds because we pause asyncAccept when the arena is full
+    SC_ASSERT_RELEASE(httpServer.allocateClient(idx));
+
+    HttpServerClient& client = httpServer.clients[idx];
+
+    client.socket = move(acceptedClient);
+    client.asyncSend.setDebugName(client.debugName);
+    client.asyncReceive.setDebugName(client.debugName);
+    client.asyncReceive.callback.bind<HttpAsyncServer, &HttpAsyncServer::onReceive>(*this);
+
+    // This cannot fail because start reports only incorrect API usage (AsyncRequest already in use etc.)
+    SC_TRUST_RESULT(client.asyncReceive.start(*eventLoop, client.socket, client.request.availableHeader));
+
+    // Only reactivate asyncAccept if arena is not full (otherwise it's being reactivated in closeAsync)
+    result.reactivateRequest(httpServer.canAcceptMoreClients());
+}
+
+void HttpAsyncServer::onReceive(AsyncSocketReceive::Result& result)
+{
+    SC_COMPILER_WARNING_PUSH_OFFSETOF
+    HttpServerClient& client = SC_COMPILER_FIELD_OFFSET(HttpServerClient, asyncReceive, result.getAsync());
+    SC_COMPILER_WARNING_POP
+    SC_ASSERT_RELEASE(&client.asyncReceive == &result.getAsync());
+    Span<char> readData;
+    if (not result.get(readData))
+    {
+        // TODO: Invoke on error
+        return;
+    }
+
+    const size_t idx     = static_cast<size_t>(&client - &httpServer.clients[0]);
+    Span<char>   outspan = httpServer.processClientReceivedData(idx, readData);
+    if (not outspan.empty())
+    {
+        client.asyncSend.setDebugName(client.debugName);
+        client.asyncSend.callback.bind<HttpAsyncServer, &HttpAsyncServer::onAfterSend>(*this);
+        auto res = client.asyncSend.start(*eventLoop, client.socket, outspan);
+        if (not res)
+        {
+            // TODO: Invoke on error
+            return;
+        }
+    }
+    else if (not result.completionData.disconnected)
+    {
+        result.reactivateRequest(true);
+    }
+}
+
+void HttpAsyncServer::onAfterSend(AsyncSocketSend::Result& result)
+{
+    if (result.isValid())
+    {
+        SC_COMPILER_WARNING_PUSH_OFFSETOF
+        HttpServerClient& requestClient = SC_COMPILER_FIELD_OFFSET(HttpServerClient, asyncSend, result.getAsync());
+        SC_COMPILER_WARNING_POP
+        closeAsync(requestClient);
+    }
+}
+
+void HttpAsyncServer::closeAsync(HttpServerClient& requestClient)
+{
+    if (not requestClient.asyncSend.isFree())
+    {
+        (void)requestClient.asyncSend.stop(*eventLoop);
+    }
+    if (not requestClient.asyncReceive.isFree())
+    {
+        (void)requestClient.asyncReceive.stop(*eventLoop);
+    }
+    SC_TRUST_RESULT(requestClient.socket.close());
+    const bool wasFull = not httpServer.canAcceptMoreClients();
+
+    SC_TRUST_RESULT(httpServer.deallocateClient(requestClient));
+    if (wasFull and not stopping)
+    {
+        // Arena was full and in onNewClient asyncAccept has been paused (by avoiding reactivation).
+        // Now a slot has been freed so it's possible to start accepting clients again.
+        SC_TRUST_RESULT(asyncServerAccept.start(*eventLoop, serverSocket));
+    }
+}
+} // namespace SC

Libraries/Http/HttpAsyncServer.h

@@ -0,0 +1,49 @@
+// Copyright (c) Stefano Cristiano
+// SPDX-License-Identifier: MIT
+#pragma once
+#include "HttpServer.h"
+namespace SC
+{
+#if SC_COMPILER_MSVC
+#pragma warning(push)
+#pragma warning(disable : 4251)
+#endif
+struct SC_COMPILER_EXPORT HttpAsyncServer
+{
+    /// @brief Starts the http server on the given AsyncEventLoop, address and port
+    /// @param loop The event loop to be used, where to add the listening socket
+    /// @param address The address of local interface where to listen to
+    /// @param port The local port where to start listening to
+    /// @param memory Memory buffers to be used by the http server
+    /// @return Valid Result if http listening has been started successfully
+    Result start(AsyncEventLoop& loop, StringSpan address, uint16_t port, HttpServer::Memory& memory);
+
+    /// @brief Stops http server asynchronously pushing cancel and close requests for next SC::AsyncEventLoop::runOnce
+    Result stopAsync();
+
+    /// @brief Stops http server synchronously waiting for SC::AsyncEventLoop::runNoWait to cancel or close all requests
+    Result stopSync();
+
+    /// @brief Returns true if the server has been started
+    [[nodiscard]] bool isStarted() const { return started; }
+
+    /// @brief The underlying http server
+    HttpServer httpServer;
+
+  private:
+    bool started  = false;
+    bool stopping = false;
+
+    void onNewClient(AsyncSocketAccept::Result& result);
+    void onReceive(AsyncSocketReceive::Result& result);
+    void onAfterSend(AsyncSocketSend::Result& result);
+    void closeAsync(HttpServerClient& requestClient);
+
+    AsyncEventLoop*   eventLoop = nullptr;
+    SocketDescriptor  serverSocket;
+    AsyncSocketAccept asyncServerAccept;
+};
+#if SC_COMPILER_MSVC
+#pragma warning(pop)
+#endif
+} // namespace SC