WhenAnyValue might not send latest value

[erikm-g]

Describe the bug
If a subscription is started for changes on a property using WhenAnyValue and the property is updated on another thread at approximately the same time, WhenAnyValue might send the old value as the initial value in the observable sequence and not send the updated value at all.

Steps To Reproduce
Let one thread update a property and set up a subscription on changes for it from another. This unit test illustrates the problem. It passes most of the times, but if repeated often enough it will eventually fail. The value of anyReactiveObject.Value will be 2 but no call to the subscription is emitted.
´´´
private class SomeReactiveObject : ReactiveObject
{
private int value;

    public int Value
    {
        get => value;
        set => this.RaiseAndSetIfChanged(ref this.value, value);
    }
}

[Test]
[Repeat(10000)]
public async Task WhenAnyShouldBeThreadSafe()
{
    var anyReactiveObject = new SomeReactiveObject { Value = 1 };
    var task = Task.Run(() => anyReactiveObject.Value = 2);
    bool found = false;
    anyReactiveObject.WhenAnyValue(x => x.Value)
        .Where(v => v == 2).Subscribe(_ => found = true);
    await task;
    Assert.That(found);
}

´´´
Expected behaviour
The test above should pass.
In a multithreaded environment, events may occur out of order, or if updates are very frequent some intermediate values might be missed, but the latest value should always be emitted to the pipeline.

Environment

• OS: Windows 10
• Device: PC
• ReactiveUI Version: 16.2.6

[kmgallahan]

Isn't this just a standard race condition?

Both threads are running in parallel and nothing is stopping the subscription from being setup after the Task has completed.

WhenAnyValue might send the old value as the initial value in the observable sequence

No initial value is "sent", you just Assert that found = false at the end, which is what you initialize it to. The subscription is never called if it is setup after the value is changed.

[kmgallahan]

@RolandPheasant I wouldn't mark this as the answer. My comment:

The subscription is never called if it is setup after the value is changed.

Is false and was based on my misunderstanding of WhenAnyValue at the time - I was unaware it emits this initial value (except when observing a property chain that throws a null exception).

Disregarding the fact that bool isn't thread safe and therefore is in an indeterministic state for the assertion (I think?), a more accurate answer is:

WhenAnyValue is not guaranteed to emit the latest value of a property if setup in a multithreaded environment.

[glennawatson]

Side note Bool are atomic read, writes but not read and write at the same time.

They are thread safe for read or write operations but most people want atomic combined read and write.

I've used the interlocked.exchange with a numeric to achieve a gate in past iterations.

[erikm-g]

If the task completes before the subscription is setup, WhenAnyValue emits 2 and the test passes.
If the subscription is setup before the task starts to execute WhenAnyValue will emit first 1 and then 2 and the test passes.
If the task is executed at the same time as the subscription is setup, sometimes WhenAnyValue emits only the value 1 but the property's value is 2. This causes the test to fail.

[erikm-g]

If I would modify the test and add .Do(TestContext.Write) before the .Where, I would expect the test to sometimes print 12, sometimes print just 2 and in a few occasions print 21, but in all cases I would expect the test to pass and never to print just 1 and fail. There is no race condition between assigning the values 1 and 2.

[glennawatson]

There's a race condition in your Code. The task will start immediately which may or may not be before the WhenAnyValue is registered.

Try converting your task into a Func and making it into a function you call after the WhenAnyValue is registered and then await that.

[kmgallahan]

@glennawatson I tested this code:

public MainWindowViewModel()
{
    var i = 1000;
    do
    {
        i--;
        WhenAnyShouldBeThreadSafe();
    } while (i > 0);
}

public void WhenAnyShouldBeThreadSafe()
{
    var anyReactiveObject = new SomeReactiveObject { Value = 1 };
    Task.Run(() => anyReactiveObject.Value = 2);
    anyReactiveObject.WhenAnyValue(x => x.Value)
        .Subscribe(x =>
        {
            if (anyReactiveObject.Value is 1) Debug.WriteLine(anyReactiveObject.Value);
        });
    }
}

