Http: Allow declaring fixed size queues side by side with HttpConnection

This simplifies the API and also improves separation of concerns, getting rid of all references to SC::Async inside the HttpConnection.h header.

Author: Pagghiu

Examples/SCExample/Examples/WebServerExample/StableArray.h

@@ -112,6 +112,9 @@ struct StableArray
     [[nodiscard]] operator Span<T>() { return {data(), sizeElements}; }
     [[nodiscard]] operator Span<const T>() const { return {data(), sizeElements}; }
 
+    [[nodiscard]] Span<T>       toSpan() { return {data(), sizeElements}; }
+    [[nodiscard]] Span<const T> toSpan() const { return {data(), sizeElements}; }
+
   private:
     VirtualMemory virtualMemory;
 

Examples/SCExample/Examples/WebServerExample/WebServerExample.cpp

@@ -82,11 +82,11 @@ struct SC::WebServerExampleModel
     StableArray<char> requestsMemory = {MAX_CONNECTIONS * REQUEST_SIZE}; // Memory sliced into buffers for streaming
     StableArray<char> headersMemory  = {MAX_CONNECTIONS * HEADER_SIZE};  // Memory holding request / response headers
 
-    StableArray<HttpConnection>              connections = {MAX_CONNECTIONS};
-    StableArray<AsyncBufferView>             buffers     = {MAX_CONNECTIONS * WRITE_QUEUE}; // WRITE_QUEUE > READ_QUEUE
-    StableArray<ReadableFileStream::Request> readQueue   = {MAX_CONNECTIONS * READ_QUEUE};
-    StableArray<WritableFileStream::Request> writeQueue  = {MAX_CONNECTIONS * WRITE_QUEUE};
-    StableArray<HttpAsyncFileServerStream>   fileStreams = {MAX_CONNECTIONS};
+    using HttpCustomConnection = HttpAsyncConnection<READ_QUEUE, WRITE_QUEUE>; // Connection with fixed read/write queue
+
+    StableArray<HttpCustomConnection>      connections = {MAX_CONNECTIONS};
+    StableArray<AsyncBufferView>           buffers     = {MAX_CONNECTIONS * (WRITE_QUEUE + READ_QUEUE)};
+    StableArray<HttpAsyncFileServerStream> fileStreams = {MAX_CONNECTIONS};
 
     Result start()
     {
@@ -96,8 +96,6 @@ struct SC::WebServerExampleModel
         SC_TRY(headersMemory.resizeWithoutInitializing(numClients * HEADER_SIZE));
         SC_TRY(connections.resize(numClients));
         SC_TRY(buffers.resize(numClients * WRITE_QUEUE));
-        SC_TRY(readQueue.resize(numClients * READ_QUEUE));
-        SC_TRY(writeQueue.resize(numClients * WRITE_QUEUE));
         SC_TRY(fileStreams.resize(numClients));
 
         // Slice requests buffer in equal parts to create re-usable sub-buffers to stream files.
@@ -111,7 +109,7 @@ struct SC::WebServerExampleModel
             SC_TRY(threadPool.create(NUM_FS_THREADS));
         }
         // Initialize and start the http and the file server
-        SC_TRY(httpServer.init(buffersPool, connections, headersMemory, readQueue, writeQueue));
+        SC_TRY(httpServer.init(buffersPool, connections.toSpan(), headersMemory));
         SC_TRY(fileServer.init(buffersPool, threadPool, modelState.directory.view(), fileStreams));
         SC_TRY(httpServer.start(*eventLoop, modelState.interface.view(), static_cast<uint16_t>(modelState.port)));
         SC_TRY(fileServer.start(httpServer));
@@ -130,8 +128,6 @@ struct SC::WebServerExampleModel
         headersMemory.release();       // Skips invoking destructors, just release virtual memory
         connections.clearAndRelease(); // Invokes destructors and releases virtual memory
         buffers.clearAndRelease();     // Invokes destructors and releases virtual memory
