SDK Performance Boost

sdks/csharp/src/BSATNHelpers.cs

@@ -1,36 +1,79 @@
 using SpacetimeDB.BSATN;
+using System.Buffers;
+using System.Collections.Generic;
 using System.IO;
 
 namespace SpacetimeDB
 {
     public static class BSATNHelpers
     {
         /// <summary>
-        /// 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 <c>IStructuralReadWrite.Read<T>(BinaryReader)</c> when
         /// deserializing many items from a buffer.
         /// </summary>
         /// <typeparam name="T"></typeparam>
         /// <param name="bsatn"></param>
         /// <returns></returns>
-        public static T Decode<T>(System.Collections.Generic.List<byte> bsatn) where T : IStructuralReadWrite, new() =>
-            Decode<T>(bsatn.ToArray());
+        public static T Decode<T>(byte[] bsatn) where T : IStructuralReadWrite, new()
+        {
+            using var stream = new MemoryStream(bsatn);
+            using var reader = new BinaryReader(stream);
+            return IStructuralReadWrite.Read<T>(reader);
+        }
 
         /// <summary>
-        /// 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 <c>IStructuralReadWrite.Read<T>(BinaryReader)</c> when
         /// deserializing many items from a buffer.
         /// </summary>
         /// <typeparam name="T"></typeparam>
         /// <param name="bsatn"></param>
         /// <returns></returns>
-        public static T Decode<T>(byte[] bsatn) where T : IStructuralReadWrite, new()
+        public static T Decode<T>(System.Collections.Generic.List<byte> 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<T>(reader);
+            }
+            finally
+            {
+                ArrayPool<byte>.Shared.Return(pooledBuffer);
+            }
+        }
+
+        /// <summary>
+        /// Decode an element of a BSATN-serializable type from a readonly byte list.
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="bsatn"></param>
+        /// <returns></returns>
+        public static T Decode<T>(IReadOnlyList<byte> bsatn) where T : IStructuralReadWrite, new()
+        {
+            if (bsatn is byte[] bytes)
+            {
+                return Decode<T>(bytes);
+            }
+
+            if (bsatn is System.Collections.Generic.List<byte> list)
+            {
+                return Decode<T>(list);
+            }
+
+            using var stream = new ListStream(bsatn);
             using var reader = new BinaryReader(stream);
             return IStructuralReadWrite.Read<T>(reader);
         }
+
+        public static MemoryStream MakePooledListStream(System.Collections.Generic.List<byte> bsatn, out byte[] pooledBuffer)
+        {
+            pooledBuffer = ArrayPool<byte>.Shared.Rent(bsatn.Count);
+            bsatn.CopyTo(pooledBuffer);
+            return new MemoryStream(pooledBuffer, 0, bsatn.Count, writable: false);
+        }
     }
 }

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));
+            }
         }
 
         /// <summary>

sdks/csharp/src/ListStream.cs

@@ -10,10 +10,10 @@
 /// </summary>
 internal class ListStream : Stream
 {
-    private List<byte> list;
+    private IReadOnlyList<byte> list;
     private int pos;
 
-    public ListStream(List<byte> data)
+    public ListStream(IReadOnlyList<byte> 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<byte> 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<byte> buffer)

sdks/csharp/src/MultiDictionary.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Linq;
 using System.Text;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -20,7 +19,7 @@
     internal struct MultiDictionary<TKey, TValue> : IEquatable<MultiDictionary<TKey, TValue>>
     {
         // The actual data.
         readonly Dictionary<TKey, (TValue Value, uint Multiplicity)> RawDict;
         readonly IEqualityComparer<TValue> ValueComparer;
 
         /// <summary>
@@ -55,7 +54,18 @@
         /// <summary>
         /// Return the count WITH multiplicities.
         /// </summary>
-        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;
+            }
+        }
 
         /// <summary>
         /// Add a key-value-pair to the multidictionary.
@@ -113,7 +123,8 @@
         /// </summary>
         /// <param name="key"></param>
         /// <returns></returns>
-        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;
 
         /// <summary>
         /// The value associated with a key.
@@ -171,16 +182,21 @@
         {
             get
             {
-
-                return RawDict.Select(item => item.Value.Value);
+                foreach (var item in RawDict)
+                {
+                    yield return item.Value.Value;
+                }
             }
         }
 
         public readonly IEnumerable<KeyValuePair<TKey, TValue>> Entries
         {
             get
             {
-                return RawDict.Select(item => new KeyValuePair<TKey, TValue>(item.Key, item.Value.Value));
+                foreach (var item in RawDict)
+                {
+                    yield return new KeyValuePair<TKey, TValue>(item.Key, item.Value.Value);
+                }
             }
         }
 
