diff --git a/sdks/csharp/examples~/event-handling-benchmarks/README.md b/sdks/csharp/examples~/event-handling-benchmarks/README.md new file mode 100644 index 00000000000..62c72beffdb --- /dev/null +++ b/sdks/csharp/examples~/event-handling-benchmarks/README.md @@ -0,0 +1,31 @@ +# C# event handling benchmarks + +This benchmark client measures 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 event-handling-bench sdks/csharp/examples~/regression-tests/server +dotnet run -c Release --project sdks/csharp/examples~/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`. + +Backends: + +- `native`: native C# multicast delegate dispatch. +- `custom`: SDK custom indexed listener dispatch. +- `sappy`: Sappy-backed custom listener dispatch. This path 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. + +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. diff --git a/sdks/csharp/examples~/event-handling-benchmarks/client/Program.cs b/sdks/csharp/examples~/event-handling-benchmarks/client/Program.cs new file mode 100644 index 00000000000..5bfb3615ba6 --- /dev/null +++ b/sdks/csharp/examples~/event-handling-benchmarks/client/Program.cs @@ -0,0 +1,353 @@ +using System.Diagnostics; +using RegressionTests.Shared; +using SpacetimeDB; +using SpacetimeDB.EventHandling; +using SpacetimeDB.Types; +using ExampleDataInsertHandler = SpacetimeDB.RemoteTableHandleBase.RowEventHandler; +#if SAPPY +using SpacetimeDB.SappyIntegration; +#endif + +const string DefaultHost = "http://localhost:3000"; +const string DefaultDatabase = "event-handling-bench"; +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 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), +}; + +RegressionTestHarness.RegisterUnhandledExceptionExitHandler(); + +Console.WriteLine($"Host: {host}"); +Console.WriteLine($"Database: {database}"); +Console.WriteLine("| Backend | Scenario | Subscriptions | First updates | Unsubscriptions | Resubscriptions | Second updates | Subscribe ms | First updates ms | Unsubscribe ms | Resubscribe ms | Second updates ms | Listener calls |"); +Console.WriteLine("| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |"); + +foreach (var backendKind in ExpandBackends(backend)) +{ + ConfigureBackend(backendKind); + + foreach (var scenario in scenarios) + { + using var runner = new BenchmarkRunner(host, database, backendKind, scenario); + var result = runner.Run(); + Console.WriteLine( + $"| {backendKind} | {scenario.Name} | {scenario.Subscriptions} | {scenario.FirstUpdates} | {scenario.Unsubscriptions} | {scenario.Resubscriptions} | {scenario.SecondUpdates} | " + + $"{result.Subscribe.TotalMilliseconds:F3} | {result.FirstUpdates.TotalMilliseconds:F3} | {result.Unsubscribe.TotalMilliseconds:F3} | {result.Resubscribe.TotalMilliseconds:F3} | {result.SecondUpdates.TotalMilliseconds:F3} | {result.ListenerCalls} |" + ); + } +} + +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; +#if SAPPY + yield return BackendKind.Sappy; +#endif +} + +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); + } +} + +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( + TimeSpan Subscribe, + TimeSpan FirstUpdates, + TimeSpan Unsubscribe, + TimeSpan Resubscribe, + TimeSpan SecondUpdates, + long ListenerCalls +); + +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; + 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]; + + 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; + } + } + + public BenchmarkResult Run() + { + Connect(); + Subscribe(); + + var subscribe = Time(() => + { + foreach (var listener in _listeners) + { + _conn.Db.ExampleData.OnInsert += listener; + } + }); + + var firstUpdates = TimeUpdates(_scenario.FirstUpdates, _scenario.Subscriptions); + + var unsubscribe = Time(() => + { + for (var i = 0; i < _scenario.Unsubscriptions; i++) + { + _conn.Db.ExampleData.OnInsert -= _listeners[i]; + } + }); + + var resubscribe = Time(() => + { + for (var i = 0; i < _scenario.Resubscriptions; i++) + { + _conn.Db.ExampleData.OnInsert += _listeners[i]; + } + }); + + var activeAfterResubscribe = _scenario.Subscriptions - _scenario.Unsubscriptions + _scenario.Resubscriptions; + var secondUpdates = _scenario.SecondUpdates > 0 ? TimeUpdates(_scenario.SecondUpdates, activeAfterResubscribe) : TimeSpan.Zero; + + return new BenchmarkResult( + subscribe, + firstUpdates, + unsubscribe, + resubscribe, + secondUpdates, + Interlocked.Read(ref _listenerCalls) + ); + } + + 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 TimeSpan TimeUpdates(int updateCount, int activeSubscriptions) + { + if (updateCount <= 0) + { + return TimeSpan.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 TimeSpan Time(Action action) + { + var stopwatch = Stopwatch.StartNew(); + action(); + stopwatch.Stop(); + return stopwatch.Elapsed; + } + + 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 class Listener + { + private readonly BenchmarkRunner _runner; + + public Listener(BenchmarkRunner runner) + { + _runner = runner; + } + + public void OnExampleDataInsert(EventContext ctx, ExampleData row) + { + _runner.RecordInsert(); + } + } +} diff --git a/sdks/csharp/examples~/event-handling-benchmarks/client/client.csproj b/sdks/csharp/examples~/event-handling-benchmarks/client/client.csproj new file mode 100644 index 00000000000..8fee3d03556 --- /dev/null +++ b/sdks/csharp/examples~/event-handling-benchmarks/client/client.csproj @@ -0,0 +1,23 @@ + + + + Exe + net8.0 + enable + enable + + + + $(DefineConstants);SAPPY + + + + + + + + + + + + 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 5847f5728df..c37faccebe8 100644 --- a/sdks/csharp/src/Table.cs +++ b/sdks/csharp/src/Table.cs @@ -196,18 +196,8 @@ public RemoteTableHandleBase(IDbConnection conn, bool isEventTable = false) : ba // 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); @@ -407,13 +397,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; @@ -508,14 +502,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 { @@ -526,7 +520,7 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa { if (oldValue is Row oldRow) { - OnInternalDeleteHandler.Invoke(oldRow); + OnInternalDelete?.Invoke(oldRow); } else { @@ -536,7 +530,7 @@ void IRemoteTableHandle.Apply(IEventContext context, IParsedTableUpdate parsedTa if (newValue is Row newRow) { - OnInternalInsertHandler.Invoke(newRow); + OnInternalInsert?.Invoke(newRow); } else { @@ -577,33 +571,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.EventListenersProvider.UseBasicEventListeners() or a custom listener factory 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.EventListenersProvider.UseBasicEventListeners() or a custom listener factory 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); + } + } } } @@ -619,21 +695,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 26ce87127ba..c487f75171e 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(); + } + } +}