-        readQueue.clearAndRelease();   // Invokes destructors and releases virtual memory
-        writeQueue.clearAndRelease();  // Invokes destructors and releases virtual memory
         fileStreams.clearAndRelease(); // Invokes destructors and releases virtual memory
         return Result(true);
     }

Libraries/Http/HttpAsyncServer.cpp

@@ -6,9 +6,19 @@
 
 namespace SC
 {
+
+Result HttpAsyncServer::initInternal(AsyncBuffersPool& pool, SpanWithStride<HttpConnection> connectionsSpan,
+                                     Span<char> headersMemory)
+{
+    // TODO: Add some validation of minimum sizes for the queues and the buffers
+    SC_TRY(connections.init(connectionsSpan, headersMemory));
+    buffersPool = &pool;
+    return Result(true);
+}
+
 Result HttpAsyncServer::start(AsyncEventLoop& loop, StringSpan address, uint16_t port)
 {
-    SC_TRY_MSG(memory, "HttpAsyncServer::start - init not called");
+    SC_TRY_MSG(buffersPool != nullptr, "HttpAsyncServer::start - init not called");
     SocketIPAddress nativeAddress;
     SC_TRY(nativeAddress.fromAddressPort(address, port));
     eventLoop = &loop;
@@ -24,31 +34,11 @@ Result HttpAsyncServer::start(AsyncEventLoop& loop, StringSpan address, uint16_t
     return Result(true);
 }
 
-Result HttpAsyncServer::init(AsyncBuffersPool& pool, Span<HttpConnection> clients, Span<char> headersMemory,
-                             Span<AsyncReadableStream::Request> readQueue,
-                             Span<AsyncWritableStream::Request> writeQueue)
-{
-
-    // TODO: Add some validation of minimum sizes for the queues and the buffers
-    SC_TRY(connections.init(clients, headersMemory));
-    readQueues  = readQueue;
-    writeQueues = writeQueue;
-
-    buffersPool = &pool;
-
-    memory = true;
-    return Result(true);
-}
-
 Result HttpAsyncServer::close()
 {
     SC_TRY(waitForStopToFinish());
     SC_TRY(connections.close());
-    readQueues  = {};
-    writeQueues = {};
     buffersPool = nullptr;
-
-    memory = false;
     return Result(true);
 }
 
@@ -64,7 +54,7 @@ Result HttpAsyncServer::stop()
 
     for (size_t idx = 0; idx < connections.getNumTotalConnections(); ++idx)
     {
-        HttpConnection& client = connections.getConnectionAt(idx);
+        HttpAsyncConnectionBase& client = static_cast<HttpAsyncConnectionBase&>(connections.getConnectionAt(idx));
         // Destroy can be safely called in any state (including already destroyed)
         client.readableSocketStream.destroy();
         client.writableSocketStream.destroy();
@@ -89,7 +79,7 @@ Result HttpAsyncServer::waitForStopToFinish()
         checkAgainAllClients = false;
         for (size_t idx = 0; idx < connections.getNumTotalConnections(); ++idx)
         {
-            HttpConnection& client = connections.getConnectionAt(idx);
+            HttpAsyncConnectionBase& client = static_cast<HttpAsyncConnectionBase&>(connections.getConnectionAt(idx));
             while (not client.readableSocketStream.request.isFree() or not client.writableSocketStream.request.isFree())
             {
                 SC_TRY(eventLoop->runNoWait());
@@ -113,23 +103,13 @@ void HttpAsyncServer::onNewClient(AsyncSocketAccept::Result& result)
     // Activation always succeeds because we pause asyncAccept when the there are not available clients
     SC_ASSERT_RELEASE(connections.activateNew(idx));
 
-    HttpConnection& client     = connections.getConnection(idx);
-    client.socket              = move(acceptedClient);
-    const size_t readQueueLen  = readQueues.sizeInElements() / connections.getNumTotalConnections();
-    const size_t writeQueueLen = writeQueues.sizeInElements() / connections.getNumTotalConnections();
-    SC_TRUST_RESULT(readQueueLen > 0);
-    SC_TRUST_RESULT(writeQueueLen > 0);
-    Span<AsyncReadableStream::Request> readQueue;
-    Span<AsyncWritableStream::Request> writeQueue;
-    SC_TRUST_RESULT(readQueues.sliceStartLength(idx.getIndex() * readQueueLen, readQueueLen, readQueue));
-    SC_TRUST_RESULT(writeQueues.sliceStartLength(idx.getIndex() * writeQueueLen, writeQueueLen, writeQueue));
-    client.readableSocketStream.setReadQueue(readQueue);
-    client.writableSocketStream.setWriteQueue(writeQueue);
+    HttpAsyncConnectionBase& client = static_cast<HttpAsyncConnectionBase&>(connections.getConnection(idx));
+    client.socket                   = move(acceptedClient);
     SC_TRUST_RESULT(client.readableSocketStream.init(*buffersPool, *eventLoop, client.socket));
     SC_TRUST_RESULT(client.writableSocketStream.init(*buffersPool, *eventLoop, client.socket));
 
     auto onData = [this, idx](AsyncBufferView::ID bufferID)
-    { onStreamReceive(connections.getConnection(idx), bufferID); };
+    { onStreamReceive(static_cast<HttpAsyncConnectionBase&>(connections.getConnection(idx)), bufferID); };
     SC_TRUST_RESULT(client.readableSocketStream.eventData.addListener(onData));
     SC_TRUST_RESULT(client.readableSocketStream.start());
 
@@ -139,7 +119,7 @@ void HttpAsyncServer::onNewClient(AsyncSocketAccept::Result& result)
     result.reactivateRequest(connections.getNumActiveConnections() < connections.getNumTotalConnections());
 }
 
-void HttpAsyncServer::onStreamReceive(HttpConnection& client, AsyncBufferView::ID bufferID)
+void HttpAsyncServer::onStreamReceive(HttpAsyncConnectionBase& client, AsyncBufferView::ID bufferID)
 {
     Span<char> readData;
     SC_ASSERT_RELEASE(buffersPool->getWritableData(bufferID, readData));
@@ -159,8 +139,8 @@ void HttpAsyncServer::onStreamReceive(HttpConnection& client, AsyncBufferView::I
         // Using a struct instead of a lambda so it can unregister itself
         struct AfterWrite
         {
-            HttpAsyncServer& pself;
-            HttpConnection&  client;
+            HttpAsyncServer&         pself;
+            HttpAsyncConnectionBase& client;
 
             void operator()()
             {
@@ -172,7 +152,7 @@ void HttpAsyncServer::onStreamReceive(HttpConnection& client, AsyncBufferView::I
     }
 }
 
-void HttpAsyncServer::closeAsync(HttpConnection& client)
+void HttpAsyncServer::closeAsync(HttpAsyncConnectionBase& client)
 {
     if (client.state == HttpConnection::State::Inactive)
     {

Libraries/Http/HttpAsyncServer.h

@@ -1,9 +1,44 @@
 // Copyright (c) Stefano Cristiano
 // SPDX-License-Identifier: MIT
 #pragma once
+#include "../AsyncStreams/AsyncRequestStreams.h"
 #include "HttpConnection.h"
+
 namespace SC
 {
+namespace TypeTraits
+{
+/// IsBaseOf evaluates to `true` if the type `Base` is a base class of `Derived`, `false` otherwise.
+template <typename Base, typename Derived>
+struct IsBaseOf
+{
+    static constexpr bool value = __is_base_of(Base, Derived);
+};
+
+} // namespace TypeTraits
+
+/// @brief Contains fields used by HttpAsyncServer for each connection
+struct SC_COMPILER_EXPORT HttpAsyncConnectionBase : public HttpConnection
+{
+    ReadableSocketStream readableSocketStream;
+    WritableSocketStream writableSocketStream;
+    SocketDescriptor     socket;
+};
+
+/// @brief Adds compile-time configurable read and write queues to HttpAsyncConnectionBase
+template <int ReadQueue, int WriteQueue>
+struct SC_COMPILER_EXPORT HttpAsyncConnection : public HttpAsyncConnectionBase
+{
+    ReadableFileStream::Request readQueue[ReadQueue];
+    WritableFileStream::Request writeQueue[WriteQueue];
+
+    HttpAsyncConnection()
+    {
+        readableSocketStream.setReadQueue(readQueue);
+        writableSocketStream.setWriteQueue(writeQueue);
+    }
+};
+
 /// @brief Async Http Server
 ///
 /// This class handles a fully asynchronous http server staying inside 5 fixed memory regions passed during init.
@@ -19,8 +54,12 @@ namespace SC
 struct SC_COMPILER_EXPORT HttpAsyncServer
 {
     /// @brief Initializes the async server with all needed memory buffers
-    Result init(AsyncBuffersPool& buffersPool, Span<HttpConnection> clients, Span<char> headersMemory,
-                Span<AsyncReadableStream::Request> readQueue, Span<AsyncWritableStream::Request> writeQueue);
+    template <typename T,
+              typename = typename TypeTraits::EnableIf<TypeTraits::IsBaseOf<HttpAsyncConnectionBase, T>::value>::type>
+    Result init(AsyncBuffersPool& pool, Span<T> clients, Span<char> headersMemory)
+    {
+        return initInternal(pool, {clients.data(), clients.sizeInElements(), sizeof(T)}, headersMemory);
+    }
 
     /// @brief Closes the server, removing references to the memory buffers passed during init
     /// @note This call will wait until all async operations will be finished before returning
@@ -40,11 +79,8 @@ struct SC_COMPILER_EXPORT HttpAsyncServer
     /// @brief Returns true if the server has been started
     [[nodiscard]] bool isStarted() const { return started; }
 
-    /// @brief Access the underlying http connections
-    HttpConnectionsPool& getConnectionsPool() { return connections; }
-
     /// @brief Access the underlying AsyncEventLoop
-    AsyncEventLoop* getEventLoop() const { return eventLoop; }
+    [[nodiscard]] AsyncEventLoop* getEventLoop() const { return eventLoop; }
 
     /// @brief Called after enough data from a newly connected client has arrived, causing all headers to be parsed.
     Function<void(HttpConnection&)> onRequest;
@@ -55,18 +91,15 @@ struct SC_COMPILER_EXPORT HttpAsyncServer
 
     uint32_t maxHeaderSize = 8 * 1024;
 
-    Span<AsyncReadableStream::Request> readQueues;
-    Span<AsyncWritableStream::Request> writeQueues;
-
     bool started  = false;
     bool stopping = false;
-    bool memory   = false;
 
     void onNewClient(AsyncSocketAccept::Result& result);
-    void closeAsync(HttpConnection& requestClient);
-    void onStreamReceive(HttpConnection& client, AsyncBufferView::ID bufferID);
+    void closeAsync(HttpAsyncConnectionBase& requestClient);
+    void onStreamReceive(HttpAsyncConnectionBase& client, AsyncBufferView::ID bufferID);
 
     Result waitForStopToFinish();
+    Result initInternal(AsyncBuffersPool& pool, SpanWithStride<HttpConnection> connections, Span<char> headersMemory);
 
     AsyncEventLoop*   eventLoop = nullptr;
     SocketDescriptor  serverSocket;

Libraries/Http/HttpConnection.cpp

@@ -172,7 +172,7 @@ void HttpResponse::reset() { headersSent = false; }
 //-------------------------------------------------------------------------------------------------------
 // HttpConnectionsPool
 //-------------------------------------------------------------------------------------------------------
-Result HttpConnectionsPool::init(Span<HttpConnection> connectionsStorage, Span<char> headersMemoryStorage)
+Result HttpConnectionsPool::init(SpanWithStride<HttpConnection> connectionsStorage, Span<char> headersMemoryStorage)
 {
     SC_TRY_MSG(numConnections == 0, "HttpConnectionsPool::init - numConnections != 0");
     connections   = connectionsStorage;

Libraries/Http/HttpConnection.h

@@ -3,9 +3,7 @@
 #pragma once
 #include "HttpParser.h"
 
-#include "../Async/Async.h"
-#include "../AsyncStreams/AsyncRequestStreams.h"
-#include "../Foundation/Function.h"
+#include "../AsyncStreams/AsyncStreams.h"
 #include "../Foundation/StringSpan.h"
 
 namespace SC
@@ -120,7 +118,7 @@ struct SC_COMPILER_EXPORT HttpConnection
     HttpRequest  request;
     HttpResponse response;
 
-  private:
+  protected:
     enum class State
     {
         Inactive,
@@ -131,18 +129,47 @@ struct SC_COMPILER_EXPORT HttpConnection
 
     State state = State::Inactive;
     ID    connectionID;
+};
+
+/// @brief View over a contiguous sequence of items with a custom stride between elements.
+template <typename Type>
+struct SC_COMPILER_EXPORT SpanWithStride
+{
+    /// @brief Builds an empty SpanWithStride
+    constexpr SpanWithStride() : data(nullptr), sizeElements(0), strideInBytes(0) {}
+
+    /// @brief Builds a SpanWithStride from data, size, and stride
+    /// @param data pointer to the first element
+    /// @param sizeInElements number of elements
+    /// @param strideInBytes stride between elements in bytes
+    constexpr SpanWithStride(void* data, size_t sizeInElements, size_t strideInBytes)
+        : data(data), sizeElements(sizeInElements), strideInBytes(strideInBytes)
+    {}
+
+    [[nodiscard]] constexpr size_t sizeInElements() const { return sizeElements; }
+    [[nodiscard]] constexpr bool   empty() const { return sizeElements == 0; }
 
-    ReadableSocketStream readableSocketStream;
-    WritableSocketStream writableSocketStream;
+    [[nodiscard]] Type& operator[](size_t idx)
+    {
+        return *reinterpret_cast<Type*>(static_cast<char*>(data) + idx * strideInBytes);
+    }
+
+    [[nodiscard]] const Type& operator[](size_t idx) const
+    {
+        return *reinterpret_cast<const Type*>(static_cast<const char*>(data) + idx * strideInBytes);
+    }
 
-    SocketDescriptor socket;
+  private:
+    void*  data;
+    size_t sizeElements;
+    size_t strideInBytes;
 };
 
 /// @brief A pool of HttpConnection that can be active or inactive
 struct SC_COMPILER_EXPORT HttpConnectionsPool
 {
     /// @brief Initializes the server with memory buffers for connections and headers
-    Result init(Span<HttpConnection> connectionsStorage, Span<char> headersMemoryStorage);
+    Result init(SpanWithStride<HttpConnection> connectionsStorage, Span<char> headersMemoryStorage);
 
     /// @brief Closes the server, removing references to the memory buffers passed during init
     Result close();
@@ -169,10 +196,10 @@ struct SC_COMPILER_EXPORT HttpConnectionsPool
     [[nodiscard]] bool deactivate(HttpConnection::ID connectionID);
 
   private:
-    Span<HttpConnection> connections;
-    Span<char>           headersMemory;
+    SpanWithStride<HttpConnection> connections;
 
-    size_t numConnections = 0;
+    Span<char> headersMemory;
+    size_t     numConnections = 0;
 };
 //! @}
 

Support/Dependencies/Dependencies.json

@@ -150,7 +150,6 @@
     },
     "Http": {
         "direct_dependencies": [
-            "Async",
             "AsyncStreams",
             "FileSystem",
             "Foundation"