@@ -191,31 +207,32 @@
         /// <returns></returns>
         public readonly IEnumerable<KeyValuePair<TKey, TValue>> WillRemove(MultiDictionaryDelta<TKey, TValue> 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<TKey, TValue>(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<TKey, TValue>(entry.Key, entry.Value.NonValueChange.Value));
+            }
         }
 
         /// <summary>
@@ -225,7 +242,7 @@
         /// <param name="wasInserted">Will be populated with inserted KVPs.</param>
         /// <param name="wasUpdated">Will be populated with updated KVPs.</param>
         /// <param name="wasRemoved">Will be populated with removed KVPs.</param>
-        public void Apply(MultiDictionaryDelta<TKey, TValue> delta, List<KeyValuePair<TKey, TValue>> wasInserted, List<(TKey Key, TValue OldValue, TValue NewValue)> wasUpdated, List<KeyValuePair<TKey, TValue>> wasRemoved)
+        public readonly void Apply(MultiDictionaryDelta<TKey, TValue> delta, List<KeyValuePair<TKey, TValue>> wasInserted, List<(TKey Key, TValue OldValue, TValue NewValue)> wasUpdated, List<KeyValuePair<TKey, TValue>> wasRemoved)
         {
             foreach (var (key, their) in delta.Entries)
             {
@@ -240,7 +257,7 @@
                         var reducedMultiplicity = (int)my.Multiplicity + before.Delta;
                         if (reducedMultiplicity != 0)
                         {
                             PseudoThrow($"Attempted to apply {their} to {my}, but this resulted in a multiplicity of {reducedMultiplicity}, failing to correctly remove the row before applying the update");
                         }
                         RawDict[key] = (after.Value, (uint)after.Delta);
 
@@ -269,7 +286,7 @@
                             // This is a removal.
                             if (newMultiplicity < 0)
                             {
                                 PseudoThrow($"Internal error: Removing row with key {key} {-theirDelta.Delta} times, but it is only present {my.Multiplicity} times.");
                             }
                             RawDict.Remove(key);
                             wasRemoved.Add(new(key, theirDelta.Value));
@@ -281,7 +298,7 @@
                     // Key is not present in map.
                     if (their.IsValueChange)
                     {
                         PseudoThrow($"Internal error: Can't perform a value change on a nonexistent key {key} (change: {their}).");
                     }
                     else
                     {
@@ -295,7 +312,7 @@
                         }
                         else if (theirDelta.Delta < 0)
                         {
                             PseudoThrow($"Internal error: Can't remove nonexistent key {theirDelta.Value}");
                         }
                         else
                         {
@@ -612,7 +629,7 @@
         /// For each key, track its value (or its old and new values).
         /// Also track the deltas associated to the values for this key.
         /// </summary>
         readonly Dictionary<TKey, KeyDelta> RawDict;
 
         readonly IEqualityComparer<TValue> ValueComparer;
 
@@ -723,4 +740,4 @@
             }
         }
     }
-}
+}

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<T>.Success(BSATNHelpers.Decode<T>(bytes.ToArray())),
+                    ProcedureCallbackResult<T>.Success(BSATNHelpers.Decode<T>(bytes)),
                 ProcedureStatus.InternalError(var error) =>
                     ProcedureCallbackResult<T>.Failure(new Exception($"Procedure failed: {error}")),
                 _ => ProcedureCallbackResult<T>.Failure(new Exception("Unknown procedure status"))

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<byte> bytes)
+            string DecodeReducerError(List<byte> 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<byte>.Shared.Return(pooledBuffer);
+                    }
                 }
                 catch
                 {

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<EventContext, Row> table)
             }
 
             public IEnumerable<Row> Filter(Column value) =>
-                cache.TryGetValue(value, out var rows) ? rows : Enumerable.Empty<Row>();
+                cache.TryGetValue(value, out var rows) ? rows : Array.Empty<Row>();
         }
 
         /// <summary>
@@ -415,7 +414,7 @@ public event RowEventHandler OnInsert
 
         public int Count => (int)Entries.CountDistinct;
 
-        public IEnumerable<Row> Iter() => Entries.Entries.Select(entry => (Row)entry.Value);
+        public IEnumerable<Row> Iter() => Entries.Values;
 
         public Task<Row[]> RemoteQuery(string query) =>
             conn.RemoteQuery<Row>($"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);
             }
         }