diff --git a/sdks/csharp/src/BSATNHelpers.cs b/sdks/csharp/src/BSATNHelpers.cs index 562e112549f..8e905b029e3 100644 --- a/sdks/csharp/src/BSATNHelpers.cs +++ b/sdks/csharp/src/BSATNHelpers.cs @@ -1,4 +1,6 @@ using SpacetimeDB.BSATN; +using System.Buffers; +using System.Collections.Generic; using System.IO; namespace SpacetimeDB @@ -6,7 +8,7 @@ namespace SpacetimeDB public static class BSATNHelpers { /// - /// Decode an element of a BSATN-serializable type from a list of bytes. + /// Decode an element of a BSATN-serializable type from a byte array. /// /// This method performs several allocations. Prefer calling IStructuralReadWrite.Read(BinaryReader) when /// deserializing many items from a buffer. @@ -14,11 +16,15 @@ public static class BSATNHelpers /// /// /// - public static T Decode(System.Collections.Generic.List bsatn) where T : IStructuralReadWrite, new() => - Decode(bsatn.ToArray()); + public static T Decode(byte[] bsatn) where T : IStructuralReadWrite, new() + { + using var stream = new MemoryStream(bsatn); + using var reader = new BinaryReader(stream); + return IStructuralReadWrite.Read(reader); + } /// - /// Decode an element of a BSATN-serializable type from a byte array. + /// Decode an element of a BSATN-serializable type from a list of bytes. /// /// This method performs several allocations. Prefer calling IStructuralReadWrite.Read(BinaryReader) when /// deserializing many items from a buffer. @@ -26,11 +32,48 @@ public static class BSATNHelpers /// /// /// - public static T Decode(byte[] bsatn) where T : IStructuralReadWrite, new() + public static T Decode(System.Collections.Generic.List bsatn) where T : IStructuralReadWrite, new() { - using var stream = new MemoryStream(bsatn); + using var stream = MakePooledListStream(bsatn, out var pooledBuffer); + try + { + using var reader = new BinaryReader(stream); + return IStructuralReadWrite.Read(reader); + } + finally + { + ArrayPool.Shared.Return(pooledBuffer); + } + } + + /// + /// Decode an element of a BSATN-serializable type from a readonly byte list. + /// + /// + /// + /// + public static T Decode(IReadOnlyList bsatn) where T : IStructuralReadWrite, new() + { + if (bsatn is byte[] bytes) + { + return Decode(bytes); + } + + if (bsatn is System.Collections.Generic.List list) + { + return Decode(list); + } + + using var stream = new ListStream(bsatn); using var reader = new BinaryReader(stream); return IStructuralReadWrite.Read(reader); } + + public static MemoryStream MakePooledListStream(System.Collections.Generic.List bsatn, out byte[] pooledBuffer) + { + pooledBuffer = ArrayPool.Shared.Rent(bsatn.Count); + bsatn.CopyTo(pooledBuffer); + return new MemoryStream(pooledBuffer, 0, bsatn.Count, writable: false); + } } } diff --git a/sdks/csharp/src/CompressionHelpers.cs b/sdks/csharp/src/CompressionHelpers.cs index 832208938ed..3155c13973e 100644 --- a/sdks/csharp/src/CompressionHelpers.cs +++ b/sdks/csharp/src/CompressionHelpers.cs @@ -55,10 +55,14 @@ internal static ServerMessage DecompressDecodeMessage(byte[] bytes) // The stream will never be empty. It will at least contain the compression algo. var compression = (CompressionAlgos)stream.ReadByte(); - // Conditionally decompress and decode. + + if (compression == CompressionAlgos.None) + { + return new ServerMessage.BSATN().Read(new BinaryReader(stream)); + } + Stream decompressedStream = compression switch { - CompressionAlgos.None => stream, CompressionAlgos.Brotli => BrotliReader(stream), CompressionAlgos.Gzip => GzipReader(stream), _ => throw new InvalidOperationException("Unknown compression type"), @@ -67,10 +71,13 @@ internal static ServerMessage DecompressDecodeMessage(byte[] bytes) // TODO: consider pooling these. // DO NOT TRY TO TAKE THIS OUT. The BrotliStream ReadByte() implementation allocates an array // PER BYTE READ. You have to do it all at once to avoid that problem. - MemoryStream memoryStream = new MemoryStream(); - decompressedStream.CopyTo(memoryStream); - memoryStream.Seek(0, SeekOrigin.Begin); - return new ServerMessage.BSATN().Read(new BinaryReader(memoryStream)); + using (decompressedStream) + using (MemoryStream memoryStream = new MemoryStream()) + { + decompressedStream.CopyTo(memoryStream); + memoryStream.Seek(0, SeekOrigin.Begin); + return new ServerMessage.BSATN().Read(new BinaryReader(memoryStream)); + } } /// diff --git a/sdks/csharp/src/ListStream.cs b/sdks/csharp/src/ListStream.cs index a50b78fd8c0..bb3657d79f8 100644 --- a/sdks/csharp/src/ListStream.cs +++ b/sdks/csharp/src/ListStream.cs @@ -10,10 +10,10 @@ /// internal class ListStream : Stream { - private List list; + private IReadOnlyList list; private int pos; - public ListStream(List data) + public ListStream(IReadOnlyList data) { this.list = data; this.pos = 0; @@ -36,16 +36,28 @@ public override void Flush() public override int Read(byte[] buffer, int offset, int count) { + var readable = Math.Min(count, Math.Min(buffer.Length - offset, list.Count - pos)); + if (readable <= 0) + { + return 0; + } + + if (list is List byteList) + { + byteList.CopyTo(pos, buffer, offset, readable); + pos += readable; + return readable; + } + int listPos = pos; - int listEnd = Math.Min(list.Count, listPos + count); int bufPos = offset; - int bufLength = buffer.Length; - for (; listPos < listEnd && bufPos < bufLength; listPos++, bufPos++) + int listEnd = pos + readable; + for (; listPos < listEnd; listPos++, bufPos++) { buffer[bufPos] = list[listPos]; } - pos = listPos; - return bufPos - offset; + pos += readable; + return readable; } public override int Read(Span buffer) diff --git a/sdks/csharp/src/MultiDictionary.cs b/sdks/csharp/src/MultiDictionary.cs index 2225f336494..3196ee101ec 100644 --- a/sdks/csharp/src/MultiDictionary.cs +++ b/sdks/csharp/src/MultiDictionary.cs @@ -1,5 +1,4 @@ using System; -using System.Linq; using System.Text; using System.Collections.Generic; using System.Diagnostics; @@ -55,7 +54,18 @@ public static MultiDictionary FromEnumerable(IEnumerable /// Return the count WITH multiplicities. /// - public readonly uint Count => RawDict.Select(item => item.Value.Multiplicity).Aggregate(0u, (a, b) => a + b); + public readonly uint Count + { + get + { + uint count = 0; + foreach (var item in RawDict) + { + count += item.Value.Multiplicity; + } + return count; + } + } /// /// Add a key-value-pair to the multidictionary. @@ -113,7 +123,8 @@ public bool Contains(KeyValuePair item) /// /// /// - public uint Multiplicity(TKey key) => RawDict.ContainsKey(key) ? RawDict[key].Multiplicity : 0; + public uint Multiplicity(TKey key) => + RawDict.TryGetValue(key, out var result) ? result.Multiplicity : 0; /// /// The value associated with a key. @@ -171,8 +182,10 @@ public readonly IEnumerable Values { get { - - return RawDict.Select(item => item.Value.Value); + foreach (var item in RawDict) + { + yield return item.Value.Value; + } } } @@ -180,7 +193,10 @@ public readonly IEnumerable> Entries { get { - return RawDict.Select(item => new KeyValuePair(item.Key, item.Value.Value)); + foreach (var item in RawDict) + { + yield return new KeyValuePair(item.Key, item.Value.Value); + } } } @@ -191,31 +207,32 @@ public readonly IEnumerable> Entries /// public readonly IEnumerable> WillRemove(MultiDictionaryDelta delta) { - var self = this; - return delta.Entries.Where(their => + foreach (var their in delta.Entries) { if (their.Value.IsValueChange) { // Value changes are translated to Updates, not removals. - return false; + continue; } var theirNonValueChange = their.Value.NonValueChange; if (theirNonValueChange.Delta >= 0) { // Adds can't result in removals. - return false; + continue; } - if (self.RawDict.TryGetValue(their.Key, out var mine)) + if (RawDict.TryGetValue(their.Key, out var mine)) { var resultMultiplicity = (int)mine.Multiplicity + theirNonValueChange.Delta; - return resultMultiplicity <= 0; // if < 0, we have a problem, but that's caught in Apply. + if (resultMultiplicity <= 0) // if < 0, we have a problem, but that's caught in Apply. + { + yield return new KeyValuePair(their.Key, theirNonValueChange.Value); + } } else { Log.Warn($"Want to remove row with key {their.Key}, but it doesn't exist!"); - return false; } - }).Select(entry => new KeyValuePair(entry.Key, entry.Value.NonValueChange.Value)); + } } /// @@ -225,7 +242,7 @@ public readonly IEnumerable> WillRemove(MultiDictiona /// Will be populated with inserted KVPs. /// Will be populated with updated KVPs. /// Will be populated with removed KVPs. - public void Apply(MultiDictionaryDelta delta, List> wasInserted, List<(TKey Key, TValue OldValue, TValue NewValue)> wasUpdated, List> wasRemoved) + public readonly void Apply(MultiDictionaryDelta delta, List> wasInserted, List<(TKey Key, TValue OldValue, TValue NewValue)> wasUpdated, List> wasRemoved) { foreach (var (key, their) in delta.Entries) { @@ -723,4 +740,4 @@ public readonly IEnumerable> Entries } } } -} \ No newline at end of file +} diff --git a/sdks/csharp/src/ProcedureCallbacks.cs b/sdks/csharp/src/ProcedureCallbacks.cs index b39102dfa08..b542f2c4b7b 100644 --- a/sdks/csharp/src/ProcedureCallbacks.cs +++ b/sdks/csharp/src/ProcedureCallbacks.cs @@ -98,7 +98,7 @@ public void Invoke(IProcedureEventContext ctx, ProcedureResult result) var callbackResult = result.Status switch { ProcedureStatus.Returned(var bytes) => - ProcedureCallbackResult.Success(BSATNHelpers.Decode(bytes.ToArray())), + ProcedureCallbackResult.Success(BSATNHelpers.Decode(bytes)), ProcedureStatus.InternalError(var error) => ProcedureCallbackResult.Failure(new Exception($"Procedure failed: {error}")), _ => ProcedureCallbackResult.Failure(new Exception("Unknown procedure status")) diff --git a/sdks/csharp/src/SpacetimeDBClient.cs b/sdks/csharp/src/SpacetimeDBClient.cs index b726a15eefb..5e8e0b49057 100644 --- a/sdks/csharp/src/SpacetimeDBClient.cs +++ b/sdks/csharp/src/SpacetimeDBClient.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; using System.Collections.Concurrent; using System.Collections.Generic; using System.IO; @@ -404,13 +405,20 @@ ParsedDatabaseUpdate ParseTransactionUpdate(TransactionUpdate update) return dbOps; } - string DecodeReducerError(IReadOnlyList bytes) + string DecodeReducerError(List bytes) { try { - using var stream = new MemoryStream(bytes.ToArray()); - using var reader = new BinaryReader(stream); - return new SpacetimeDB.BSATN.String().Read(reader); + using var stream = BSATNHelpers.MakePooledListStream(bytes, out var pooledBuffer); + try + { + using var reader = new BinaryReader(stream); + return new SpacetimeDB.BSATN.String().Read(reader); + } + finally + { + ArrayPool.Shared.Return(pooledBuffer); + } } catch { diff --git a/sdks/csharp/src/Table.cs b/sdks/csharp/src/Table.cs index 063d45bfdbe..680dfc11b52 100644 --- a/sdks/csharp/src/Table.cs +++ b/sdks/csharp/src/Table.cs @@ -2,7 +2,6 @@ using System.Collections.Generic; using System.Diagnostics; using System.IO; -using System.Linq; using System.Threading.Tasks; #if UNITY_5_3_OR_NEWER using UnityEngine; @@ -154,7 +153,7 @@ public BTreeIndexBase(RemoteTableHandleBase table) } public IEnumerable Filter(Column value) => - cache.TryGetValue(value, out var rows) ? rows : Enumerable.Empty(); + cache.TryGetValue(value, out var rows) ? rows : Array.Empty(); } /// @@ -415,7 +414,7 @@ public event RowEventHandler OnInsert public int Count => (int)Entries.CountDistinct; - public IEnumerable Iter() => Entries.Entries.Select(entry => (Row)entry.Value); + public IEnumerable Iter() => Entries.Values; public Task RemoteQuery(string query) => conn.RemoteQuery($"SELECT {RemoteTableName}.* FROM {RemoteTableName} {query}"); @@ -504,42 +503,16 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa // in order to avoid keys an error with the same key already added. foreach (var (_, value) in wasRemoved) { - if (value is Row oldRow) - { - OnInternalDeleteHandler.Invoke(oldRow); - } + OnInternalDeleteHandler.Invoke(value); } foreach (var (_, value) in wasInserted) { - if (value is Row newRow) - { - OnInternalInsertHandler.Invoke(newRow); - } - else - { - throw new Exception($"Invalid row type for table {RemoteTableName}: {value.GetType().Name}"); - } + OnInternalInsertHandler.Invoke(value); } foreach (var (_, oldValue, newValue) in wasUpdated) { - if (oldValue is Row oldRow) - { - OnInternalDeleteHandler.Invoke(oldRow); - } - else - { - throw new Exception($"Invalid row type for table {RemoteTableName}: {oldValue.GetType().Name}"); - } - - - if (newValue is Row newRow) - { - OnInternalInsertHandler.Invoke(newRow); - } - else - { - throw new Exception($"Invalid row type for table {RemoteTableName}: {newValue.GetType().Name}"); - } + OnInternalDeleteHandler.Invoke(oldValue); + OnInternalInsertHandler.Invoke(newValue); } } diff --git a/sdks/csharp/src/WebSocket.cs b/sdks/csharp/src/WebSocket.cs index 8ae335db8f0..97703c6b716 100644 --- a/sdks/csharp/src/WebSocket.cs +++ b/sdks/csharp/src/WebSocket.cs @@ -3,7 +3,6 @@ using System; using System.Collections.Concurrent; -using System.Linq; using System.Net.Sockets; using System.Net.WebSockets; using System.Runtime.InteropServices; @@ -31,10 +30,11 @@ public struct ConnectOptions // WebSocket buffer for incoming messages private static readonly int MAXMessageSize = 0x4000000; // 64MB + private static readonly int InitialReceiveBufferSize = 16 * 1024; // Connection parameters private readonly ConnectOptions _options; - private readonly byte[] _receiveBuffer = new byte[MAXMessageSize]; + private byte[] _receiveBuffer = new byte[InitialReceiveBufferSize]; private readonly ConcurrentQueue dispatchQueue = new(); protected ClientWebSocket Ws = new(); @@ -365,15 +365,17 @@ await Ws.CloseAsync(WebSocketCloseStatus.MessageTooBig, closeMessage, return; } + EnsureReceiveCapacity(count + 1); receiveResult = await Ws.ReceiveAsync( - new ArraySegment(_receiveBuffer, count, MAXMessageSize - count), + new ArraySegment(_receiveBuffer, count, _receiveBuffer.Length - count), CancellationToken.None); count += receiveResult.Count; } if (OnMessage != null) { - var message = _receiveBuffer.Take(count).ToArray(); + var message = new byte[count]; + Buffer.BlockCopy(_receiveBuffer, 0, message, 0, count); // directly invoke message handling OnMessage(message, startReceive); } @@ -425,6 +427,23 @@ public Task Close(WebSocketCloseStatus code = WebSocketCloseStatus.NormalClosure return Task.CompletedTask; } + private void EnsureReceiveCapacity(int minimumCapacity) + { + if (_receiveBuffer.Length >= minimumCapacity) + { + return; + } + + var newCapacity = _receiveBuffer.Length; + do + { + newCapacity = Math.Min(newCapacity * 2, MAXMessageSize); + } + while (newCapacity < minimumCapacity); + + Array.Resize(ref _receiveBuffer, newCapacity); + } + /// /// Forcefully abort the WebSocket connection. This terminates any in-flight connect/receive/send /// and ensures the server-side socket is torn down promptly. Prefer Close() for graceful shutdowns. diff --git a/sdks/csharp/tests~/PerformanceBenchmarks/PerformanceBenchmarks.csproj b/sdks/csharp/tests~/PerformanceBenchmarks/PerformanceBenchmarks.csproj new file mode 100644 index 00000000000..31c1b075bd5 --- /dev/null +++ b/sdks/csharp/tests~/PerformanceBenchmarks/PerformanceBenchmarks.csproj @@ -0,0 +1,16 @@ + + + + Exe + net8.0 + enable + enable + false + + + + + + + + diff --git a/sdks/csharp/tests~/PerformanceBenchmarks/Program.cs b/sdks/csharp/tests~/PerformanceBenchmarks/Program.cs new file mode 100644 index 00000000000..522ad6b20d7 --- /dev/null +++ b/sdks/csharp/tests~/PerformanceBenchmarks/Program.cs @@ -0,0 +1,120 @@ +using BenchmarkDotNet.Attributes; +using BenchmarkDotNet.Jobs; +using BenchmarkDotNet.Running; +using SpacetimeDB; +using SpacetimeDB.BSATN; + +BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args); + +[MemoryDiagnoser] +[SimpleJob(RuntimeMoniker.Net80, launchCount: 1, warmupCount: 5, iterationCount: 10)] +public class ReducerProcedurePayloadBenchmarks +{ + private readonly PerfRows.BSATN rowsRW = new(); + private List rows = []; + private byte[] encodedBytes = []; + private List encodedList = []; + private IReadOnlyList encodedReadOnlyList = []; + + [Params(1, 100, 10_000)] + public int RowCount { get; set; } + + [GlobalSetup] + public void GlobalSetup() + { + rows = Enumerable.Range(0, RowCount).Select(i => new PerfRow { Id = (uint)i }).ToList(); + encodedBytes = IStructuralReadWrite.ToBytes(rowsRW, new PerfRows { Rows = rows }); + encodedList = encodedBytes.ToList(); + encodedReadOnlyList = encodedList; + } + + [Benchmark(Baseline = true)] + public PerfRows XUnitReducerProcedureShape() + { + var encoded = IStructuralReadWrite.ToBytes(rowsRW, new PerfRows { Rows = rows }).ToList(); + return BSATNHelpers.Decode(encoded); + } + + [Benchmark] + public PerfRows SerializeAndDecodeFromByteArray() + { + var encoded = IStructuralReadWrite.ToBytes(rowsRW, new PerfRows { Rows = rows }); + return BSATNHelpers.Decode(encoded); + } + + [Benchmark] + public List SerializeToListOnly() => + IStructuralReadWrite.ToBytes(rowsRW, new PerfRows { Rows = rows }).ToList(); + + [Benchmark] + public PerfRows DecodeFromListOnly() => + BSATNHelpers.Decode(encodedList); + + [Benchmark] + public PerfRows DecodeFromReadOnlyListOnly() => + BSATNHelpers.Decode(encodedReadOnlyList); + + [Benchmark] + public PerfRows DecodeFromByteArrayOnly() => + BSATNHelpers.Decode(encodedBytes); + + public sealed class PerfRow : IStructuralReadWrite, IEquatable + { + public uint Id; + + public void ReadFields(BinaryReader reader) + { + Id = new U32().Read(reader); + } + + public void WriteFields(BinaryWriter writer) + { + new U32().Write(writer, Id); + } + + public object GetSerializer() => new BSATN(); + + public bool Equals(PerfRow? other) => other != null && Id == other.Id; + + public override bool Equals(object? obj) => obj is PerfRow other && Equals(other); + + public override int GetHashCode() => Id.GetHashCode(); + + public readonly struct BSATN : IReadWrite + { + public PerfRow Read(BinaryReader reader) => new() { Id = new U32().Read(reader) }; + + public void Write(BinaryWriter writer, PerfRow value) => new U32().Write(writer, value.Id); + + public AlgebraicType GetAlgebraicType(ITypeRegistrar registrar) => throw new NotImplementedException(); + } + } + + public sealed class PerfRows : IStructuralReadWrite + { + public List Rows = []; + + public void ReadFields(BinaryReader reader) + { + Rows = new SpacetimeDB.BSATN.List().Read(reader); + } + + public void WriteFields(BinaryWriter writer) + { + new SpacetimeDB.BSATN.List().Write(writer, Rows); + } + + public object GetSerializer() => new BSATN(); + + public readonly struct BSATN : IReadWrite + { + public PerfRows Read(BinaryReader reader) => + new() { Rows = new SpacetimeDB.BSATN.List().Read(reader) }; + + public void Write(BinaryWriter writer, PerfRows value) => + new SpacetimeDB.BSATN.List().Write(writer, value.Rows); + + public AlgebraicType GetAlgebraicType(ITypeRegistrar registrar) => throw new NotImplementedException(); + } + } +} diff --git a/sdks/csharp/tests~/PerformanceTests.cs b/sdks/csharp/tests~/PerformanceTests.cs new file mode 100644 index 00000000000..bd7f3647fc1 --- /dev/null +++ b/sdks/csharp/tests~/PerformanceTests.cs @@ -0,0 +1,322 @@ +using System.Diagnostics; +using System.IO.Compression; +using SpacetimeDB; +using SpacetimeDB.BSATN; +using SpacetimeDB.ClientApi; +using Xunit; +using Xunit.Abstractions; + +public sealed class PerformanceTests +{ + private readonly ITestOutputHelper output; + + public PerformanceTests(ITestOutputHelper output) + { + this.output = output; + } + + [Theory] + [InlineData(1)] + [InlineData(100)] + [InlineData(10_000)] + public void ParseRowListBaseline(int rowCount) + { + var rowList = MakeRowList(rowCount); + + Measure($"ParseRowList rows={rowCount}", Iterations(rowCount), () => + { + var (reader, count) = CompressionHelpers.ParseRowList(rowList); + for (var i = 0; i < count; i++) + { + new PerfRow.BSATN().Read(reader); + } + }); + } + + [Theory] + [InlineData(1)] + [InlineData(100)] + [InlineData(10_000)] + public void MultiDictionaryApplyBaseline(int rowCount) + { + Measure($"MultiDictionary.Apply rows={rowCount}", Iterations(rowCount), () => + { + var dict = new MultiDictionary(EqualityComparer.Default, EqualityComparer.Default); + var delta = new MultiDictionaryDelta(EqualityComparer.Default, EqualityComparer.Default); + for (uint i = 0; i < rowCount; i++) + { + delta.Add(i, new PerfRow { Id = i }); + } + + var inserted = new List>(); + var updated = new List<(uint key, PerfRow oldValue, PerfRow newValue)>(); + var removed = new List>(); + dict.Apply(delta, inserted, updated, removed); + }); + } + + [Theory] + [InlineData(1)] + [InlineData(100)] + [InlineData(10_000)] + public void PayloadSerializationAndProcedureDecodeBaseline(int rowCount) + { + var rows = Enumerable.Range(0, rowCount).Select(i => new PerfRow { Id = (uint)i }).ToList(); + + Measure($"Reducer/procedure payload rows={rowCount}", Iterations(rowCount), () => + { + var encoded = IStructuralReadWrite.ToBytes(new PerfRows.BSATN(), new PerfRows { Rows = rows }).ToList(); + _ = BSATNHelpers.Decode(encoded); + }); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public void InitialConnectionDecodeBaseline(bool brotli) + { + var message = MakeInitialConnectionMessage(); + var bytes = EncodeServerMessage(message, brotli); + + Measure($"DecompressDecodeMessage initial_connection brotli={brotli}", 10_000, () => + { + _ = CompressionHelpers.DecompressDecodeMessage(bytes); + }); + } + + [Theory] + [InlineData(1, false)] + [InlineData(1, true)] + [InlineData(100, false)] + [InlineData(100, true)] + [InlineData(10_000, false)] + [InlineData(10_000, true)] + public void TransactionUpdateDecodeBaseline(int rowCount, bool brotli) + { + var message = MakeTransactionUpdateMessage(rowCount); + var bytes = EncodeServerMessage(message, brotli); + var decoded = CompressionHelpers.DecompressDecodeMessage(bytes); + + AssertTransactionUpdateRows(decoded, rowCount); + Measure($"DecompressDecodeMessage transaction_update rows={rowCount} brotli={brotli} compressedBytes={bytes.Length}", Iterations(rowCount), () => + { + _ = CompressionHelpers.DecompressDecodeMessage(bytes); + }); + } + + [Theory] + [InlineData(1)] + [InlineData(100)] + [InlineData(10_000)] + public void BrotliInflateToMemoryStreamBaseline(int rowCount) + { + var message = MakeTransactionUpdateMessage(rowCount); + var bytes = EncodeServerMessage(message, brotli: true); + + Measure($"Brotli inflate transaction_update rows={rowCount} compressedBytes={bytes.Length}", Iterations(rowCount), () => + { + using var input = new MemoryStream(bytes); + Assert.Equal((int)CompressionHelpers.CompressionAlgos.Brotli, input.ReadByte()); + using var brotli = new BrotliStream(input, CompressionMode.Decompress); + using var output = new MemoryStream(); + brotli.CopyTo(output); + }); + } + + [Theory] + [InlineData(1)] + [InlineData(100)] + [InlineData(10_000)] + public void BrotliAndUncompressedTransactionUpdateDecodeEquivalence(int rowCount) + { + var message = MakeTransactionUpdateMessage(rowCount); + + var decodedUncompressed = CompressionHelpers.DecompressDecodeMessage(EncodeServerMessage(message, brotli: false)); + var decodedBrotli = CompressionHelpers.DecompressDecodeMessage(EncodeServerMessage(message, brotli: true)); + + AssertTransactionUpdateRows(decodedUncompressed, rowCount); + AssertTransactionUpdateRows(decodedBrotli, rowCount); + Assert.Equal(GetTransactionUpdateRowsData(decodedUncompressed), GetTransactionUpdateRowsData(decodedBrotli)); + } + + private void Measure(string name, int iterations, Action action) + { + action(); + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + var gen0 = GC.CollectionCount(0); + var gen1 = GC.CollectionCount(1); + var allocatedBefore = GC.GetAllocatedBytesForCurrentThread(); + var stopwatch = Stopwatch.StartNew(); + + for (var i = 0; i < iterations; i++) + { + action(); + } + + stopwatch.Stop(); + var allocatedBytes = GC.GetAllocatedBytesForCurrentThread() - allocatedBefore; + var meanMicroseconds = stopwatch.Elapsed.TotalMilliseconds * 1000 / iterations; + var allocatedBytesPerOp = allocatedBytes / iterations; + + output.WriteLine( + $"{name}: mean={meanMicroseconds:F2}us allocated={allocatedBytesPerOp}B/op gen0={GC.CollectionCount(0) - gen0} gen1={GC.CollectionCount(1) - gen1} iterations={iterations}" + ); + } + + private static int Iterations(int rowCount) => + rowCount switch + { + <= 1 => 50_000, + <= 100 => 10_000, + _ => 1_000, + }; + + private static BsatnRowList MakeRowList(int rowCount) + { + var bytes = new List(rowCount * sizeof(uint)); + using var stream = new MemoryStream(); + using var writer = new BinaryWriter(stream); + var rowRW = new PerfRow.BSATN(); + for (uint i = 0; i < rowCount; i++) + { + rowRW.Write(writer, new PerfRow { Id = i }); + } + bytes.AddRange(stream.ToArray()); + return new BsatnRowList(new RowSizeHint.FixedSize(sizeof(uint)), bytes); + } + + private static ServerMessage MakeInitialConnectionMessage() => + new ServerMessage.InitialConnection(new InitialConnection + { + Identity = Identity.From(Convert.FromBase64String("l0qzG1GPRtC1mwr+54q98tv0325gozLc6cNzq4vrzqY=")), + Token = "token", + ConnectionId = ConnectionId.From(Convert.FromBase64String("Vd4dFzcEzhLHJ6uNL8VXFg==")) + ?? throw new InvalidDataException("connection id"), + }); + + private static ServerMessage MakeTransactionUpdateMessage(int rowCount) => + new ServerMessage.TransactionUpdate(new TransactionUpdate + { + QuerySets = + [ + new QuerySetUpdate + { + QuerySetId = new QuerySetId(1), + Tables = + [ + new TableUpdate + { + TableName = "perf_row", + Rows = + [ + new TableUpdateRows.PersistentTable(new PersistentTableRows + { + Inserts = MakeRowList(rowCount), + Deletes = new BsatnRowList(new RowSizeHint.FixedSize(sizeof(uint)), []), + }), + ], + }, + ], + }, + ], + }); + + private static void AssertTransactionUpdateRows(ServerMessage message, int expectedRowCount) + { + var rowsData = GetTransactionUpdateRowsData(message); + Assert.Equal(expectedRowCount * sizeof(uint), rowsData.Count); + } + + private static List GetTransactionUpdateRowsData(ServerMessage message) + { + var update = Assert.IsType(message); + var querySet = Assert.Single(update.TransactionUpdate_.QuerySets); + var table = Assert.Single(querySet.Tables); + var rows = Assert.Single(table.Rows); + var persistentRows = Assert.IsType(rows); + return persistentRows.PersistentTable_.Inserts.RowsData; + } + + private static byte[] EncodeServerMessage(ServerMessage message, bool brotli) + { + using var output = new MemoryStream(); + output.WriteByte(brotli ? (byte)1 : (byte)0); + if (brotli) + { + using (var compressed = new BrotliStream(output, CompressionMode.Compress, leaveOpen: true)) + using (var writer = new BinaryWriter(compressed)) + { + new ServerMessage.BSATN().Write(writer, message); + } + } + else + { + using var writer = new BinaryWriter(output); + new ServerMessage.BSATN().Write(writer, message); + } + return output.ToArray(); + } + + public sealed class PerfRow : IStructuralReadWrite, IEquatable + { + public uint Id; + + public void ReadFields(BinaryReader reader) + { + Id = new U32().Read(reader); + } + + public void WriteFields(BinaryWriter writer) + { + new U32().Write(writer, Id); + } + + public object GetSerializer() => new BSATN(); + + public bool Equals(PerfRow? other) => other != null && Id == other.Id; + + public override bool Equals(object? obj) => obj is PerfRow other && Equals(other); + + public override int GetHashCode() => Id.GetHashCode(); + + public readonly struct BSATN : IReadWrite + { + public PerfRow Read(BinaryReader reader) => new() { Id = new U32().Read(reader) }; + + public void Write(BinaryWriter writer, PerfRow value) => new U32().Write(writer, value.Id); + + public AlgebraicType GetAlgebraicType(ITypeRegistrar registrar) => throw new NotImplementedException(); + } + } + + public sealed class PerfRows : IStructuralReadWrite + { + public List Rows = new(); + + public void ReadFields(BinaryReader reader) + { + Rows = new SpacetimeDB.BSATN.List().Read(reader); + } + + public void WriteFields(BinaryWriter writer) + { + new SpacetimeDB.BSATN.List().Write(writer, Rows); + } + + public object GetSerializer() => new BSATN(); + + public readonly struct BSATN : IReadWrite + { + public PerfRows Read(BinaryReader reader) => + new() { Rows = new SpacetimeDB.BSATN.List().Read(reader) }; + + public void Write(BinaryWriter writer, PerfRows value) => + new SpacetimeDB.BSATN.List().Write(writer, value.Rows); + + public AlgebraicType GetAlgebraicType(ITypeRegistrar registrar) => throw new NotImplementedException(); + } + } +} diff --git a/sdks/csharp/tests~/Tests.cs b/sdks/csharp/tests~/Tests.cs index 3adb4970cba..619c3e9e8af 100644 --- a/sdks/csharp/tests~/Tests.cs +++ b/sdks/csharp/tests~/Tests.cs @@ -128,4 +128,33 @@ public static void ListstreamWorks() } }); } -} \ No newline at end of file + + [Fact] + public static void BSATNHelpersDecodeListMatchesByteArray() + { + var expectedRows = new PerformanceTests.PerfRows + { + Rows = + [ + new PerformanceTests.PerfRow { Id = 1 }, + new PerformanceTests.PerfRow { Id = 2 }, + new PerformanceTests.PerfRow { Id = 10_000 }, + ], + }; + var bytes = IStructuralReadWrite.ToBytes(new PerformanceTests.PerfRows.BSATN(), expectedRows); + var list = bytes.ToList(); + + var decodedFromBytes = BSATNHelpers.Decode(bytes); + var decodedFromList = BSATNHelpers.Decode(list); + var decodedFromReadOnlyList = BSATNHelpers.Decode((IReadOnlyList)list); + + Assert.Equal( + decodedFromBytes.Rows.Select(row => row.Id), + decodedFromList.Rows.Select(row => row.Id) + ); + Assert.Equal( + decodedFromBytes.Rows.Select(row => row.Id), + decodedFromReadOnlyList.Rows.Select(row => row.Id) + ); + } +}