diff --git a/crates/bindings-csharp/Runtime/AuthCtx.cs b/crates/bindings-csharp/Runtime/AuthCtx.cs index a99ebe7d21c..c7fcdea47e0 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,13 +49,17 @@ 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 jwt = System.Text.Encoding.UTF8.GetString(bytes); - return jwt != null ? new JwtClaims(jwt, identity) : null; + var jwt = System.Text.Encoding.UTF8.GetString( + jwtBuffer, + 0, + checked((int)stream.Length) + ); + return new JwtClaims(jwt, identity); } ); } diff --git a/crates/bindings-csharp/Runtime/Http.cs b/crates/bindings-csharp/Runtime/Http.cs index 0a6bd712de0..3482f1affd7 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,11 @@ out var out_ { case Errno.OK: { - var responseWireBytes = out_.A.Consume(); - var responseWire = FromBytes(new HttpResponseWire.BSATN(), responseWireBytes); + using var responseWireStream = out_.A.Consume(ref responseWireBuffer); + var responseWire = FromBytes(new HttpResponseWire.BSATN(), responseWireStream); - var body = new HttpBody(out_.B.Consume()); + using var responseBodyStream = out_.B.Consume(ref responseBodyBuffer); + var body = new HttpBody(responseBodyStream.ToArray()); var (statusCode, version, headers) = FromWireResponse(responseWire); return Result.Ok( @@ -353,8 +357,8 @@ out var out_ } case Errno.HTTP_ERROR: { - var errorWireBytes = out_.A.Consume(); - var err = FromBytes(new SpacetimeDB.BSATN.String(), errorWireBytes); + using var errorWireStream = out_.A.Consume(ref errorWireBuffer); + var err = FromBytes(new SpacetimeDB.BSATN.String(), errorWireStream); return Result.Err(new HttpError(err)); } case Errno.WOULD_BLOCK_TRANSACTION: @@ -378,9 +382,8 @@ out var out_ } } - private static T FromBytes(IReadWrite rw, byte[] bytes) + private static T FromBytes(IReadWrite rw, MemoryStream ms) { - using var ms = new MemoryStream(bytes); using var reader = new BinaryReader(ms); var value = rw.Read(reader); if (ms.Position != ms.Length) diff --git a/crates/bindings-csharp/Runtime/Internal/FFI.cs b/crates/bindings-csharp/Runtime/Internal/FFI.cs index 380f10ba5f0..afc56abbc8f 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 1bfcc05cd38..b0223486059 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 658be37ea43..51e6fa0e6ee 100644 --- a/crates/bindings-csharp/Runtime/Internal/ITable.cs +++ b/crates/bindings-csharp/Runtime/Internal/ITable.cs @@ -1,30 +1,23 @@ 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); - while (true) + public IEnumerator GetEnumerator() + { + IterStart(out var handle); + var buffer = ArrayPool.Shared.Rent(InitialBufferSize); + try + { + while (handle != FFI.RowIter.INVALID) { var requestedLen = buffer.Length; var bufferLen = requestedLen; @@ -38,77 +31,44 @@ public bool MoveNext() System.Diagnostics.Debug.Assert(!(ret == Errno.OK && bufferLen == 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, bufferLen); - return bufferLen != 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, + bufferLen, + 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(bufferLen); - 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 @@ -161,7 +121,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 d850b930c77..83e04453964 100644 --- a/crates/bindings-csharp/Runtime/Internal/Module.cs +++ b/crates/bindings-csharp/Runtime/Internal/Module.cs @@ -440,10 +440,17 @@ public static void RegisterExplicitIndexName(string sourceName, string canonical moduleDef.RegisterExplicitIndexName(sourceName, canonicalName); public static byte[] Consume(this BytesSource source) + { + var buffer = Array.Empty(); + using var stream = source.Consume(ref buffer); + return stream.ToArray(); + } + + internal static MemoryStream Consume(this BytesSource source, ref byte[] buffer) { if (source == BytesSource.INVALID) { - return []; + return new(); } var len = (uint)0; @@ -458,36 +465,32 @@ public static byte[] Consume(this BytesSource source) throw new UnknownException(ret); } - var buffer = new byte[checked((int)len)]; + var requiredLen = checked((int)len); + if (buffer.Length < requiredLen) + { + Array.Resize(ref buffer, requiredLen); + } + var written = 0; - // 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(written); var buf_len = 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, written); - return buffer; - // Wrote the entire spare capacity. - // Need to reserve more space in the buffer. + return new MemoryStream( + buffer, + 0, + written, + writable: false, + publiclyVisible: true + ); 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: break; case Errno.NO_SUCH_BYTES: @@ -508,6 +511,14 @@ private static void Write(this BytesSink sink, ReadOnlySpan 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) @@ -550,7 +561,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[id].Invoke(reader, ctx); if (stream.Position != stream.Length) @@ -592,7 +603,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[id].Invoke(reader, ctx); if (stream.Position != stream.Length) @@ -628,8 +639,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) @@ -637,8 +647,12 @@ BytesSink responseBodySink throw new Exception("Unrecognised extra bytes in the HTTP handler request"); } + using var requestBodyStream = requestBody.Consume(ref httpRequestBodyBuffer); var response = httpHandlers[id] - .Invoke(ctx, SpacetimeDB.HttpClient.FromWire(requestWire, requestBody.Consume())); + .Invoke( + ctx, + SpacetimeDB.HttpClient.FromWire(requestWire, requestBodyStream.ToArray()) + ); var (responseWire, responseBody) = SpacetimeDB.HttpClient.ToWire(response); responseSink.Write( IStructuralReadWrite.ToBytes(new HttpResponseWire.BSATN(), responseWire) @@ -694,7 +708,7 @@ BytesSink rows MemoryMarshal.AsBytes([sender_0, sender_1, sender_2, sender_3]) ); 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[id].Invoke(reader, ctx); rows.Write(bytes); @@ -732,7 +746,7 @@ public static Errno __call_view_anon__(int 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[id].Invoke(reader, ctx); rows.Write(bytes);