AsyncStreams: Wrap Strings / Buffers in AsyncBufferView though type erasure

Strings/Buffers are copied (or moved) to the inline storage provided by AsyncBufferView and destroyed when no longer needed.

Author: Pagghiu

Libraries/AsyncStreams/AsyncStreams.cpp

@@ -28,6 +28,7 @@ void AsyncBuffersPool::unrefBuffer(AsyncBufferView::ID bufferID)
         {
         case AsyncBufferView::Type::Writable: buffer->writableData = buffer->originalWritableData; break;
         case AsyncBufferView::Type::ReadOnly: *buffer = {}; break;
+        case AsyncBufferView::Type::Growable: *buffer = {}; break;
         case AsyncBufferView::Type::Empty: Assert::unreachable(); break;
         }
     }
@@ -41,6 +42,14 @@ Result AsyncBuffersPool::getReadableData(AsyncBufferView::ID bufferID, Span<cons
     {
     case AsyncBufferView::Type::Writable: data = buffer->writableData; break;
     case AsyncBufferView::Type::ReadOnly: data = buffer->readonlyData; break;
+    case AsyncBufferView::Type::Growable: {
+        AsyncBufferView::GrowableStorage storage;
+
+        auto da = buffer->getGrowableBuffer(storage, true)->getDirectAccess();
+        data    = {static_cast<char*>(da.data), da.sizeInBytes};
+        (void)buffer->getGrowableBuffer(storage, false); // destruct
+        break;
+    }
     case AsyncBufferView::Type::Empty: Assert::unreachable(); break;
     }
     return Result(true);
@@ -77,6 +86,16 @@ Result AsyncBuffersPool::requestNewBuffer(size_t minimumSizeInBytes, AsyncBuffer
             {
             case AsyncBufferView::Type::Writable: buffer.originalWritableData = buffer.writableData; break;
             case AsyncBufferView::Type::ReadOnly: buffer.originalReadonlyData = buffer.readonlyData; break;
+            case AsyncBufferView::Type::Growable: {
+                AsyncBufferView::GrowableStorage storage;
+
+                auto da = buffer.getGrowableBuffer(storage, true)->getDirectAccess();
+
+                buffer.writableData         = {static_cast<char*>(da.data), da.sizeInBytes};
+                buffer.originalWritableData = {static_cast<char*>(da.data), da.sizeInBytes};
+                (void)buffer.getGrowableBuffer(storage, false); // destruct
+                break;
+            }
             case AsyncBufferView::Type::Empty: SC_ASSERT_RELEASE(false); break;
             }
             bufferID = AsyncBufferView::ID(static_cast<AsyncBufferView::ID::NumericType>(&buffer - buffers.begin()));
@@ -105,6 +124,22 @@ void AsyncBuffersPool::setNewBufferSize(AsyncBufferView::ID bufferID, size_t new
                 buffer->readonlyData = {buffer->readonlyData.data(), newSizeInBytes};
             }
             break;
+        case AsyncBufferView::Type::Growable: {
+            AsyncBufferView::GrowableStorage storage;
+
+            IGrowableBuffer* growable = buffer->getGrowableBuffer(storage, true);
+            if (growable->resizeWithoutInitializing(newSizeInBytes))
+            {
+                auto da = growable->getDirectAccess();
+
+                buffer->writableData         = {static_cast<char*>(da.data), da.sizeInBytes};
+                buffer->originalWritableData = {static_cast<char*>(da.data), da.sizeInBytes};
+            }
+            (void)buffer->getGrowableBuffer(storage, false); // destruct
+            break;
+        }
+
+        break;
         case AsyncBufferView::Type::Empty: SC_ASSERT_RELEASE(false); break;
         }
     }

Libraries/AsyncStreams/AsyncStreams.h

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: MIT
 #pragma once
 
+#include "../Foundation/AlignedStorage.h"
 #include "../Foundation/Function.h"
+#include "../Foundation/Internal/IGrowableBuffer.h"
 #include "../Foundation/Result.h"
 #include "../Foundation/Span.h"
 #include "Internal/CircularQueue.h"
@@ -61,12 +63,39 @@ struct AsyncBufferView
         Empty,
         Writable,
         ReadOnly,
+        Growable,
     };
 
     AsyncBufferView() { type = Type::Empty; }
     AsyncBufferView(Span<char> data) : writableData(data) { type = Type::Writable; }
     AsyncBufferView(Span<const char> data) : readonlyData(data) { type = Type::ReadOnly; }
 