public class SomeReactiveObject : ReactiveObject
{
    public int Value
    {
        get => value;
        set => this.RaiseAndSetIfChanged(ref this.value, value);
    }

    private int value;
}

In theory 1 should never be written to debug out because the subscription is always made after var anyReactiveObject = new SomeReactiveObject { Value = 1 };. However, in the 1000 runs it does show up multiple times for me...

[glennawatson]

Task.Run(() => anyReactiveObject.Value = 2); // This can run or not run before the WhenAnyValue happens, use a Func<Task>
anyReactiveObject.WhenAnyValue(x => x.Value)
.Subscribe(x =>
{
if (anyReactiveObject.Value is 1) Debug.WriteLine(anyReactiveObject.Value);
});

[kmgallahan]

@glennawatson Without Task.Run(() => anyReactiveObject.Value = 2); the observable / subscription should never fire because the subscription is made after the value is assigned here var anyReactiveObject = new SomeReactiveObject { Value = 1 };, correct?

That would mean Value = 1 should never show up in the observable / subscription (be sent to debug out), or am I missing something?

[glennawatson]

Func<Task> myFancyValueChanger = () => Task.Run(() => ....

something.WhenAnyValue(x => ....
await myFancyValueChanger.Invoke();

[kmgallahan]

@glennawatson

Ignoring all of the code above - from my understanding, just these 2 lines of code should never fire the subscription, correct?

anyReactiveObject.Value = 1;
anyReactiveObject.WhenAnyValue(x => x.Value).Subscribe(x => Debug.WriteLine(anyReactiveObject.Value));

So value = 1 should never show up in the debug output.

This code however will very seldomly show 1 in the debug output:

anyReactiveObject.Value = 1;
Task.Run(() => anyReactiveObject.Value = 2);
anyReactiveObject.WhenAnyValue(x => x.Value).Subscribe(x => Debug.WriteLine(anyReactiveObject.Value));

I can't reason as to why 1 should ever show up, but it does.

[glennawatson]

You don't control when Task.Run runs, it's up to the task scheduler

[kmgallahan]

@glennawatson

I understand there is a race condition happening.

In this race condition there should be two scenarios:

1. Value is set to 2 AFTER the subscription is setup, so the subscription is called printing 2 to the output
2. Value is set to 2 BEFORE the subscription is setup, so nothing is printed to the output

There should be no scenario 3:

3. Value is set to 2 AFTER the subscription is setup, the subscription prints 1 to the output

But that 3rd scenario does indeed occur 5-10 times out of 1000.

@RLittlesII Can you sanity check me here?

[glennawatson]

WhenAnyValue always emits the current initial value of the Property unless you do a Skip(1)

[glennawatson]

Scenario 3 can happen

1. Initial Value is 1
2. Task sets value to 2
3. WhenAnyValue grabs initial value of property and emits 2

[RLittlesII]

@kmgallahan

There should be no scenario 3:

Value is set to 2 AFTER the subscription is setup, the subscription prints 1 to the output
But that 3rd scenario does indeed occur 5-10 times out of 1000.

The fact you are using Task.Run by definition means you are not in control of when it is set. Asynchronous code does not happen where you write the line of code. It happens at some point in the future.

[kmgallahan]

Sorry guys, didn't know it emitted the initial value on setup, please forget I ever existed, lol

I checked the docs here when I first saw this issue and it only mentions things like this for a while:

This returns an observable that yields the current value of Foo each time it changes:

You can Subscribe to the observable returned by the WhenAny variant and get notifications whenever the value changes.

Funny thing is I've been using it for months now...

[erikm-g]

I know how await and Task.Run works. When await task returns I know that the task has completed, that is what is important here. I want the task and setting up the subscription to be run in parallel in this test since that is what is triggering the error. In my real application it is a completely different part of the code running on a different thread updating the value. This is just a very simple example of reproducing the behavior of a multi threaded application where my WhenAnyValue subscriptions does not trigger if the value is updated at the same time as the subscription is set up.

[kmgallahan]

I see your problem now. The subscription is run on the thread that changed the value, not the one where it was setup.

When the Task runs on a different thread then the bool is changed on that thread, and bool is inherently not thread safe, so back on the original thread it may or may not Assert to true at the end.

If you make sure to change the bool on the original thread in the subscription then it should work as you expect.

[erikm-g]

That is not the issue. The problem is that the value 2 is never emitted. I can change the bool to an int, increment it with Interlocked.Increment in the subscription and add a memory barrier before my assert, it is still the same error. The lambda passed to subscribe is simply never executed.

[kmgallahan]

The problem is that the value 2 is never emitted.
The lambda passed to subscribe is simply never executed.

I have no idea what you are talking about now. If the lamba didn't execute then why does this print to debug output?

anyReactiveObject.WhenAnyValue(x => x.Value).Where(v => v == 2).Subscribe(_ => Debug.WriteLine("Hello World!"));

[RLittlesII]

@erikm-g Have you considered using Synchronize to ensure that the observables from multiple threads are received correctly?

[kmgallahan]

You may think there are cases where the lamba isn't being executed because the Assert shows found is false.

Each one of those cases is because bool isn't thread safe. Yes, when you Assert.That(found); the task has completed and the subscription was executed, but if the bool was set to true in the lambda on a different thread then it may or may not be true on the original thread at the time of the assertion - it isn't thread safe so this isn't a deterministic check.

[kmgallahan]

Also, in your original codebase initial values won't be emitted (the lamba won't be executed) if you use WhenAnyValue(x.y.z).Subscribe and x or y is null at the time of the subscription. That may be throwing you off as well.

[glennawatson]

We will be locking this thread soon but welcome to continue a help topic in the discussion area. Just due to the mass emails it produces

[erikm-g]

@RLittlesII I am not sure how Synchronize would help. There are no OnError or OnCompleted events and there is no issue if 2 OnNext are called at the same time. I tried adding a call to Synchronize directly after WhenAnyValue in my test and it did no difference.

@kmgallahan The test works about 299 times of 300, in those cases you will get your print out. If you check the 1 time it doesn't work it is not printed.

I think the issue is that WhenAnyValue internally gets the initial value first and starts to listen on changes after that, so any changes made in between are missed. It needs to be done the other way around to be thread safe, and if you want to avoid sending the initial value twice some check for that need to be added.

If I run this modified test where the initial value from WhenAnyValue is skipped and the current value is emitted after the subscription to changes is set up, the test succeeds every time.

        var anyReactiveObject = new SomeReactiveObject { Value = 1 };
        var task = Task.Run(() => anyReactiveObject.Value = 2);
        bool found = false;
        var subject = new Subject<int>();
        subject.Where(v => v == 2).Subscribe(_ => found = true);
        anyReactiveObject.WhenAnyValue(x => x.Value)
            .Skip(1).Subscribe(v => subject.OnNext(v));
        subject.OnNext(anyReactiveObject.Value);
        await task;
        Assert.That(found);

[kmgallahan]

@glennawatson @RLittlesII

According to the ConcurrentBag docs:

All public and protected members of ConcurrentBag are thread-safe and may be used concurrently from multiple threads.

So this code should never fail, but it does sometimes:

var bag = new ConcurrentBag<int>();
var anyReactiveObject = new SomeReactiveObject();
var task = Task.Run(() => anyReactiveObject.Value = 2);

anyReactiveObject.WhenAnyValue(x => x.Value).Subscribe(x =>
{
    if (x == 2) bag.Add(2);
});

await task;

Debug.Assert(!bag.IsEmpty);

Either:

1. The task completes before the subscription, so a 2 should be added to the bag after WhenAnyValue.Subscribe since it is the initial value at that time
2. The task completes after the subscription, so the subscription should observe the property change notification and a 2 should be added to the bag

The bag should never be empty for the assertion, yet it sometimes is.

There might be a race condition inside SubscribeToExpressionChain:

ReactiveUI/src/ReactiveUI/Mixins/ReactiveNotifyPropertyChangedMixin.cs

Line 131 in 8628353

public static IObservable<IObservedChange<TSender, TValue>> SubscribeToExpressionChain<TSender, TValue>(

Where the sequence goes:

1. Create a ReactiveObject with a property that calls RaiseAndSetIfChanged on set
2. Call Task.Run to change the property on thread B
3. Call WhenAnyValue.Subscribe on the ReactiveObject on thread A
4. Create an IObservable named notifier loaded with an ObservedChange that contains the initial value of the ReactiveObject here:

ReactiveUI/src/ReactiveUI/Mixins/ReactiveNotifyPropertyChangedMixin.cs

Line 138 in 8628353

IObservable<IObservedChange<object?, object?>> notifier =

5. Thread B changes the property value on the ReactiveObject, calling RaiseAndSetIfChanged which does nothing
6. Subscribe to notifications on the property here (I think?):

ReactiveUI/src/ReactiveUI/Mixins/ReactiveNotifyPropertyChangedMixin.cs

Line 142 in 8628353

notifier = chain.Aggregate(notifier, (n, expr) => n

7. Return the IObservable containing the ObservedChange which still has Value = to the initial/unchanged value of the property as set here:

ReactiveUI/src/ReactiveUI/Interfaces/ObservedChange.cs

Line 37 in 8628353

public TValue Value { get; }

In this scenario a subscription to WhenAnyValue will only emit the initial unchanged value and not the second value.

This might be resolved if property change notifications are setup before an ObservedChange containing the initial value is created.

I'm not entirely sure if this covers all race conditions in the logic, but it may help.

[erikm-g]

I have read that article before and read it again. I don't see your point. I don't know the order things are happening, but don't see any order where just 1 is an expected output. If I use a BehaviorSubject with initial value 1, let the task call OnNext(2) and subscribing to the BehaviorSubject instead of using WhenAnyValue I never just 1. I always get at least a 2.

[erikm-g]

The task is actually a callback from a MQTT communication library setting the value of a property in the model if it is connected or not, the WhenAnyValue and subscription is made in a VM. I don't know how they should be linked in any other way than that one sets the property and the other is listening to changes of it.

[glennawatson]

Can you move the WhenAnyValue before any potential value being set. That's likely to eliminate your issue

[erikm-g]

No, in this case I can't do that. In most cases they are set up in the constructors during program initialization and then I don't run into any issue, but in a few cases the subscription needs to be started later.

[glennawatson]

Getting late here so will have to look more tomorrow. Likely a fix could be provided once off in a ReactiveObject for your scenario with the property and the trigger for INotifyPropertyChanged being locked ()

[anaisbetts]

WhenAny is not "thread-safe", it has the same thread safety rules as any other C# object. You should not write to ViewModel properties on other threads, as they inherit the thread affinity of the Views they are bound to. If you wouldn't write from other threads to a View, you shouldn't do it to a ViewModel!

[erikm-g]

It is (common practice / not unheard of / valid)? to use WhenAnyValue to observe properties that can be modified by other threads, as subscriptions can be handled on the UI thread

I do not agree with this, if you're doing this, you are making a Mistake, full-stop.

If so, why is this code example:

this.WhenAnyValue(x => x.MyImportantProperty).ObserveOn(RxApp.MainThreadScheduler).Subscribe(x => ...);

listed on https://www.reactiveui.net/docs/handbook/scheduling/ ?

[glennawatson]

If you use it in the constructor you can handle properties with different threads being set.

[glennawatson]

But multithreaded property setting is not recommended. The documentation could use improvement.

[erikm-g]

That is my point. If this is not a bug that should be corrected, the documentation needs to be updated. I would of course have preferred if the code was updated to handle multithreading as I thought it would after reading the documentation, but I can understand if that is not feasible.

[anaisbetts]

If so, why is this code example listed on https://www.reactiveui.net/docs/handbook/scheduling/

That is a bad code example and we should for sure yank it, thanks for pointing this out @erikm-g