diff --git a/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs b/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs new file mode 100644 index 000000000..cce51279f --- /dev/null +++ b/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs @@ -0,0 +1,42 @@ +using System.Threading; +using BenchmarkDotNet.Attributes; + +namespace DynamicData.Benchmarks.Miscellaneous; + +[MemoryDiagnoser] +[MarkdownExporterAttribute.GitHub] +public class LockImplementations +{ + public LockImplementations() + { + _objectGate = new(); + _threadingGate = new(); + } + + [Benchmark(Baseline = true)] + public int NoLock() + { + return 0; + } + + [Benchmark] + public int ObjectLock() + { + lock (_objectGate) + { + return 0; + } + } + + [Benchmark] + public int ThreadingLock() + { + lock (_threadingGate) + { + return 0; + } + } + + private readonly object _objectGate; + private readonly Lock _threadingGate; +} diff --git a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs index 99b79fbfd..38fc02e73 100644 --- a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs @@ -1,12 +1,13 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; using DynamicData.Tests.Domain; - +using DynamicData.Tests.Utilities; using FluentAssertions; using Xunit; @@ -162,8 +163,6 @@ public void EmptyChangesWithFilter() change!.Count.Should().Be(0); } - - [Fact] public void StaticFilterRemove() { @@ -191,7 +190,6 @@ public void StaticFilterRemove() public record class SomeObject(int Id, int Value); - [Fact] public async Task MultiCacheFanInDoesNotDeadlock() { @@ -354,5 +352,75 @@ public void ConnectDuringDeliveryDoesNotDuplicate() addCounts.GetValueOrDefault("k2").Should().Be(1, "k2 should appear once, not duplicated from snapshot + queued delivery"); } + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditsDoesNotDuplicate() + { + using var items = new SourceCache(static item => item); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + CacheItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.AddOrUpdate(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets(static item => item) + .RecordCacheItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.AddOrUpdate(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + // Explicitly doing a nested edit, as that system is closely intertwined with the edit-tracking system that + // .Connect() uses. + items.Remove(item: 1); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItemsByKey.Should().BeEquivalentTo( + new Dictionary() { [2] = 2 }, + options => options.WithoutStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + + [Fact] + public void ConnectContinuesToWorkNormallyAfterAFailedEdit() + { + using var source = new SourceCache(static item => item); + + source.AddOrUpdate(1); + + source.Invoking(source => source.Edit(_ => throw new Exception("Test"))) + .Should().Throw() + .WithMessage("Test"); + + using var subscription = source.Connect().RecordCacheItems(out var results); + + results.Error.Should().BeNull("new subscribers should not receive previous errors"); + results.RecordedChangeSets.Should().ContainSingle("an initial changeset should have been published."); + } + private sealed record TestItem(string Key, string Value); } diff --git a/src/DynamicData.Tests/List/SourceListFixture.cs b/src/DynamicData.Tests/List/SourceListFixture.cs index dc37bdaed..06905c5a4 100644 --- a/src/DynamicData.Tests/List/SourceListFixture.cs +++ b/src/DynamicData.Tests/List/SourceListFixture.cs @@ -1,13 +1,63 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; + using FluentAssertions; using Xunit; +using DynamicData.Tests.Utilities; + namespace DynamicData.Tests.List; public class SourceListFixture { + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditDoesNotDuplicate() + { + using var items = new SourceList(); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + ListItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.Add(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets() + .RecordListItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.Add(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItems.Should().BeEquivalentTo( + new[] { 1, 2, }, + options => options.WithStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + [Fact] public void InitialChangeIsRange() { @@ -21,4 +71,21 @@ public void InitialChangeIsRange() changeSets[0].First().Type.Should().Be(ChangeType.Range); changeSets[0].First().Range.Index.Should().Be(0); } + + [Fact] + public void ConnectContinuesToWorkNormallyAfterAFailedEdit() + { + using var source = new SourceList(); + + source.Add(1); + + source.Invoking(source => source.Edit(_ => throw new Exception("Test"))) + .Should().Throw() + .WithMessage("Test"); + + using var subscription = source.Connect().RecordListItems(out var results); + + results.Error.Should().BeNull("new subscribers should not receive previous errors"); + results.RecordedChangeSets.Should().ContainSingle("an initial changeset should have been published."); + } } diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 60642ee29..745e06ba1 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -6,9 +6,8 @@ using System.Reactive.Disposables; using System.Reactive.Linq; using System.Reactive.Subjects; -using System.Threading; + using DynamicData.Binding; -using DynamicData.Cache; using DynamicData.Cache.Internal; // ReSharper disable once CheckNamespace @@ -42,6 +41,9 @@ internal sealed class ObservableCache : IObservableCache _notifications; + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposed with _cleanUp")] + private readonly Lazy> _isEditInProgress; + private int _editLevel; // The level of recursion in editing. private long _currentVersion; // Monotonic counter incremented under lock for each enqueued change notification. @@ -53,6 +55,7 @@ public ObservableCache(IObservable> source) _readerWriter = new ReaderWriter(); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(() => new(_editLevel is not 0)); var loader = source.Subscribe( changeSet => @@ -83,6 +86,7 @@ public ObservableCache(Func? keySelector = null) _readerWriter = new ReaderWriter(keySelector); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(() => new(_editLevel is not 0)); _cleanUp = Disposable.Create(NotifyCompleted); } @@ -110,28 +114,10 @@ public ObservableCache(Func? keySelector = null) public IReadOnlyDictionary KeyValues => _readerWriter.KeyValues; - public IObservable> Connect(Func? predicate = null, bool suppressEmptyChangeSets = true) => - Observable.Create>(observer => - { - lock (_locker) - { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) - - // Create the Connection Observable - ? CreateConnectObservable(predicate, suppressEmptyChangeSets) - - // Defer until notifications are no longer suspended. Take(1) means there is only - // ever one inner sequence, so SelectMany carries the terminal event of the gate - // through on its own: the connection ends when the cache does, and fails when it - // fails, rather than reporting a failure as a successful completion. - : _suspensionTracker.Value.NotificationsSuspendedObservable - .Where(static areNotificationsSuspended => !areNotificationsSuspended) - .Take(1) - .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); - - return observable.SubscribeSafe(observer); - } - }); + public IObservable> Connect( + Func? predicate = null, + bool suppressEmptyChangeSets = true) + => CreateDeferredNotificationObservable(() => CreateConnectObservable(predicate, suppressEmptyChangeSets)); public void Dispose() => _cleanUp.Dispose(); @@ -139,26 +125,8 @@ public IObservable> Connect(Func? predi public IObservable> Preview(Func? predicate = null) => predicate is null ? _changesPreview : _changesPreview.Filter(predicate); - public IObservable> Watch(TKey key) => - Observable.Create>(observer => - { - lock (_locker) - { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) - - // Create the Watch Observable - ? CreateWatchObservable(key) - - // Defer until notifications are no longer suspended. See Connect() for why - // SelectMany is used here. - : _suspensionTracker.Value.NotificationsSuspendedObservable - .Where(static areNotificationsSuspended => !areNotificationsSuspended) - .Take(1) - .SelectMany(_ => CreateWatchObservable(key)); - - return observable.SubscribeSafe(observer); - } - }); + public IObservable> Watch(TKey key) + => CreateDeferredNotificationObservable(() => CreateWatchObservable(key)); public IDisposable SuspendCount() { @@ -189,21 +157,31 @@ internal void UpdateFromIntermediate(Action> update ChangeSet? changes = null; _editLevel++; - if (_editLevel == 1) + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); + try { - var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; - changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + if (_editLevel == 1) + { + var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; + changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + } + else + { + _readerWriter.WriteNested(updateAction); + } } - else + finally { - _readerWriter.WriteNested(updateAction); - } + _editLevel--; - _editLevel--; + if (changes is not null && _editLevel == 0) + { + notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + } - if (changes is not null && _editLevel == 0) - { - notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } @@ -216,21 +194,31 @@ internal void UpdateFromSource(Action> updateActio ChangeSet? changes = null; _editLevel++; - if (_editLevel == 1) + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); + try { - var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; - changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + if (_editLevel == 1) + { + var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; + changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + } + else + { + _readerWriter.WriteNested(updateAction); + } } - else + finally { - _readerWriter.WriteNested(updateAction); - } + _editLevel--; - _editLevel--; + if (changes is not null && _editLevel == 0) + { + notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + } - if (changes is not null && _editLevel == 0) - { - notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } @@ -264,6 +252,65 @@ private IObservable> CreateConnectObservable(Func CreateDeferredNotificationObservable(Func> factory) + => Observable.Create(observer => + { + lock (_locker) + { + var observable = ( + // A suspension can't be in-progress if the suspension system hasn't been activated. + _suspensionTracker.IsValueCreated, + // An edit can be in-progress before the edit-tracking notification system is activated, so this + // one needs an extra check. + _isEditInProgress.IsValueCreated || (_editLevel is not 0)) + switch + { + // Neither the suspension system nor the edit system is active, create the connection + // immediately. + (false, false) => factory.Invoke(), + + // Edit system is active, suspension system isn't + // Need to avoid activating the suspension system if we don't absolutely need to, as it adds + // locking overhead to edit operations. But then when the edit is done, we need to check if a + // suspension came in. If so, do a followup wait with the full logic for both systems. It + // needs to be the full logic, in case an edit comes in during the suspension, and so on. + (false, true) => _isEditInProgress.Value + .Where(static isEditInProgress => !isEditInProgress) + .Take(1) + .SelectMany(_ => (_suspensionTracker.IsValueCreated && _suspensionTracker.Value.AreNotificationsSuspended) + ? CreateFullyDeferredConnection() + : factory.Invoke()), + + // Suspension system is active, edit system isn't + // Same case as above, but reversed. Just wait on the suspension system, but then do a followup + // wait if needed. + (true, false) => _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static isSuspensionInProgress => !isSuspensionInProgress) + .Take(1) + .SelectMany(_ => (_isEditInProgress.IsValueCreated && _isEditInProgress.Value.Value) + ? CreateFullyDeferredConnection() + : factory.Invoke()), + + // If both systems are already active, we can monitor both systems simultaneously, and make the + // connection as soon as both are idle at the same time. + _ => CreateFullyDeferredConnection() + }; + + return observable.SubscribeSafe(observer); + + IObservable CreateFullyDeferredConnection() + => Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress.Value, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Do(static _ => { }, observer.OnCompleted) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) + .Take(1) + .Select(_ => factory.Invoke()) + .Switch(); + } + }); + private IObservable> CreateWatchObservable(TKey key) => Observable.Create>( observer => @@ -395,6 +442,11 @@ public void OnError(Exception error) cache._changesPreview.OnError(error); cache._changes.OnError(error); + if (cache._isEditInProgress.IsValueCreated) + { + cache._isEditInProgress.Value.OnError(error); + } + if (cache._countChanged.IsValueCreated) { cache._countChanged.Value.OnError(error); @@ -411,6 +463,11 @@ public void OnCompleted() cache._changes.OnCompleted(); cache._changesPreview.OnCompleted(); + if (cache._isEditInProgress.IsValueCreated) + { + cache._isEditInProgress.Value.OnCompleted(); + } + if (cache._countChanged.IsValueCreated) { cache._countChanged.Value.OnCompleted(); diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 3db42bc11..a159af490 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -36,6 +36,9 @@ public sealed class SourceList : ISourceList private readonly ReaderWriter _readerWriter = new(); + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposal is superfluous after completion, and causes a bunch of test failures")] + private readonly Lazy> _isEditInProgress; + private int _editLevel; /// @@ -44,6 +47,8 @@ public sealed class SourceList : ISourceList /// The source. public SourceList(IObservable>? source = null) { + _isEditInProgress = new(() => new(_editLevel is not 0)); + var loader = source is null ? Disposable.Empty : LoadFromSource(source); _cleanUp = Disposable.Create( @@ -78,40 +83,34 @@ public SourceList(IObservable>? source = null) /// public IObservable> Connect(Func? predicate = null) - { - var observable = Observable.Create>( - observer => + => Observable.Create>(observer => + { + lock (_locker) { - lock (_locker) - { - if (_readerWriter.Items.Length > 0) - { - observer.OnNext( - new ChangeSet - { - new(ListChangeReason.AddRange, _readerWriter.Items, 0) - }); - } + var observable = _isEditInProgress.IsValueCreated || (_editLevel is not 0) - var source = _changes.Finally(observer.OnCompleted); + // Defer connection until there is no longer an in-progress edit. + ? _isEditInProgress.Value + .Where(static isEditInProgress => !isEditInProgress) + .Take(1) + .SelectMany(_ => CreateConnectObservable(predicate)) - return source.SubscribeSafe(observer); - } - }); + // Otherwise, just connect immediately, and avoid forcing the edit-tracking system to initialize. + : CreateConnectObservable(predicate); - if (predicate is not null) - { - observable = new FilterStatic(observable, predicate).Run(); - } - - return observable; - } + return observable.SubscribeSafe(observer); + } + }); /// public void Dispose() { _cleanUp.Dispose(); _changesPreview.Dispose(); + // Intentionally skipping disposal for _isEditInProgress, as it's technically redundant after _cleanUp.Dispose() + // calls .OnCompleted(), and doing disposal anyway causes a whole bunch of test failures. That really suggests + // we need to rework the lifecycle mechanics of this class, as a whole, but that's probably going to involve + // breaking changes. } /// @@ -124,21 +123,35 @@ public void Edit(Action> updateAction) IChangeSet? changes = null; _editLevel++; - - if (_editLevel == 1) - { - changes = _changesPreview.HasObservers ? _readerWriter.WriteWithPreview(updateAction, InvokeNextPreview) : _readerWriter.Write(updateAction); - } - else + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); + try { - _readerWriter.WriteNested(updateAction); - } - - _editLevel--; + try + { + if (_editLevel == 1) + { + changes = _changesPreview.HasObservers ? _readerWriter.WriteWithPreview(updateAction, InvokeNextPreview) : _readerWriter.Write(updateAction); + } + else + { + _readerWriter.WriteNested(updateAction); + } + } + finally + { + _editLevel--; + } - if (changes is not null && _editLevel == 0) + if (changes is not null && (_editLevel is 0)) + { + InvokeNext(changes); + } + } + finally { - InvokeNext(changes); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } } @@ -156,6 +169,36 @@ public IObservable> Preview(Func? predicate = null) return observable; } + private IObservable> CreateConnectObservable(Func? predicate) + { + var observable = Observable.Create>( + observer => + { + lock (_locker) + { + if (_readerWriter.Items.Length > 0) + { + observer.OnNext( + new ChangeSet + { + new(ListChangeReason.AddRange, _readerWriter.Items, 0) + }); + } + + var source = _changes.Finally(observer.OnCompleted); + + return source.SubscribeSafe(observer); + } + }); + + if (predicate is not null) + { + observable = new FilterStatic(observable, predicate).Run(); + } + + return observable; + } + private void InvokeNext(IChangeSet changes) { if (changes.Count == 0) @@ -195,6 +238,8 @@ private void OnCompleted() { _changesPreview.OnCompleted(); _changes.OnCompleted(); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnCompleted(); } } @@ -204,6 +249,8 @@ private void OnError(Exception exception) { _changesPreview.OnError(exception); _changes.OnError(exception); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnError(exception); } } }