+    /// @brief Saves a copy (or a moved instance) of a String / Buffer (or anything that works with GrowableBuffer<T>)
+    /// inside an AsyncBufferView in order to access its data later, as long as its size fits inside the inline storage.
+    /// Destroying the AsyncBufferView will also destroy the copied / moved instance.
+    template <typename T>
+    AsyncBufferView(T&& t) // universal reference, it can capture both lvalue and rvalue
+    {
+        type = Type::Growable;
+        // Here we're type-erasing T in our own inline storage provided by a slightly oversized Function<>
+        // that it will be able to construct (and destruct) the right GrowableBuffer<T> from just a piece of storage
+        // and return a pointer to the corresponding IGrowableBuffer* interface
+        getGrowableBuffer = [t = forward<T>(t)](GrowableStorage& storage, bool construct) mutable -> IGrowableBuffer*
+        {
+            using Type = typename TypeTraits::RemoveReference<T>::type;
+            if (construct)
+            {
+                placementNew(storage.reinterpret_as<GrowableBuffer<Type>>(), t);
+                return &storage.reinterpret_as<GrowableBuffer<Type>>();
+            }
+            else
+            {
+                dtor(storage.reinterpret_as<GrowableBuffer<Type>>());
+                return nullptr;
+            }
+        };
+    }
+
     template <int N>
     AsyncBufferView(const char (&literal)[N])
     {
@@ -77,6 +106,12 @@ struct AsyncBufferView
     Type getType() const { return type; }
 
   private:
+    static constexpr int TypeErasedCaptureSize  = sizeof(void*) * 3; // This is enough to hold String / Buffer by copy
+    static constexpr int TypeErasedGrowableSize = sizeof(void*) * 6;
+
+    using GrowableStorage = AlignedStorage<TypeErasedGrowableSize>;
+    Function<IGrowableBuffer*(GrowableStorage&, bool), TypeErasedCaptureSize> getGrowableBuffer;
+
     union
     {
         Span<char>       writableData;

Tests/Libraries/AsyncStreams/AsyncStreamsTest.cpp

@@ -323,19 +323,21 @@ void SC::AsyncStreamsTest::writableStream()
     };
     int numDrain = 0;
     (void)writable.eventDrain.addListener([&numDrain] { numDrain++; });
-    SC_TEST_EXPECT(writable.write("1")); // Executes asyncWrites and queue slot is freed immediately
+
+    // When passing String(...) the writable takes ownership of the String destroying it after the write
+    SC_TEST_EXPECT(writable.write(String("1"))); // Executes asyncWrites and queue slot is freed immediately
     SC_TEST_EXPECT(context.numAsyncWrites == 1);
-    SC_TEST_EXPECT(writable.write("2"));     // queued, uses first write slot
-    SC_TEST_EXPECT(writable.write("3"));     // queued, uses second write slot
-    SC_TEST_EXPECT(not writable.write("4")); // no more write queue slots
+    SC_TEST_EXPECT(writable.write("2"));         // queued, uses first write slot
+    SC_TEST_EXPECT(writable.write(String("3"))); // queued, uses second write slot
+    SC_TEST_EXPECT(not writable.write("4"));     // no more write queue slots
     SC_TEST_EXPECT(context.numAsyncWrites == 1);
     writable.finishedWriting(context.bufferID, {}, Result(true)); // writes 2
     SC_TEST_EXPECT(context.concatenated == "12");
     SC_TEST_EXPECT(numDrain == 0);
     SC_TEST_EXPECT(context.numAsyncWrites == 2);
     SC_TEST_EXPECT(writable.write("4"));
     SC_TEST_EXPECT(context.numAsyncWrites == 2);
-    SC_TEST_EXPECT(not writable.write("5"));
+    SC_TEST_EXPECT(not writable.write(String("5")));
     writable.finishedWriting(context.bufferID, {}, Result(true)); // writes 3
     SC_TEST_EXPECT(context.concatenated == "123");
     SC_TEST_EXPECT(numDrain == 0);
@@ -348,10 +350,10 @@ void SC::AsyncStreamsTest::writableStream()
     SC_TEST_EXPECT(context.numAsyncWrites == 4);
     SC_TEST_EXPECT(writable.write("5"));
     SC_TEST_EXPECT(context.numAsyncWrites == 5);
-    SC_TEST_EXPECT(writable.write("6"));
+    SC_TEST_EXPECT(writable.write(String("6")));
     SC_TEST_EXPECT(context.numAsyncWrites == 5);
     SC_TEST_EXPECT(writable.write("7"));
-    SC_TEST_EXPECT(not writable.write("8"));
+    SC_TEST_EXPECT(not writable.write(String("8")));
     writable.finishedWriting(context.bufferID, {}, Result(true));
     SC_TEST_EXPECT(context.concatenated == "123456");
     SC_TEST_EXPECT(context.numAsyncWrites == 6);