From c72d4a4d93d488dcd97b2c4c2f1299f7fa9a2d0c Mon Sep 17 00:00:00 2001 From: Lisandro Crespo Date: Thu, 9 Jul 2026 14:37:41 -0500 Subject: [PATCH] Bring #2151 to post .NET 10 --- crates/bindings-csharp/Runtime/AuthCtx.cs | 7 +- crates/bindings-csharp/Runtime/Http.cs | 9 +- .../bindings-csharp/Runtime/Internal/FFI.cs | 5 + .../Runtime/Internal/IIndex.cs | 12 +- .../Runtime/Internal/ITable.cs | 110 ++++++------------ .../Runtime/Internal/Module.cs | 59 +++++----- 6 files changed, 86 insertions(+), 116 deletions(-) diff --git a/crates/bindings-csharp/Runtime/AuthCtx.cs b/crates/bindings-csharp/Runtime/AuthCtx.cs index a99ebe7d21c..7bf844719fc 100644 --- a/crates/bindings-csharp/Runtime/AuthCtx.cs +++ b/crates/bindings-csharp/Runtime/AuthCtx.cs @@ -4,6 +4,8 @@ namespace SpacetimeDB; public sealed class AuthCtx { + private static byte[] jwtBuffer = new byte[0x10_000]; + private readonly bool _isInternal; private readonly Lazy _jwtLazy; @@ -47,11 +49,12 @@ private static AuthCtx FromConnectionId(ConnectionId connectionId, Identity iden { var result = SpacetimeDB.Internal.FFI.get_jwt(ref connectionId, out var source); SpacetimeDB.Internal.FFI.CheckedStatus.Marshaller.ConvertToManaged(result); - var bytes = SpacetimeDB.Internal.Module.Consume(source); - if (bytes == null || bytes.Length == 0) + using var stream = SpacetimeDB.Internal.Module.Consume(source, ref jwtBuffer); + if (stream.Length == 0) { return null; } + var bytes = stream.ToArray(); var jwt = System.Text.Encoding.UTF8.GetString(bytes); return jwt != null ? new JwtClaims(jwt, identity) : null; } diff --git a/crates/bindings-csharp/Runtime/Http.cs b/crates/bindings-csharp/Runtime/Http.cs index 2567f4a11e5..78850782590 100644 --- a/crates/bindings-csharp/Runtime/Http.cs +++ b/crates/bindings-csharp/Runtime/Http.cs @@ -155,6 +155,9 @@ public sealed class HttpError(string message) : Exception(message) public sealed class HttpClient { private static readonly TimeSpan MaxTimeout = TimeSpan.FromMilliseconds(500); + private static byte[] responseWireBuffer = new byte[0x10_000]; + private static byte[] responseBodyBuffer = new byte[0x10_000]; + private static byte[] errorWireBuffer = new byte[0x10_000]; /// /// Send a simple GET request to with no headers. @@ -341,10 +344,10 @@ out var out_ { case Errno.OK: { - var responseWireBytes = out_.A.Consume(); + var responseWireBytes = out_.A.Consume(ref responseWireBuffer).ToArray(); var responseWire = FromBytes(new HttpResponseWire.BSATN(), responseWireBytes); - var body = new HttpBody(out_.B.Consume()); + var body = new HttpBody(out_.B.Consume(ref responseBodyBuffer).ToArray()); var (statusCode, version, headers) = FromWireResponse(responseWire); return Result.Ok( @@ -353,7 +356,7 @@ out var out_ } case Errno.HTTP_ERROR: { - var errorWireBytes = out_.A.Consume(); + var errorWireBytes = out_.A.Consume(ref errorWireBuffer).ToArray(); var err = FromBytes(new SpacetimeDB.BSATN.String(), errorWireBytes); return Result.Err(new HttpError(err)); } diff --git a/crates/bindings-csharp/Runtime/Internal/FFI.cs b/crates/bindings-csharp/Runtime/Internal/FFI.cs index c9f07994f1d..96984a4f281 100644 --- a/crates/bindings-csharp/Runtime/Internal/FFI.cs +++ b/crates/bindings-csharp/Runtime/Internal/FFI.cs @@ -50,6 +50,11 @@ public enum Errno : short HTTP_ERROR = 21, } +internal static class ErrnoExtensions +{ + public static void Check(this Errno status) => FFI.ErrnoHelpers.ThrowIfError(status); +} + #pragma warning disable IDE1006 // Naming Styles - Not applicable to FFI stuff. internal static partial class FFI { diff --git a/crates/bindings-csharp/Runtime/Internal/IIndex.cs b/crates/bindings-csharp/Runtime/Internal/IIndex.cs index 33525924f2d..5b0b9e330f4 100644 --- a/crates/bindings-csharp/Runtime/Internal/IIndex.cs +++ b/crates/bindings-csharp/Runtime/Internal/IIndex.cs @@ -44,7 +44,7 @@ out ReadOnlySpan rend } protected IEnumerable DoFilter(Bounds bounds) - where Bounds : IBTreeIndexBounds => new RawTableIter(indexId, bounds).Parse(); + where Bounds : IBTreeIndexBounds => new RawTableIter(indexId, bounds); protected uint DoDelete(Bounds bounds) where Bounds : IBTreeIndexBounds @@ -129,7 +129,7 @@ out var numDeleted new RW().Write(w, key); var point = s.ToArray(); - using var e = new RawPointIter(indexId, point).Parse().GetEnumerator(); + using var e = new RawPointIter(indexId, point).GetEnumerator(); if (!e.MoveNext()) { return null; @@ -192,7 +192,7 @@ out var numDeleted new RW().Write(w, key); var point = s.ToArray(); - using var e = new RawPointIter(indexId, point).Parse().GetEnumerator(); + using var e = new RawPointIter(indexId, point).GetEnumerator(); if (!e.MoveNext()) { return null; @@ -241,7 +241,7 @@ protected override void IterStart(out FFI.RowIter handle) => new RW().Write(w, key); var point = s.ToArray(); - using var e = new RawPointIter(indexId, point).Parse().GetEnumerator(); + using var e = new RawPointIter(indexId, point).GetEnumerator(); if (!e.MoveNext()) { return null; @@ -280,7 +280,7 @@ protected override void IterStart(out FFI.RowIter handle) => new RW().Write(w, key); var point = s.ToArray(); - using var e = new RawPointIter(indexId, point).Parse().GetEnumerator(); + using var e = new RawPointIter(indexId, point).GetEnumerator(); if (!e.MoveNext()) { return null; @@ -319,5 +319,5 @@ protected ulong DoCount() return count; } - protected IEnumerable DoIter() => new TableIter(tableId).Parse(); + protected IEnumerable DoIter() => new TableIter(tableId); } diff --git a/crates/bindings-csharp/Runtime/Internal/ITable.cs b/crates/bindings-csharp/Runtime/Internal/ITable.cs index a322216a1d3..7b284f5eb66 100644 --- a/crates/bindings-csharp/Runtime/Internal/ITable.cs +++ b/crates/bindings-csharp/Runtime/Internal/ITable.cs @@ -1,34 +1,26 @@ namespace SpacetimeDB.Internal; using System.Buffers; +using System.Collections; using SpacetimeDB.BSATN; -internal abstract class RawTableIterBase +internal abstract class RawTableIterBase : IEnumerable where T : IStructuralReadWrite, new() { - public sealed class Enumerator(FFI.RowIter handle) : IDisposable - { - private const int InitialBufferSize = 1024; - private byte[]? buffer = ArrayPool.Shared.Rent(InitialBufferSize); - public ArraySegment Current { get; private set; } = ArraySegment.Empty; - - public bool MoveNext() - { - if (handle == FFI.RowIter.INVALID) - { - return false; - } + private const int InitialBufferSize = 1024; - if (buffer is null) - { - return false; - } + protected abstract void IterStart(out FFI.RowIter handle); - uint buffer_len; - while (true) + public IEnumerator GetEnumerator() + { + IterStart(out var handle); + var buffer = ArrayPool.Shared.Rent(InitialBufferSize); + try + { + while (handle != FFI.RowIter.INVALID) { var requested_len = (uint)buffer.Length; - buffer_len = requested_len; + var buffer_len = requested_len; var ret = FFI.row_iter_bsatn_advance(handle, buffer, ref buffer_len); if (ret == Errno.EXHAUSTED) { @@ -38,82 +30,50 @@ public bool MoveNext() buffer_len = 0; } } + // On success, the only way `buffer_len == 0` is for the iterator to be exhausted. // This happens when the host iterator was empty from the start. System.Diagnostics.Debug.Assert(!(ret == Errno.OK && buffer_len == 0)); switch (ret) { - // Iterator advanced and may also be `EXHAUSTED`. - // When `OK`, we'll need to advance the iterator in the next call to `MoveNext`. - // In both cases, update `Current` to point at the valid range in the scratch `buffer`. case Errno.EXHAUSTED or Errno.OK: - Current = new ArraySegment(buffer, 0, (int)buffer_len); - return buffer_len != 0; - // Couldn't find the iterator, error! - case Errno.NO_SUCH_ITER: - throw new NoSuchIterException(); - // The scratch `buffer` is too small to fit a row / chunk. - // Grow `buffer` and try again. - // The `buffer_len` will have been updated with the necessary size. + { + using var stream = new MemoryStream( + buffer, + 0, + (int)buffer_len, + writable: false, + publiclyVisible: true + ); + using var reader = new BinaryReader(stream); + while (stream.Position < stream.Length) + { + yield return IStructuralReadWrite.Read(reader); + } + break; + } case Errno.BUFFER_TOO_SMALL: ArrayPool.Shared.Return(buffer); buffer = ArrayPool.Shared.Rent((int)buffer_len); - continue; + break; default: - throw new UnknownException(ret); + ret.Check(); + break; } } } - - public void Dispose() + finally { if (handle != FFI.RowIter.INVALID) { FFI.row_iter_bsatn_close(handle); - handle = FFI.RowIter.INVALID; } - - if (buffer is not null) - { - ArrayPool.Shared.Return(buffer); - buffer = null; - } - } - - public void Reset() - { - throw new NotImplementedException(); + ArrayPool.Shared.Return(buffer); } } - protected abstract void IterStart(out FFI.RowIter handle); - - // Note: using the GetEnumerator() duck-typing protocol instead of IEnumerable to avoid extra boxing. - public Enumerator GetEnumerator() - { - IterStart(out var handle); - return new(handle); - } - - public IEnumerable Parse() - { - foreach (var chunk in this) - { - using var stream = new MemoryStream( - chunk.Array!, - chunk.Offset, - chunk.Count, - writable: false, - publiclyVisible: true - ); - using var reader = new BinaryReader(stream); - while (stream.Position < stream.Length) - { - yield return IStructuralReadWrite.Read(reader); - } - } - } + IEnumerator IEnumerable.GetEnumerator() => GetEnumerator(); } public interface ITableView @@ -166,7 +126,7 @@ protected static ulong DoCount() return count; } - protected static IEnumerable DoIter() => new RawTableIter(tableId).Parse(); + protected static IEnumerable DoIter() => new RawTableIter(tableId); protected static T DoInsert(T row) { diff --git a/crates/bindings-csharp/Runtime/Internal/Module.cs b/crates/bindings-csharp/Runtime/Internal/Module.cs index eebd573cfc5..14653026220 100644 --- a/crates/bindings-csharp/Runtime/Internal/Module.cs +++ b/crates/bindings-csharp/Runtime/Internal/Module.cs @@ -439,11 +439,11 @@ public static void RegisterExplicitFunctionName(string sourceName, string canoni public static void RegisterExplicitIndexName(string sourceName, string canonicalName) => moduleDef.RegisterExplicitIndexName(sourceName, canonicalName); - public static byte[] Consume(this BytesSource source) + internal static MemoryStream Consume(this BytesSource source, ref byte[] buffer) { if (source == BytesSource.INVALID) { - return []; + return new(); } var len = (uint)0; @@ -458,42 +458,28 @@ public static byte[] Consume(this BytesSource source) throw new UnknownException(ret); } - var buffer = new byte[len]; + if (buffer.Length < len) + { + Array.Resize(ref buffer, (int)len); + } + var written = 0U; - // Because we've reserved space in our buffer already, this loop should be unnecessary. - // We expect the first call to `bytes_source_read` to always return `-1`. - // I (pgoldman 2025-09-26) am leaving the loop here because there's no downside to it, - // and in the future we may want to support `BytesSource`s which don't have a known length ahead of time - // (i.e. put arbitrary streams in `BytesSource` on the host side rather than just `Bytes` buffers), - // at which point the loop will become useful again. while (true) { - // Write into the spare capacity of the buffer. var spare = buffer.AsSpan((int)written); var buf_len = (uint)spare.Length; ret = FFI.bytes_source_read(source, spare, ref buf_len); written += buf_len; switch (ret) { - // Host side source exhausted, we're done. case Errno.EXHAUSTED: - Array.Resize(ref buffer, (int)written); - return buffer; - // Wrote the entire spare capacity. - // Need to reserve more space in the buffer. + return new(buffer, 0, (int)written); case Errno.OK when written == buffer.Length: Array.Resize(ref buffer, buffer.Length + 1024); break; - // Host didn't write as much as possible. - // Try to read some more. - // The host will likely not trigger this branch (current host doesn't), - // but a module should be prepared for it. case Errno.OK: + ret.Check(); break; - case Errno.NO_SUCH_BYTES: - throw new NoSuchBytesException(); - default: - throw new UnknownException(ret); } } } @@ -510,6 +496,14 @@ private static void Write(this BytesSink sink, byte[] bytes) } } + // __call_reducer__ is not invoked in parallel because modules do not support multithreading in Wasm. + private static byte[] reducerArgsBuffer = new byte[0x10_000]; + private static byte[] procedureArgsBuffer = new byte[0x10_000]; + private static byte[] httpRequestBuffer = new byte[0x10_000]; + private static byte[] httpRequestBodyBuffer = new byte[0x10_000]; + private static byte[] viewArgsBuffer = new byte[0x10_000]; + private static byte[] anonymousViewArgsBuffer = new byte[0x10_000]; + #pragma warning disable IDE1006 // Naming Styles - methods below are meant for FFI. public static void __describe_module__(BytesSink description) @@ -554,7 +548,7 @@ BytesSink error var ctx = newReducerContext!(senderIdentity, connectionId, random, time); - using var stream = new MemoryStream(args.Consume()); + using var stream = args.Consume(ref reducerArgsBuffer); using var reader = new BinaryReader(stream); reducers[(int)id].Invoke(reader, ctx); if (stream.Position != stream.Length) @@ -598,7 +592,7 @@ BytesSink resultSink var ctx = newProcedureContext!(sender, connectionId, random, time); - using var stream = new MemoryStream(args.Consume()); + using var stream = args.Consume(ref procedureArgsBuffer); using var reader = new BinaryReader(stream); var bytes = procedures[(int)id].Invoke(reader, ctx); if (stream.Position != stream.Length) @@ -634,8 +628,7 @@ BytesSink responseBodySink var time = timestamp.ToStd(); var ctx = newHandlerContext!(random, time); - var requestBytes = request.Consume(); - using var stream = new MemoryStream(requestBytes); + using var stream = request.Consume(ref httpRequestBuffer); using var reader = new BinaryReader(stream); var requestWire = new HttpRequestWire.BSATN().Read(reader); if (stream.Position != stream.Length) @@ -644,7 +637,13 @@ BytesSink responseBodySink } var response = httpHandlers[(int)id] - .Invoke(ctx, SpacetimeDB.HttpClient.FromWire(requestWire, requestBody.Consume())); + .Invoke( + ctx, + SpacetimeDB.HttpClient.FromWire( + requestWire, + requestBody.Consume(ref httpRequestBodyBuffer).ToArray() + ) + ); var (responseWire, responseBody) = SpacetimeDB.HttpClient.ToWire(response); responseSink.Write( IStructuralReadWrite.ToBytes(new HttpResponseWire.BSATN(), responseWire) @@ -700,7 +699,7 @@ BytesSink rows MemoryMarshal.AsBytes([sender_0, sender_1, sender_2, sender_3]).ToArray() ); var ctx = newViewContext!(sender); - using var stream = new MemoryStream(args.Consume()); + using var stream = args.Consume(ref viewArgsBuffer); using var reader = new BinaryReader(stream); var bytes = viewDispatchers[(int)id].Invoke(reader, ctx); rows.Write(bytes); @@ -738,7 +737,7 @@ public static Errno __call_view_anon__(uint id, BytesSource args, BytesSink rows try { var ctx = newAnonymousViewContext!(); - using var stream = new MemoryStream(args.Consume()); + using var stream = args.Consume(ref anonymousViewArgsBuffer); using var reader = new BinaryReader(stream); var bytes = anonymousViewDispatchers[(int)id].Invoke(reader, ctx); rows.Write(bytes);