[Bug]: ExpireAfter holding items after dispose

[aguahombre]

Describe the bug 🐞

If you dispose a observable using ExpireAfter, the items are not available for GC until they expire.
The same occurs using ToObservableChangeSet with expireAfter.
If the items had very long expiry times e.g. days, this could lead to high memory usage.

Step to reproduce

1. Run this ConsoleApp example and press enter. The observable is disposed, but the items are not GCed until they expire.
   Two alternative variants are commented out, both exhibit the problem.

using DynamicData;
using System.Reactive.Concurrency;
using System.Reactive.Linq;

namespace ExpireAfterLeak;

internal class Program
{
    static void Main(string[] args)
    {
        var stopwatch = new System.Diagnostics.Stopwatch();
        stopwatch.Start();

        Console.WriteLine("Started");
        var subscription = Observable.Interval(TimeSpan.FromMilliseconds(250))
            .Select(x => new Item(x % 100))
            .Do(x => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Creating Item with value {x.Value}"))
            .Take(5)

            //.Do(x => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {stopwatch.ElapsedMilliseconds:D6} {string.Join('\n', x)}"), ex => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {ex}"), () => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Source Completed"))
            //.ToObservableChangeSet(x => x.Value, expireAfter: _ => TimeSpan.FromSeconds(10), scheduler: NewThreadScheduler.Default)

            //.ToObservableChangeSet(x => x.Value, scheduler: NewThreadScheduler.Default)
            //.Do(x => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {string.Join('\n', x)}"), ex => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {ex}"), () => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Source Completed"))
            //.ExpireAfter(_ => TimeSpan.FromSeconds(10), pollingInterval: TimeSpan.FromSeconds(5), scheduler: NewThreadScheduler.Default)

            .ToObservableChangeSet(x => x.Value, scheduler: NewThreadScheduler.Default)
            .Do(x => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {string.Join('\n', x)}"), ex => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {ex}"), () => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Source Completed"))
            .ExpireAfter(_ => TimeSpan.FromSeconds(10), scheduler: NewThreadScheduler.Default)

            .Subscribe(x => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {string.Join('\n', x)}"), ex => Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} {ex}"), () =>
            {
                Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Subscribe Completed");
                GC.Collect();
            });

        Console.ReadLine();
        Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Dispose");
        subscription.Dispose();

        while (!Console.KeyAvailable)
        {
            GC.Collect();
            GC.WaitForPendingFinalizers();
            GC.Collect();
            Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6}  GC.Collect");
            Thread.Sleep(1000);
        }
        Console.WriteLine($"{stopwatch.ElapsedMilliseconds:D6} Exit");
    }
}

internal class Item
{
    public long Value { get; }

    public Item(long v) => Value = v;

    ~Item()
    {
        Console.WriteLine($"Item with value {Value} is being finalized.");
    }
}

Results showing items being held until they expire

Started
000296 Creating Item with value 0
000305 Add, Key: 0, Current: ExpireAfterLeak.Item, Previous: <None>
000327 Add, Key: 0, Current: ExpireAfterLeak.Item, Previous: <None>
000548 Creating Item with value 1
000548 Add, Key: 1, Current: ExpireAfterLeak.Item, Previous: <None>
000549 Add, Key: 1, Current: ExpireAfterLeak.Item, Previous: <None>
000799 Creating Item with value 2
000800 Add, Key: 2, Current: ExpireAfterLeak.Item, Previous: <None>
000800 Add, Key: 2, Current: ExpireAfterLeak.Item, Previous: <None>
001049 Creating Item with value 3
001050 Add, Key: 3, Current: ExpireAfterLeak.Item, Previous: <None>
001050 Add, Key: 3, Current: ExpireAfterLeak.Item, Previous: <None>
001301 Creating Item with value 4
001301 Add, Key: 4, Current: ExpireAfterLeak.Item, Previous: <None>
001301 Add, Key: 4, Current: ExpireAfterLeak.Item, Previous: <None>
001302 Source Completed

001619 Dispose
001625  GC.Collect
002637  GC.Collect
003650  GC.Collect
004662  GC.Collect
005673  GC.Collect
006679  GC.Collect
007687  GC.Collect
008702  GC.Collect
009707  GC.Collect
Item with value 2 is being finalized.
Item with value 1 is being finalized.
Item with value 0 is being finalized.
Item with value 4 is being finalized.
Item with value 3 is being finalized.
010720  GC.Collect
011733  GC.Collect
012744  GC.Collect

Reproduction repository

https://github.com/aguahombre/ExpireAfterLeak

Expected behavior

When the observable subscription is disposed, all unexpired items should be released for GC and the timer thread terminated.

Screenshots 🖼️

No response

IDE

No response

Operating system

Windows 10

Version

No response

Device

PC

DynamicData Version

9.4.1

Additional information ℹ️

No response

[JakenVeina]

Neat. Thanks for the detailed breakdown.

Ultimately, this comes down to the RX Scheduler implementations. The .ExpireAfter() operators are making a proper effort to do cleanup during disposal of a subscription, by asking the scheduler to cancel the next scheduled expiration, if there is one. However, the scheduler ignores this request, keeps the entire subscription alive until the cancelled operation reaches its due time, at which point the scheduler will recognize that it was cancelled and just throw it away, instead of invoking it.

To be fair, the IScheduler documentation specifies that cancellation is only a "best effort" but all the most-common implementations, like DefaultScheduler and NewThreadScheduler, make absolutely NO effort. Calling .Dispose() on a scheduled operation that hasn't yet started is literally a NOP. Once the operation starts, it becomes a simple proxy for .Dispose() on the IDisposable that the schedule action itself returns.

Basically, if there's any possibility of you churning through (I.E. scheduling, and later cancelling before they finish) a lot of long-running operations, you need a specialized scheduler that will prune its queue to avoid memory leaks. I'll probably dig through the RX repository some more to see if I can glean some more wisdom of why it's setup this way, or if perhaps queue pruning could be added.

All that being said, we can work around this in DynamicData by adding a Weak or breakable reference between the state object that the scheduler captures and the subscription state of the operator. I was able to demonstrate that just clearing the subscription caches upon disposal eliminates the memory leaking of the items. We could just leave it at that, but for very-large collections, even a now-empty cache (internally a Dictionary<TKey, TValue>) can still consume a big chunk of memory, so I'd like to take the extra step and break references to the entire subscription object.

[aguahombre]

The hardest problems are always in the third party code.
Thanks for investigating this, hopefully there is a good solution.

For the moment I have used a short poll interval on ExpireAfter to ensure the items are disposed fairly quickly

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.