diff --git a/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md b/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md index e6de19c91d7..7ce7403a429 100644 --- a/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md +++ b/docs/docs/00200-core-concepts/00600-clients/00600-csharp-reference.md @@ -42,6 +42,7 @@ Before diving into the reference, you may want to review: | [`ErrorContext` type](#type-errorcontext) | Implements [`IDbContext`](#interface-idbcontext) for subscription error callbacks. | | [Query Builder API](#query-builder-api) | Type-safe query builder for typed subscription queries. | | [Access the client cache](#access-the-client-cache) | Access to your local view of the database. | +| [Configure event dispatch](#configure-event-dispatch) | Choose native C# events or a custom event listener backend. | | [Observe and invoke reducers](#observe-and-invoke-reducers) | Send requests to the database to run reducers, and register callbacks to run when notified of reducers. | | [Identify a client](#identify-a-client) | Types for identifying users and client connections. | @@ -1035,6 +1036,55 @@ The `OnUpdate` callback runs whenever an already-resident row in the client cach See [the quickstart](../../00100-intro/00200-quickstarts/00600-c-sharp.md) for examples of registering and unregistering row callbacks. +### Configure event dispatch + +By default, the C# SDK stores table row callbacks as regular native C# events. For most applications, no extra setup is required: + +```csharp +conn.Db.User.OnInsert += OnUserInsert; +conn.Db.User.OnInsert -= OnUserInsert; +``` + +Native events are simple, idiomatic, and should be your default choice unless profiling shows that event subscription management is a problem in your application. + +If your client frequently adds and removes many row callbacks, the cost of native multicast delegate updates can become noticeable. For those cases, the SDK can use a custom event listener backend instead: + +```csharp +using SpacetimeDB; + +SpacetimeDB.EventHandling.Backend.UseCustomListeners(); + +var conn = DbConnection.Builder() + .WithUri("http://localhost:3000") + .WithDatabaseName("my-database") + .Build(); +``` + +Call `Backend.UseCustomListeners()` before creating the generated `DbConnection`. Table handles capture the selected backend when they are constructed, so changing the backend later does not update existing handles. + +The default custom backend keeps listeners in an indexed collection. It is useful when you have many listener removals, duplicate subscriptions, or integration code that attaches and detaches callbacks aggressively. Registering and unregistering callbacks still uses the same generated `OnInsert`, `OnDelete`, and `OnUpdate` event APIs. + +Reducer result events, such as `conn.Reducers.OnSendMessage`, are always regular C# events. + +If your project includes [Sappy](https://github.com/clockworklabs/SappyEvents/), the SDK can use Sappy-backed listener storage: + +```csharp +using SpacetimeDB.SappyIntegration; + +SpacetimeDB.EventHandling.Backend.UseCustomListeners(new SappyEventListenersFactory()); +``` + +Use the Sappy backend only in projects that already reference Sappy. It is intended for applications that have standardized on Sappy's event/listener model; it is not required for normal C# or Unity clients. + +For Sappy-backed table callbacks, register and unregister generated Sappy targets through the listener accessors instead of using normal C# event syntax: + +```csharp +conn.Db.User.OnInsertListeners.AddSapTarget(Sappy.OnUserInsert); +conn.Db.User.OnInsertListeners.RemoveSapTarget(Sappy.OnUserInsert); +``` + +Use the matching listener accessor for each row callback: `OnInsertListeners`, `OnDeleteListeners`, and `OnUpdateListeners`. This lets Sappy manage the callback target directly, which is required for the Sappy backend to behave correctly and avoid unnecessary delegate-management overhead. + ### Unique constraint index access For each unique constraint on a table, its table handle has a property which is a unique index handle and whose name is the unique column name. This unique index handle has a method `.Find(Column value)`. If a `Row` with `value` in the unique column is resident in the client cache, `.Find` returns it. Otherwise it returns null. diff --git a/sdks/csharp/src/AssemblyInfo.cs b/sdks/csharp/src/AssemblyInfo.cs new file mode 100644 index 00000000000..bc7cfcf281c --- /dev/null +++ b/sdks/csharp/src/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("com.clockworklabs.spacetimedbsdk.sappyintegration")] diff --git a/sdks/csharp/src/AssemblyInfo.cs.meta b/sdks/csharp/src/AssemblyInfo.cs.meta new file mode 100644 index 00000000000..c29b17f70a4 --- /dev/null +++ b/sdks/csharp/src/AssemblyInfo.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 5a80f81d7b4545c7965a34c942583f06 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/EventHandling/AbstractEventHandler.cs b/sdks/csharp/src/EventHandling/AbstractEventHandler.cs deleted file mode 100644 index 1d6e03b4ada..00000000000 --- a/sdks/csharp/src/EventHandling/AbstractEventHandler.cs +++ /dev/null @@ -1,100 +0,0 @@ -using System; - -namespace SpacetimeDB.EventHandling -{ - internal class AbstractEventHandler - { - private EventListeners Listeners { get; } = new(); - - public void Invoke() - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } - internal class AbstractEventHandler - { - private EventListeners> Listeners { get; } = new(); - - public void Invoke(T value) - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(value); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } - - internal class AbstractEventHandler - { - private EventListeners> Listeners { get; } = new(); - - public void Invoke(T1 v1, T2 v2) - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(v1, v2); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } - - internal class AbstractEventHandler - { - private EventListeners> Listeners { get; } = new(); - - public void Invoke(T1 v1, T2 v2, T3 v3) - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(v1, v2, v3); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } - - internal class AbstractEventHandler - { - private EventListeners> Listeners { get; } = new(); - - public void Invoke(T1 v1, T2 v2, T3 v3, T4 v4) - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(v1, v2, v3, v4); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } - - internal class AbstractEventHandler - { - private EventListeners> Listeners { get; } = new(); - - public void Invoke(T1 v1, T2 v2, T3 v3, T4 v4, T5 v5) - { - for (var i = Listeners.Count - 1; i >= 0; i--) - { - Listeners[i]?.Invoke(v1, v2, v3, v4, v5); - } - } - - public void AddListener(Action listener) => Listeners.Add(listener); - public void RemoveListener(Action listener) => Listeners.Remove(listener); - } -} \ No newline at end of file diff --git a/sdks/csharp/src/EventHandling/Backend.cs b/sdks/csharp/src/EventHandling/Backend.cs new file mode 100644 index 00000000000..ec540e83c91 --- /dev/null +++ b/sdks/csharp/src/EventHandling/Backend.cs @@ -0,0 +1,24 @@ +using System; + +namespace SpacetimeDB.EventHandling +{ + public static class Backend + { + internal static bool UseNativeDispatch { get; private set; } = true; + private static IEventListenersFactory? CustomFactory { get; set; } + + public static void UseNativeEvents() + { + UseNativeDispatch = true; + CustomFactory = null; + } + + public static void UseCustomListeners(IEventListenersFactory? factory = null) + { + UseNativeDispatch = false; + CustomFactory = factory; + } + + internal static IEventListeners Create() where T : Delegate => CustomFactory?.Create() ?? new EventListeners(); + } +} \ No newline at end of file diff --git a/sdks/csharp/src/EventHandling/AbstractEventHandler.cs.meta b/sdks/csharp/src/EventHandling/Backend.cs.meta similarity index 83% rename from sdks/csharp/src/EventHandling/AbstractEventHandler.cs.meta rename to sdks/csharp/src/EventHandling/Backend.cs.meta index 2ceef79f6e5..e76cb3b3981 100644 --- a/sdks/csharp/src/EventHandling/AbstractEventHandler.cs.meta +++ b/sdks/csharp/src/EventHandling/Backend.cs.meta @@ -1,5 +1,5 @@ fileFormatVersion: 2 -guid: a3ff844e9ff394788a1bc7e8e83ac86b +guid: e6840d90a7134fdd92769b5e5d3f24b4 MonoImporter: externalObjects: {} serializedVersion: 2 diff --git a/sdks/csharp/src/EventHandling/EventListeners.cs b/sdks/csharp/src/EventHandling/EventListeners.cs index d4acc4b7218..817044aafd8 100644 --- a/sdks/csharp/src/EventHandling/EventListeners.cs +++ b/sdks/csharp/src/EventHandling/EventListeners.cs @@ -1,40 +1,308 @@ -using System; +using System; using System.Collections.Generic; +using System.Runtime.CompilerServices; namespace SpacetimeDB.EventHandling { - internal class EventListeners where T : Delegate + internal sealed class EventListeners : IEventListeners where T : Delegate { - private List List { get; } - private Dictionary Indices { get; } + private const int SmallListenerThreshold = 8; + private const int CollisionBucket = -1; - public int Count => List.Count; + private static readonly EqualityComparer Comparer = EqualityComparer.Default; - public T this[int index] => List[index]; + private int[] _hashes; + private T?[] _listeners; + private int _count; + private Dictionary? _indices; + private Dictionary>? _collisions; + private Stack>? _collisionsPool; + + public int Count + { + [MethodImpl(MethodImplOptions.AggressiveInlining)] + get => _count; + } + + public T this[int index] + { + [MethodImpl(MethodImplOptions.AggressiveInlining)] + get => _listeners[index]!; + } + + public EventListeners() : this(4) { } - public EventListeners() : this(0) { } public EventListeners(int initialSize) { - List = new List(initialSize); - Indices = new Dictionary(initialSize); + var capacity = Math.Max(1, initialSize); + _hashes = new int[capacity]; + _listeners = new T[capacity]; } public void Add(T listener) { - if (listener == null || !Indices.TryAdd(listener, List.Count)) return; - List.Add(listener); + if (listener == null) return; + + var hashCode = listener.GetHashCode(); + + if (_count <= SmallListenerThreshold) + { + AddRaw(hashCode, listener); + + if (_count > SmallListenerThreshold) + { + RebuildIndex(); + } + + return; + } + + var newIndex = AddRaw(hashCode, listener); + var indices = _indices!; + + if (!indices.TryGetValue(hashCode, out var index)) + { + indices.Add(hashCode, newIndex); + return; + } + + if (index != CollisionBucket) + { + _collisions ??= new Dictionary>(); + _collisions[hashCode] = GetCollisionsListFromPool(index, newIndex); + indices[hashCode] = CollisionBucket; + return; + } + + var bucket = _collisions![hashCode]; + bucket.Add(newIndex); } public void Remove(T listener) { - if (listener == null || List.Count <= 0 || !Indices.Remove(listener, out var index)) return; - var lastListener = List[^1]; - if (lastListener != listener) + if (listener == null || _count <= 0) return; + + if (_count <= SmallListenerThreshold) + { + var index = FindLinear(listener); + if (index >= 0) + { + RemoveAtSwapBackRaw(index); + } + + return; + } + + var hashCode = listener.GetHashCode(); + var indices = _indices; + if (indices == null) return; + + if (!indices.TryGetValue(hashCode, out var mappedIndex)) return; + + var removeIndex = -1; + + if (mappedIndex != CollisionBucket) + { + if (!DelegateEquals(_listeners[mappedIndex]!, listener)) return; + + removeIndex = mappedIndex; + indices.Remove(hashCode); + } + else + { + var bucket = _collisions![hashCode]; + + for (var i = 0; i < bucket.Count; i++) + { + var candidate = bucket[i]; + if (!DelegateEquals(_listeners[candidate]!, listener)) continue; + + removeIndex = candidate; + RemoveBucketSlot(bucket, i); + + if (bucket.Count == 1) + { + indices[hashCode] = bucket[0]; + _collisions.Remove(hashCode); + ReturnCollisionsListToPool(bucket); + } + else if (bucket.Count == 0) + { + indices.Remove(hashCode); + _collisions.Remove(hashCode); + ReturnCollisionsListToPool(bucket); + } + + break; + } + + if (removeIndex < 0) return; + } + + var movedFrom = _count - 1; + var movedHashCode = _hashes[movedFrom]; + + RemoveAtSwapBackRaw(removeIndex); + + if (_count <= SmallListenerThreshold) + { + ClearIndex(); + } + else if (removeIndex != movedFrom) + { + UpdateMovedIndex(movedHashCode, movedFrom, removeIndex); + } + } + + private int FindLinear(T listener) + { + for (var i = 0; i < _count; i++) + { + if (DelegateEquals(_listeners[i]!, listener)) return i; + } + + return -1; + } + + private int AddRaw(int hashCode, T listener) + { + EnsureCapacity(); + + var index = _count; + _hashes[index] = hashCode; + _listeners[index] = listener; + _count++; + return index; + } + + private void RemoveAtSwapBackRaw(int index) + { + var lastIndex = _count - 1; + + if (index != lastIndex) + { + _hashes[index] = _hashes[lastIndex]; + _listeners[index] = _listeners[lastIndex]; + } + + _hashes[lastIndex] = 0; + _listeners[lastIndex] = null; + _count = lastIndex; + } + + private void EnsureCapacity() + { + var capacity = _listeners.Length; + if (_count < capacity) return; + + capacity *= 2; + Array.Resize(ref _hashes, capacity); + Array.Resize(ref _listeners, capacity); + } + + private void UpdateMovedIndex(int hashCode, int oldIndex, int newIndex) + { + var indices = _indices!; + var mappedIndex = indices[hashCode]; + + if (mappedIndex != CollisionBucket) + { + indices[hashCode] = newIndex; + return; + } + + var bucket = _collisions![hashCode]; + + for (var i = 0; i < bucket.Count; i++) + { + if (bucket[i] == oldIndex) + { + bucket[i] = newIndex; + return; + } + } + } + + private static void RemoveBucketSlot(List bucket, int slot) + { + var lastSlot = bucket.Count - 1; + + if (slot != lastSlot) { - Indices[lastListener] = index; + bucket[slot] = bucket[lastSlot]; } - List.RemoveAtSwapBack(index); + bucket.RemoveAt(lastSlot); } + + private void RebuildIndex() + { + if (_indices == null) + { + _indices = new Dictionary(_listeners.Length); + } + else + { + ClearIndex(); + } + + var indices = _indices; + for (var i = 0; i < _count; i++) + { + var hashCode = _hashes[i]; + + if (!indices.TryGetValue(hashCode, out var existing)) + { + indices.Add(hashCode, i); + continue; + } + + _collisions ??= new Dictionary>(); + + if (existing != CollisionBucket) + { + _collisions[hashCode] = GetCollisionsListFromPool(existing, i); + indices[hashCode] = CollisionBucket; + } + else + { + _collisions[hashCode].Add(i); + } + } + } + + private void ClearIndex() + { + _indices?.Clear(); + + if (_collisions == null) return; + + foreach (var collisions in _collisions.Values) + { + ReturnCollisionsListToPool(collisions); + } + + _collisions.Clear(); + } + + private List GetCollisionsListFromPool(int a, int b) + { + if (_collisionsPool == null || _collisionsPool.Count <= 0) return new List(2) { a, b }; + + var list = _collisionsPool.Pop(); + list.Add(a); + list.Add(b); + return list; + } + + private void ReturnCollisionsListToPool(List list) + { + list.Clear(); + _collisionsPool ??= new Stack>(); + _collisionsPool.Push(list); + } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + private static bool DelegateEquals(T a, T b) => Comparer.Equals(a, b); } -} \ No newline at end of file +} diff --git a/sdks/csharp/src/EventHandling/IEventListeners.cs b/sdks/csharp/src/EventHandling/IEventListeners.cs new file mode 100644 index 00000000000..fad7c71e9e9 --- /dev/null +++ b/sdks/csharp/src/EventHandling/IEventListeners.cs @@ -0,0 +1,18 @@ +using System; + +namespace SpacetimeDB.EventHandling +{ + public interface IEventListeners where T : Delegate + { + int Count { get; } + T this[int index] { get; } + + void Add(T listener); + void Remove(T listener); + } + + public interface IEventListenersFactory + { + IEventListeners Create() where T : Delegate; + } +} diff --git a/sdks/csharp/src/EventHandling/IEventListeners.cs.meta b/sdks/csharp/src/EventHandling/IEventListeners.cs.meta new file mode 100644 index 00000000000..f4adfa93fac --- /dev/null +++ b/sdks/csharp/src/EventHandling/IEventListeners.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: b75e0a997a0c4637854774242df43635 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SappyIntegration.meta b/sdks/csharp/src/SappyIntegration.meta new file mode 100644 index 00000000000..c173c92b79c --- /dev/null +++ b/sdks/csharp/src/SappyIntegration.meta @@ -0,0 +1,8 @@ +fileFormatVersion: 2 +guid: 7c4698fe2727425488205582952273db +folderAsset: yes +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SappyIntegration/Extensions.cs b/sdks/csharp/src/SappyIntegration/Extensions.cs new file mode 100644 index 00000000000..1ae7f5438eb --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/Extensions.cs @@ -0,0 +1,35 @@ +#if SAPPY +using System; +using SpacetimeDB.EventHandling; +using Sappy; + +namespace SpacetimeDB.SappyIntegration +{ + public static class Extensions + { + public static void AddSapTarget(this IEventListeners listeners, SapTarget value) where T : Delegate + { + if (listeners is SappyEventListeners sappyEventListeners) + { + sappyEventListeners.Add(value); + } + else + { + listeners.Add(value.Callback); + } + } + + public static void RemoveSapTarget(this IEventListeners listeners, SapTarget value) where T : Delegate + { + if (listeners is SappyEventListeners sappyEventListeners) + { + sappyEventListeners.Remove(value); + } + else + { + listeners.Remove(value.Callback); + } + } + } +} +#endif \ No newline at end of file diff --git a/sdks/csharp/src/SappyIntegration/Extensions.cs.meta b/sdks/csharp/src/SappyIntegration/Extensions.cs.meta new file mode 100644 index 00000000000..859a0f2071b --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/Extensions.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: e9e0bd96416e4b9eb254e5303f468e9f +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs b/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs new file mode 100644 index 00000000000..7c165f80f97 --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs @@ -0,0 +1,69 @@ +#if SAPPY +using System; +using Sappy; +using SpacetimeDB.EventHandling; +using System.Runtime.CompilerServices; + +namespace SpacetimeDB.SappyIntegration +{ + public class SappyEventListeners : IEventListeners where T : Delegate + { + private EventListeners? _eventListeners; + private SapDelegate? _sapDelegate; + + public int Count + { + [MethodImpl(MethodImplOptions.AggressiveInlining)] + get => (_eventListeners?.Count ?? 0) + (_sapDelegate?.Count ?? 0); + } + + public T this[int index] + { + get + { + var eventListeners = _eventListeners; + var eventListenersCount = eventListeners?.Count ?? 0; + + if ((uint)index < (uint)eventListenersCount) + { + return eventListeners![index]; + } + + var sapDelegate = _sapDelegate; + var sapIndex = index - eventListenersCount; + + if (sapDelegate != null && (uint)sapIndex < (uint)sapDelegate.Count) + { + return sapDelegate[sapIndex]; + } + + throw new IndexOutOfRangeException(); + } + } + + public void Add(SapTarget listener) + { + if (listener == null) return; + (_sapDelegate ??= new SapDelegate()).Add(listener); + } + + public void Remove(SapTarget listener) + { + if (listener == null || _sapDelegate == null) return; + _sapDelegate.Remove(listener); + } + + public void Add(T listener) + { + if (listener == null) return; + (_eventListeners ??= new EventListeners()).Add(listener); + } + + public void Remove(T listener) + { + if (listener == null || _eventListeners == null) return; + _eventListeners.Remove(listener); + } + } +} +#endif diff --git a/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs.meta b/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs.meta new file mode 100644 index 00000000000..1c7ed528b41 --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/SappyEventListeners.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 670852799878403c999226dbdb5c6b39 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs b/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs new file mode 100644 index 00000000000..4c5863d9307 --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs @@ -0,0 +1,13 @@ +#if SAPPY +using System; +using UnityEngine; +using SpacetimeDB.EventHandling; + +namespace SpacetimeDB.SappyIntegration +{ + public class SappyEventListenersFactory : IEventListenersFactory + { + public IEventListeners Create() where T : Delegate => new SappyEventListeners(); + } +} +#endif \ No newline at end of file diff --git a/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs.meta b/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs.meta new file mode 100644 index 00000000000..3f1a5297aa5 --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/SappyEventListenersFactory.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 8a2f65d5ec304b5786ec99e70a680bb8 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef b/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef new file mode 100644 index 00000000000..e72d9382a6e --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef @@ -0,0 +1,18 @@ +{ + "name": "com.clockworklabs.spacetimedbsdk.sappyintegration", + "rootNamespace": "SpacetimeDB", + "references": [ + "com.clockworklabs.spacetimedbsdk", + "ClockworkLabs.Sappy" + ], + "defineConstraints": [ + "SAPPY" + ], + "versionDefines": [ + { + "name": "io.clockworklabs.sappy", + "expression": "1.1.0", + "define": "SAPPY" + } + ] +} diff --git a/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef.meta b/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef.meta new file mode 100644 index 00000000000..1aca66b0b8a --- /dev/null +++ b/sdks/csharp/src/SappyIntegration/com.clockworklabs.spacetimedbsdk.sappyintegration.asmdef.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: 872005274f4d4868889ff510e2bf73fc +AssemblyDefinitionImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/sdks/csharp/src/Table.cs b/sdks/csharp/src/Table.cs index 063d45bfdbe..3df8c6b64cc 100644 --- a/sdks/csharp/src/Table.cs +++ b/sdks/csharp/src/Table.cs @@ -212,18 +212,8 @@ static RemoteTableHandleBase() // I didn't do that because that delays the index updates until after the row is processed. // In theory, that shouldn't be the issue, but I didn't want to break it right before leaving :) // - Ingvar - private AbstractEventHandler OnInternalInsertHandler { get; } = new(); - private event Action OnInternalInsert - { - add => OnInternalInsertHandler.AddListener(value); - remove => OnInternalInsertHandler.RemoveListener(value); - } - private AbstractEventHandler OnInternalDeleteHandler { get; } = new(); - private event Action OnInternalDelete - { - add => OnInternalDeleteHandler.AddListener(value); - remove => OnInternalDeleteHandler.RemoveListener(value); - } + private event Action OnInternalInsert; + private event Action OnInternalDelete; // These are implementations of the type-erased interface. object? IRemoteTableHandle.GetPrimaryKey(IStructuralReadWrite row) => GetPrimaryKey((Row)row); @@ -405,13 +395,17 @@ void IRemoteTableHandle.Parse(TableUpdate update, ParsedDatabaseUpdate dbOps) } public delegate void RowEventHandler(EventContext context, Row row); + public delegate void UpdateEventHandler(EventContext context, Row oldRow, Row newRow); + private CustomRowEventHandler OnInsertHandler { get; } = new(); public event RowEventHandler OnInsert { - add => OnInsertHandler.AddListener(value); - remove => OnInsertHandler.RemoveListener(value); + add => OnInsertHandler.Add(value); + remove => OnInsertHandler.Remove(value); } - public delegate void UpdateEventHandler(EventContext context, Row oldRow, Row newRow); +#if SAPPY + public IEventListeners OnInsertListeners => OnInsertHandler.Listeners; +#endif public int Count => (int)Entries.CountDistinct; @@ -506,14 +500,14 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa { if (value is Row oldRow) { - OnInternalDeleteHandler.Invoke(oldRow); + OnInternalDelete?.Invoke(oldRow); } } foreach (var (_, value) in wasInserted) { if (value is Row newRow) { - OnInternalInsertHandler.Invoke(newRow); + OnInternalInsert?.Invoke(newRow); } else { @@ -524,7 +518,7 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa { if (oldValue is Row oldRow) { - OnInternalDeleteHandler.Invoke(oldRow); + OnInternalDelete?.Invoke(oldRow); } else { @@ -534,7 +528,7 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa if (newValue is Row newRow) { - OnInternalInsertHandler.Invoke(newRow); + OnInternalInsert?.Invoke(newRow); } else { @@ -575,33 +569,115 @@ void IRemoteTableHandle.PostApply(IEventContext context) protected class CustomRowEventHandler { - private EventListeners Listeners { get; } = new(); + private readonly bool _useNativeDispatch = Backend.UseNativeDispatch; + private RowEventHandler? _nativeListeners; + private readonly IEventListeners? _indexedListeners; - public void Invoke(EventContext ctx, Row row) + public IEventListeners Listeners => _indexedListeners ?? throw new InvalidOperationException( + "This event is using native C# event dispatch and does not expose indexed listeners. " + + "Use SpacetimeDB.EventHandling.Backend.UseCustomListeners() before creating table handles." + ); + + public CustomRowEventHandler() { - for (var i = Listeners.Count - 1; i >= 0; i--) + if (!_useNativeDispatch) { - Listeners[i]?.Invoke(ctx, row); + _indexedListeners = Backend.Create(); } } - public void AddListener(RowEventHandler listener) => Listeners.Add(listener); - public void RemoveListener(RowEventHandler listener) => Listeners.Remove(listener); + public void Add(RowEventHandler listener) + { + if (_useNativeDispatch) + { + _nativeListeners += listener; + return; + } + + _indexedListeners!.Add(listener); + } + + public void Remove(RowEventHandler listener) + { + if (_useNativeDispatch) + { + _nativeListeners -= listener; + return; + } + + _indexedListeners!.Remove(listener); + } + + public void Invoke(EventContext ctx, Row row) + { + if (_useNativeDispatch) + { + _nativeListeners?.Invoke(ctx, row); + return; + } + + var listeners = _indexedListeners!; + for (var i = listeners.Count - 1; i >= 0; i--) + { + listeners[i].Invoke(ctx, row); + } + } } protected class CustomUpdateEventHandler { - private EventListeners Listeners { get; } = new(); + private readonly bool _useNativeDispatch = Backend.UseNativeDispatch; + private UpdateEventHandler? _nativeListeners; + private readonly IEventListeners? _indexedListeners; - public void Invoke(EventContext ctx, Row oldRow, Row newRow) + public IEventListeners Listeners => _indexedListeners ?? throw new InvalidOperationException( + "This event is using native C# event dispatch and does not expose indexed listeners. " + + "Use SpacetimeDB.EventHandling.Backend.UseCustomListeners() before creating table handles." + ); + + public CustomUpdateEventHandler() + { + if (!_useNativeDispatch) + { + _indexedListeners = Backend.Create(); + } + } + + public void Add(UpdateEventHandler listener) + { + if (_useNativeDispatch) + { + _nativeListeners += listener; + return; + } + + _indexedListeners!.Add(listener); + } + + public void Remove(UpdateEventHandler listener) { - for (var i = Listeners.Count - 1; i >= 0; i--) + if (_useNativeDispatch) { - Listeners[i]?.Invoke(ctx, oldRow, newRow); + _nativeListeners -= listener; + return; } + + _indexedListeners!.Remove(listener); } - public void AddListener(UpdateEventHandler listener) => Listeners.Add(listener); - public void RemoveListener(UpdateEventHandler listener) => Listeners.Remove(listener); + public void Invoke(EventContext ctx, Row oldRow, Row newRow) + { + if (_useNativeDispatch) + { + _nativeListeners?.Invoke(ctx, oldRow, newRow); + return; + } + + var listeners = _indexedListeners!; + for (var i = listeners.Count - 1; i >= 0; i--) + { + listeners[i].Invoke(ctx, oldRow, newRow); + } + } } } @@ -617,21 +693,32 @@ protected RemoteTableHandle(IDbConnection conn) : base(conn) { } private CustomRowEventHandler OnDeleteHandler { get; } = new(); public event RowEventHandler OnDelete { - add => OnDeleteHandler.AddListener(value); - remove => OnDeleteHandler.RemoveListener(value); + add => OnDeleteHandler.Add(value); + remove => OnDeleteHandler.Remove(value); } +#if SAPPY + public IEventListeners OnDeleteListeners => OnDeleteHandler.Listeners; +#endif + private CustomRowEventHandler OnBeforeDeleteHandler { get; } = new(); public event RowEventHandler OnBeforeDelete { - add => OnBeforeDeleteHandler.AddListener(value); - remove => OnBeforeDeleteHandler.RemoveListener(value); + add => OnBeforeDeleteHandler.Add(value); + remove => OnBeforeDeleteHandler.Remove(value); } +#if SAPPY + public IEventListeners OnBeforeDeleteListeners => OnBeforeDeleteHandler.Listeners; +#endif + private CustomUpdateEventHandler OnUpdateHandler { get; } = new(); public event UpdateEventHandler OnUpdate { - add => OnUpdateHandler.AddListener(value); - remove => OnUpdateHandler.RemoveListener(value); + add => OnUpdateHandler.Add(value); + remove => OnUpdateHandler.Remove(value); } +#if SAPPY + public IEventListeners OnUpdateListeners => OnUpdateHandler.Listeners; +#endif protected override void InvokeDelete(IEventContext context, IStructuralReadWrite row) { diff --git a/sdks/csharp/src/WebSocket.cs b/sdks/csharp/src/WebSocket.cs index 8ae335db8f0..c3e27c9f812 100644 --- a/sdks/csharp/src/WebSocket.cs +++ b/sdks/csharp/src/WebSocket.cs @@ -48,6 +48,7 @@ public WebSocket(ConnectOptions options) #endif } + // TODO: This never has subscriptions public event OpenEventHandler? OnConnect; public event ConnectErrorEventHandler? OnConnectError; public event SendErrorEventHandler? OnSendError; diff --git a/sdks/csharp/src/com.clockworklabs.spacetimedbsdk.asmdef b/sdks/csharp/src/com.clockworklabs.spacetimedbsdk.asmdef index ab9733e7bcb..73774dfbaa7 100644 --- a/sdks/csharp/src/com.clockworklabs.spacetimedbsdk.asmdef +++ b/sdks/csharp/src/com.clockworklabs.spacetimedbsdk.asmdef @@ -1,3 +1,11 @@ { - "name": "com.clockworklabs.spacetimedbsdk" + "name": "com.clockworklabs.spacetimedbsdk", + "rootNamespace": "SpacetimeDB", + "versionDefines": [ + { + "name": "io.clockworklabs.sappy", + "expression": "1.1.0", + "define": "SAPPY" + } + ] } diff --git a/sdks/csharp/tests~/EventListenersTests.cs b/sdks/csharp/tests~/EventListenersTests.cs new file mode 100644 index 00000000000..d867cd91872 --- /dev/null +++ b/sdks/csharp/tests~/EventListenersTests.cs @@ -0,0 +1,64 @@ +using System; +using SpacetimeDB.EventHandling; +using Xunit; + +public class EventListenersTests +{ + [Fact] + public void EventListenersAllowDuplicatesAndRemoveOneSubscriptionAtATime() + { + var eventListeners = new EventListeners(); + var callCount = 0; + var listeners = new Action[12]; + + for (var i = 0; i < listeners.Length; i++) + { + listeners[i] = new Listener(() => callCount++).Invoke; + eventListeners.Add(listeners[i]); + } + + eventListeners.Add(listeners[3]); + Assert.Equal(listeners.Length + 1, eventListeners.Count); + + InvokeAll(eventListeners); + Assert.Equal(listeners.Length + 1, callCount); + + eventListeners.Remove(listeners[3]); + eventListeners.Remove(listeners[9]); + Assert.Equal(listeners.Length - 1, eventListeners.Count); + + callCount = 0; + InvokeAll(eventListeners); + Assert.Equal(listeners.Length - 1, callCount); + + eventListeners.Remove(listeners[3]); + Assert.Equal(listeners.Length - 2, eventListeners.Count); + + eventListeners.Add(listeners[3]); + eventListeners.Add(listeners[9]); + Assert.Equal(listeners.Length, eventListeners.Count); + } + + private static void InvokeAll(EventListeners listeners) + { + for (var i = listeners.Count - 1; i >= 0; i--) + { + listeners[i](); + } + } + + private sealed class Listener + { + private readonly Action Callback; + + public Listener(Action callback) + { + Callback = callback; + } + + public void Invoke() + { + Callback(); + } + } +} diff --git a/sdks/csharp/tests~/event-handling-benchmarks/README.md b/sdks/csharp/tests~/event-handling-benchmarks/README.md new file mode 100644 index 00000000000..576e729bf99 --- /dev/null +++ b/sdks/csharp/tests~/event-handling-benchmarks/README.md @@ -0,0 +1,36 @@ +# C# event handling benchmarks + +This benchmark client compares native C# event dispatch with the SDK custom event listener backend. It measures elapsed time and process-wide allocated bytes for table event subscription, update dispatch, unsubscription, and resubscription against a real SpacetimeDB module. + +It reuses the C# regression-test module and generated bindings. Publish that module first, then run this client against it. + +```sh +spacetime start +spacetime publish --module-path sdks/csharp/examples~/regression-tests/server event-handling-bench +dotnet run -c Release --project sdks/csharp/tests~/event-handling-benchmarks/client/client.csproj +``` + +Environment variables: + +- `SPACETIMEDB_SERVER_URL`: server URL, default `http://localhost:3000`. +- `SPACETIMEDB_DATABASE`: database name, default `event-handling-bench`. +- `SPACETIMEDB_EVENT_BACKEND`: `all`, `native`, `custom`, or `sappy`. `all` runs `native` and `custom`. +- `SPACETIMEDB_BENCHMARK_WARMUPS`: warmup runs per scenario/backend, default `1`. +- `SPACETIMEDB_BENCHMARK_ITERATIONS`: measured runs per scenario/backend, default `3`. + +Backends: + +- `native`: native C# multicast delegate dispatch. +- `custom`: SDK custom indexed listener dispatch. +- `sappy`: Sappy-backed custom listener dispatch. This path is only run when explicitly requested and requires compiling with `SAPPY=1` in a project that references the Sappy package, such as a Unity project with the SDK and Sappy integration assemblies present. The Sappy benchmark registers generated Sappy targets through `OnInsertListeners` so it exercises the intended Sappy path. + +Scenarios: + +- Few subscriptions, many updates. +- Many subscriptions, some updates. +- Many subscriptions, many updates. +- Many subscriptions, some updates, many unsubscriptions. +- Many subscriptions, some updates, many unsubscriptions, many resubscriptions, some updates. +- Many subscriptions, many updates, many unsubscriptions, many resubscriptions, many updates. + +The update timings and allocation counts include reducer calls, server round trips, client frame ticks, row decoding, and listener dispatch. Allocation counts use `GC.GetTotalAllocatedBytes(precise: true)`, so they include allocations from background client work as well as main-thread dispatch. Compare native and custom runs on the same machine/server rather than treating the values as isolated in-process dispatch costs. diff --git a/sdks/csharp/tests~/event-handling-benchmarks/client/Program.cs b/sdks/csharp/tests~/event-handling-benchmarks/client/Program.cs new file mode 100644 index 00000000000..a034dc117dd --- /dev/null +++ b/sdks/csharp/tests~/event-handling-benchmarks/client/Program.cs @@ -0,0 +1,535 @@ +using System.Diagnostics; +using RegressionTests.Shared; +using SpacetimeDB; +using SpacetimeDB.EventHandling; +using SpacetimeDB.Types; +using ExampleDataInsertHandler = SpacetimeDB.RemoteTableHandleBase.RowEventHandler; +#if SAPPY +using Sappy; +using SpacetimeDB.SappyIntegration; +#endif + +const string DefaultHost = "http://localhost:3000"; +const string DefaultDatabase = "event-handling-bench"; +const int DefaultWarmups = 1; +const int DefaultIterations = 3; + +var host = Environment.GetEnvironmentVariable("SPACETIMEDB_SERVER_URL") ?? DefaultHost; +var database = Environment.GetEnvironmentVariable("SPACETIMEDB_DATABASE") ?? DefaultDatabase; +var backend = ParseBackend(Environment.GetEnvironmentVariable("SPACETIMEDB_EVENT_BACKEND") ?? "all"); +var warmups = ParseNonNegativeEnv("SPACETIMEDB_BENCHMARK_WARMUPS", DefaultWarmups); +var iterations = ParsePositiveEnv("SPACETIMEDB_BENCHMARK_ITERATIONS", DefaultIterations); +var summaries = new List(); + +var scenarios = new[] +{ + new Scenario("few-subscriptions-many-updates", Subscriptions: 10, FirstUpdates: 1_000), + new Scenario("many-subscriptions-some-updates", Subscriptions: 1_000, FirstUpdates: 10), + new Scenario("many-subscriptions-many-updates", Subscriptions: 1_000, FirstUpdates: 1_000), + new Scenario("many-subscriptions-some-updates-many-unsubscriptions", Subscriptions: 1_000, FirstUpdates: 10, Unsubscriptions: 1_000), + new Scenario("many-subscriptions-some-updates-many-unsubscriptions-resubscriptions-some-updates", Subscriptions: 1_000, FirstUpdates: 10, Unsubscriptions: 1_000, Resubscriptions: 1_000, SecondUpdates: 10), + new Scenario("many-subscriptions-some-updates-many-unsubscriptions-resubscriptions-some-updates", Subscriptions: 1_000, FirstUpdates: 1_000, Unsubscriptions: 1_000, Resubscriptions: 1_000, SecondUpdates: 1_000), +}; + +RegressionTestHarness.RegisterUnhandledExceptionExitHandler(); + +Console.WriteLine($"Host: {host}"); +Console.WriteLine($"Database: {database}"); +Console.WriteLine($"Warmups: {warmups}"); +Console.WriteLine($"Iterations: {iterations}"); +Console.WriteLine(); +Console.WriteLine("Per-iteration timings:"); +Console.WriteLine("| Backend | Scenario | Iteration | Subscriptions | First updates | Unsubscriptions | Resubscriptions | Second updates | Subscribe ms | Subscribe bytes | First updates ms | First updates bytes | Unsubscribe ms | Unsubscribe bytes | Resubscribe ms | Resubscribe bytes | Second updates ms | Second updates bytes | Listener calls |"); +Console.WriteLine("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"); + +foreach (var backendKind in ExpandBackends(backend)) +{ + ConfigureBackend(backendKind); + + foreach (var scenario in scenarios) + { + for (var i = 0; i < warmups; i++) + { + RunOne(host, database, backendKind, scenario); + } + + var results = new BenchmarkResult[iterations]; + for (var i = 0; i < iterations; i++) + { + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + + results[i] = RunOne(host, database, backendKind, scenario); + PrintResult(backendKind, scenario, i + 1, results[i]); + } + + summaries.Add(BenchmarkSummary.From(backendKind, scenario, results)); + } +} + +Console.WriteLine(); +Console.WriteLine("Summary timings:"); +Console.WriteLine("| Backend | Scenario | Subscribe ms avg | Subscribe bytes avg | First updates ms avg | First updates bytes avg | Unsubscribe ms avg | Unsubscribe bytes avg | Resubscribe ms avg | Resubscribe bytes avg | Second updates ms avg | Second updates bytes avg |"); +Console.WriteLine("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"); +foreach (var summary in summaries) +{ + Console.WriteLine( + $"| {summary.Backend} | {summary.Scenario.Name} | " + + $"{summary.Subscribe.MeanMilliseconds:F3} | {summary.Subscribe.MeanAllocatedBytes:F0} | " + + $"{summary.FirstUpdates.MeanMilliseconds:F3} | {summary.FirstUpdates.MeanAllocatedBytes:F0} | " + + $"{summary.Unsubscribe.MeanMilliseconds:F3} | {summary.Unsubscribe.MeanAllocatedBytes:F0} | " + + $"{summary.Resubscribe.MeanMilliseconds:F3} | {summary.Resubscribe.MeanAllocatedBytes:F0} | " + + $"{summary.SecondUpdates.MeanMilliseconds:F3} | {summary.SecondUpdates.MeanAllocatedBytes:F0} |" + ); +} + +static BackendKind ParseBackend(string value) => + value.Trim().ToLowerInvariant() switch + { + "all" => BackendKind.All, + "native" => BackendKind.Native, + "custom" or "basic" => BackendKind.Custom, + "sappy" => BackendKind.Sappy, + _ => throw new ArgumentOutOfRangeException(nameof(value), value, "Expected all, native, custom, basic, or sappy."), + }; + +static IEnumerable ExpandBackends(BackendKind backend) +{ + if (backend != BackendKind.All) + { + yield return backend; + yield break; + } + + yield return BackendKind.Native; + yield return BackendKind.Custom; +} + +static void ConfigureBackend(BackendKind backend) +{ + switch (backend) + { + case BackendKind.Native: + Backend.UseNativeEvents(); + return; + case BackendKind.Custom: + Backend.UseCustomListeners(); + return; + case BackendKind.Sappy: +#if SAPPY + Backend.UseCustomListeners(new SappyEventListenersFactory()); + return; +#else + throw new InvalidOperationException("The Sappy benchmark requires compiling with SAPPY=1 inside a project that references Sappy."); +#endif + default: + throw new ArgumentOutOfRangeException(nameof(backend), backend, null); + } +} + +static BenchmarkResult RunOne(string host, string database, BackendKind backendKind, Scenario scenario) +{ + using var runner = new BenchmarkRunner(host, database, backendKind, scenario); + return runner.Run(); +} + +static void PrintResult(BackendKind backendKind, Scenario scenario, int iteration, BenchmarkResult result) +{ + Console.WriteLine( + $"| {backendKind} | {scenario.Name} | {iteration} | {scenario.Subscriptions} | {scenario.FirstUpdates} | {scenario.Unsubscriptions} | {scenario.Resubscriptions} | {scenario.SecondUpdates} | " + + $"{result.Subscribe.Elapsed.TotalMilliseconds:F3} | {result.Subscribe.AllocatedBytes} | " + + $"{result.FirstUpdates.Elapsed.TotalMilliseconds:F3} | {result.FirstUpdates.AllocatedBytes} | " + + $"{result.Unsubscribe.Elapsed.TotalMilliseconds:F3} | {result.Unsubscribe.AllocatedBytes} | " + + $"{result.Resubscribe.Elapsed.TotalMilliseconds:F3} | {result.Resubscribe.AllocatedBytes} | " + + $"{result.SecondUpdates.Elapsed.TotalMilliseconds:F3} | {result.SecondUpdates.AllocatedBytes} | {result.ListenerCalls} |" + ); +} + +static int ParseNonNegativeEnv(string name, int defaultValue) +{ + var value = Environment.GetEnvironmentVariable(name); + if (string.IsNullOrWhiteSpace(value)) + { + return defaultValue; + } + + if (int.TryParse(value, out var parsed) && parsed >= 0) + { + return parsed; + } + + throw new ArgumentOutOfRangeException(name, value, "Expected a non-negative integer."); +} + +static int ParsePositiveEnv(string name, int defaultValue) +{ + var value = Environment.GetEnvironmentVariable(name); + if (string.IsNullOrWhiteSpace(value)) + { + return defaultValue; + } + + if (int.TryParse(value, out var parsed) && parsed > 0) + { + return parsed; + } + + throw new ArgumentOutOfRangeException(name, value, "Expected a positive integer."); +} + +internal enum BackendKind +{ + All, + Native, + Custom, + Sappy, +} + +internal sealed record Scenario( + string Name, + int Subscriptions, + int FirstUpdates, + int Unsubscriptions = 0, + int Resubscriptions = 0, + int SecondUpdates = 0 +); + +internal readonly record struct BenchmarkResult( + Measurement Subscribe, + Measurement FirstUpdates, + Measurement Unsubscribe, + Measurement Resubscribe, + Measurement SecondUpdates, + long ListenerCalls +); + +internal readonly record struct Measurement(TimeSpan Elapsed, long AllocatedBytes) +{ + public static Measurement Zero { get; } = new(TimeSpan.Zero, 0); +} + +internal readonly record struct Stats( + double MeanMilliseconds, + double MinMilliseconds, + double MaxMilliseconds, + double MeanAllocatedBytes, + long MinAllocatedBytes, + long MaxAllocatedBytes +) +{ + public static Stats From(IEnumerable values) + { + var measurements = values.ToArray(); + var milliseconds = measurements.Select(value => value.Elapsed.TotalMilliseconds).ToArray(); + var allocatedBytes = measurements.Select(value => value.AllocatedBytes).ToArray(); + return new Stats( + milliseconds.Average(), + milliseconds.Min(), + milliseconds.Max(), + allocatedBytes.Average(), + allocatedBytes.Min(), + allocatedBytes.Max() + ); + } +} + +internal readonly record struct BenchmarkSummary( + BackendKind Backend, + Scenario Scenario, + Stats Subscribe, + Stats FirstUpdates, + Stats Unsubscribe, + Stats Resubscribe, + Stats SecondUpdates +) +{ + public static BenchmarkSummary From(BackendKind backend, Scenario scenario, BenchmarkResult[] results) => + new( + backend, + scenario, + Stats.From(results.Select(r => r.Subscribe)), + Stats.From(results.Select(r => r.FirstUpdates)), + Stats.From(results.Select(r => r.Unsubscribe)), + Stats.From(results.Select(r => r.Resubscribe)), + Stats.From(results.Select(r => r.SecondUpdates)) + ); +} + +internal static class BenchmarkSettings +{ + public const int TimeoutSeconds = 120; + public const int FrameSleepMilliseconds = 1; +} + +internal sealed class BenchmarkRunner : IDisposable +{ + private readonly string _host; + private readonly string _database; + private readonly BackendKind _backend; + private readonly Scenario _scenario; + private readonly ExampleDataInsertHandler[] _listeners; + private readonly Listener[] _listenerTargets; +#if SAPPY + private readonly SapTarget[]? _sapTargets; +#endif + private readonly object _lock = new(); + private DbConnection _conn = null!; + private SubscriptionHandle? _subscription; + private long _listenerCalls; + private long _targetListenerCalls; + private bool _connected; + private bool _subscriptionApplied; + private Exception? _error; + private uint _nextId; + + public BenchmarkRunner(string host, string database, BackendKind backend, Scenario scenario) + { + _host = host; + _database = database; + _backend = backend; + _scenario = scenario; + _listeners = new ExampleDataInsertHandler[scenario.Subscriptions]; + _listenerTargets = new Listener[scenario.Subscriptions]; +#if SAPPY + if (backend == BackendKind.Sappy) + { + _sapTargets = new SapTarget[scenario.Subscriptions]; + } +#endif + + var idBase = unchecked((uint)HashCode.Combine(Environment.ProcessId, DateTime.UtcNow.Ticks, backend, scenario.Name)); + _nextId = idBase == 0 ? 1 : idBase; + + for (var i = 0; i < _listeners.Length; i++) + { + _listenerTargets[i] = new Listener(this); + _listeners[i] = _listenerTargets[i].OnExampleDataInsert; +#if SAPPY + if (_sapTargets != null) + { + _sapTargets[i] = _listenerTargets[i].Sappy.OnExampleDataInsert; + } +#endif + } + } + + public BenchmarkResult Run() + { + Connect(); + Subscribe(); + + var subscribe = Time(() => + { + for (var i = 0; i < _listeners.Length; i++) + { + AddListener(i); + } + }); + + var firstUpdates = TimeUpdates(_scenario.FirstUpdates, _scenario.Subscriptions); + + var unsubscribe = Time(() => + { + for (var i = 0; i < _scenario.Unsubscriptions; i++) + { + RemoveListener(i); + } + }); + + var resubscribe = Time(() => + { + for (var i = 0; i < _scenario.Resubscriptions; i++) + { + AddListener(i); + } + }); + + var activeAfterResubscribe = _scenario.Subscriptions - _scenario.Unsubscriptions + _scenario.Resubscriptions; + var secondUpdates = _scenario.SecondUpdates > 0 ? TimeUpdates(_scenario.SecondUpdates, activeAfterResubscribe) : Measurement.Zero; + + return new BenchmarkResult( + subscribe, + firstUpdates, + unsubscribe, + resubscribe, + secondUpdates, + Interlocked.Read(ref _listenerCalls) + ); + } + + private void AddListener(int index) + { +#if SAPPY + if (_backend == BackendKind.Sappy) + { + _conn.Db.ExampleData.OnInsertListeners.AddSapTarget(_sapTargets![index]); + return; + } +#endif + _conn.Db.ExampleData.OnInsert += _listeners[index]; + } + + private void RemoveListener(int index) + { +#if SAPPY + if (_backend == BackendKind.Sappy) + { + _conn.Db.ExampleData.OnInsertListeners.RemoveSapTarget(_sapTargets![index]); + return; + } +#endif + _conn.Db.ExampleData.OnInsert -= _listeners[index]; + } + + private void Connect() + { + _conn = RegressionTestHarness.ConnectToDatabase( + _host, + _database, + (conn, _, _) => + { + lock (_lock) + { + _connected = true; + } + }, + error => RecordError(error), + error => + { + if (error != null) + { + RecordError(error); + } + } + ); + + TickUntil(() => _connected, "connect"); + } + + private void Subscribe() + { + _subscription = _conn.SubscriptionBuilder() + .OnApplied(_ => + { + lock (_lock) + { + _subscriptionApplied = true; + } + }) + .OnError((_, error) => RecordError(error)) + .AddQuery(q => q.From.ExampleData()) + .Subscribe(); + + TickUntil(() => _subscriptionApplied, "subscription applied"); + } + + private Measurement TimeUpdates(int updateCount, int activeSubscriptions) + { + if (updateCount <= 0) + { + return Measurement.Zero; + } + + var expectedCalls = activeSubscriptions * updateCount; + var before = Interlocked.Read(ref _listenerCalls); + Interlocked.Exchange(ref _targetListenerCalls, before + expectedCalls); + + return Time(() => + { + for (var i = 0; i < updateCount; i++) + { + _conn.Reducers.Add(NextId(), (uint)i); + } + + TickUntil(() => Interlocked.Read(ref _listenerCalls) >= Interlocked.Read(ref _targetListenerCalls), $"{updateCount} updates for {_scenario.Name}/{_backend}"); + }); + } + + private uint NextId() + { + var id = _nextId++; + if (id == 0) + { + id = _nextId++; + } + + return id; + } + + private void RecordInsert() + { + Interlocked.Increment(ref _listenerCalls); + } + + private Measurement Time(Action action) + { + var allocatedBefore = GC.GetTotalAllocatedBytes(precise: true); + var stopwatch = Stopwatch.StartNew(); + action(); + stopwatch.Stop(); + var allocatedAfter = GC.GetTotalAllocatedBytes(precise: true); + return new Measurement(stopwatch.Elapsed, allocatedAfter - allocatedBefore); + } + + private void TickUntil(Func complete, string phase) + { + var deadline = DateTime.UtcNow.AddSeconds(BenchmarkSettings.TimeoutSeconds); + while (!complete()) + { + ThrowIfError(); + _conn.FrameTick(); + Thread.Sleep(BenchmarkSettings.FrameSleepMilliseconds); + + if (DateTime.UtcNow > deadline) + { + throw new TimeoutException($"Timed out waiting for {phase}."); + } + } + + ThrowIfError(); + } + + private void RecordError(Exception error) + { + lock (_lock) + { + _error ??= error; + } + } + + private void ThrowIfError() + { + lock (_lock) + { + if (_error != null) + { + throw new InvalidOperationException("Benchmark connection failed.", _error); + } + } + } + + public void Dispose() + { + _subscription?.UnsubscribeThen(_ => { }); + _conn?.Disconnect(); + } + + private sealed partial class Listener + { + private readonly BenchmarkRunner _runner; + + public Listener(BenchmarkRunner runner) + { + _runner = runner; + } + +#if SAPPY + [SapTarget(typeof(ExampleDataInsertHandler))] +#endif + public void OnExampleDataInsert(EventContext ctx, ExampleData row) + { + _runner.RecordInsert(); + } + } +} diff --git a/sdks/csharp/tests~/event-handling-benchmarks/client/client.csproj b/sdks/csharp/tests~/event-handling-benchmarks/client/client.csproj new file mode 100644 index 00000000000..a6ea7569fa1 --- /dev/null +++ b/sdks/csharp/tests~/event-handling-benchmarks/client/client.csproj @@ -0,0 +1,23 @@ + + + + Exe + net8.0 + enable + enable + + + + $(DefineConstants);SAPPY + + + + + + + + + + + +