Reactive: Sample never completes if the sampler sequence completes before the sampled sequence

[glopesdev]

Is this expected behavior? It looks to me that if sampler terminates, then there is a guarantee that there will be no more values coming out of Sample, and that it could be safely terminated. I was surprised to find that it never terminates. Is there some problematic scenario where the current behavior would be useful?

To reproduce:

c# var data = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(0.5)).Take(10); var sampler = Observable.Timer(TimeSpan.FromSeconds(1)); var result = data.Sample(sampler).Materialize(); result.Subscribe(Console.WriteLine); Console.ReadLine();

[shiftkey]

I think this is due to the data observable never completing, and is by design. Here's a snapshot of how it looks as an Rx marble diagram:

Note where the sequence terminates...

[glopesdev]

Actually this happens independently of whether or not the sampled sequence terminates (I've updated my example with Take).

If this is by design, then my next question is: why? Is there any situation in which it would make sense to have this "by design"? I don't like the idea of operators that promote dangling sequences, and in general Rx seems to be quite conservative in this respect. Most operators terminate as soon as they can reasonably be allowed to terminate in order to make sure resource allocation side-effects are disposed of.

That's very reasonable and also particularly important when mixing observable streams with operators that convert observables from infinite/finite and hot/cold such as Window or Replay. I don't want race conditions changing the behavior of my slices where some processing terminates and others don't because the last sampling tick came at the wrong time...

[shiftkey]

Just pointed two observers at each sequence (this is on Rx 2.2.5 from NuGet) and the sampler's subscription doesn't raise OnCompleted. Interesting.

Source:

static void Main(string[] args)
{
    var data = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(0.5)).Take(10);
    data.Subscribe(
        next => Console.WriteLine("Data - Next {0}", next),
        () => Console.WriteLine("Data - Completed"));

    var sampler = Observable.Timer(TimeSpan.FromSeconds(1));
    data.Sample(sampler).Subscribe(
        next => Console.WriteLine("Sampler - Next {0}", next),
        () => Console.WriteLine("Sampler - Completed"));

    Console.ReadLine();
}

Result:

Data - Next 0
Data - Next 1
Data - Next 2
Sampler - Next 2
Data - Next 3
Data - Next 4
Data - Next 5
Data - Next 6
Data - Next 7
Data - Next 8
Data - Next 9
Data - Completed

EDIT: clarified sample

[glopesdev]

You mean the result from Sample doesn't raise OnCompleted, right? The sampler sequence itself should complete.

Here's my own print out of all three sequences, which agrees with yours if you replace Sampler by Result.

[shiftkey]

@glopesdev I edited my previous comment to show the full sample (including subscriptions)

[glopesdev]

@shiftkey great, we're in agreement then, although I would rename the second WriteLine to print "Sample" instead of "Sampler", since it is effectively printing out the result from the Sample operator.

I have just run the sample against the latest head and the basic point of the issue remains: Sample never completes if the sampler sequence terminates first.

[RxDave]

Wow I wouldn't have expected that. Definitely a bug.

[RxDave]

To be clear, I still think that OnCompleted shouldn't occur exactly when the sampler calls OnCompleted, but when all sequences have OnCompleted.

OnCompleted carries information. In this case, it should, assuming the bug has been fixed, tell observers when the source and the sampler have both OnCompleted. The reason is simply that this overload samples the notifications of a given observable based on the timing of another. It would actually seem strange to me for the sampler to say when to sample the latest OnNext from the source, but when OnCompleted arrives, instead of sampling OnCompleted from the source (which basically just means to wait for the source to complete naturally), do what you're assuming and simply terminate the output immediately, ignoring the source's actual OnCompleted. I can understand your point of view, that it seems wasteful to have a bunch of silence between the sampler OnCompleted and the (eventual, perhaps) OnCompleted of the source, but I think that's just an inconsistent way of looking at this operator's semantics. I.e., I see it as the sampler says when to sample the source, if it goes away then you just won't receive any more sampled notifications, though you just _may_ eventually get the source's OnCompleted, if it ever has one.

On the flip side, if the sampler never completes then I'd take that to mean that you're not done sampling yet -- regardless of whether the source has OnCompleted. The fact that there's nothing left to sample is irrelevant. You'll get the OnCompleted after both inputs have OnCompleted.

Note that if you need the OnCompleted of either of them independently, not the conjunction, then you can easily get them outside of the Sample method and combine them back into the query; e.g., TakeUntil.

[RxDave]

c# sampler.Publish(s => xs.Sample(s).TakeUntil(s.IgnoreElements()));

[glopesdev]

@RxDave thanks for the clarification, I would be happy with the semantics you described.

Can you give some context to the code sample you posted? Initially I thought it would be some kind of workaround for having Sample terminate when the sampler completes, but then I guess you would need to use TakeUntil(s.TakeLast(1)), so maybe you had something else in mind?

[RxDave]

My mistake, thanks for the correction. I've added IgnoreElements to fix the problem.

[glopesdev]

@RxDave I don't think IgnoreElements is enough because TakeUntil ignores completion messages for the purposes of terminating the sequence (i.e. only OnNext calls are considered). I've had this problem before and I converged on using TakeLast(1) because at least it ensures that the termination token is sent out only when the sequence completes. It doesn't work for empty sequences though, which is annoying...

[RxDave]

Hmm, you're right. I've thought for the longest time that TakeUntil reacts to OnCompleted as well. I wonder if they changed it at some point and I just never noticed. That's good to know.

[RxDave]

Here's an alternative:
sampler.Publish(s => xs.Sample(s).TakeUntil(s.TakeLast(1).DefaultIfEmpty()));

[RxDave]

Yeah, I think I'll use DefaultIfEmpty from now on. It has the nice additional effect of ensuring that OnCompleted -> OnNext, if for any reason the source is an empty sequence.

[RxDave]

Oops, forgot TakeLast(1) again. Corrected.

[glopesdev]

@RxDave cool, I never noticed DefaultIfEmpty. However, it looks like you still need to couple it with a TakeLast(1), because if the sequence is non-empty it just starts streaming all the elements from the beginning, which is not what you want if you want to "TakeUntil the end".

However, after checking just now I think I found an even better alternative: LastOrDefaultAsync returns the last element of an observable sequence or a default value if no such element exists. I think I'll use this one from now on, nice!

[RxDave]

Oooh, LastOrDefaultAsync, that's it.

[shiftkey]

@glopesdev @RxDave is this something we should actually fix?

[RxDave]

Hmm, it's a breaking change and there's a reasonable workaround. Perhaps it's not worth fixing at this point?

[glopesdev]

@shiftkey Yes, as it was clearly identified as being a bug. I recommend adopting the behavior suggested by @RxDave above. I can make a pull request with unit tests, but maybe it's best for @bartdesmet or @mattpodwysocki to weigh in?

[glopesdev]

@RxDave You're right about that, but the new major release 3.0.0 would be a good opportunity to fix this, especially as Rx.NET is expanding it's reach to even more platforms. You'll just be propagating the bug further and further into the future, which will only make the problem worse...

[RxDave]

@glopesdev Makes sense to me. (To be clear, I'm agreeing with your latest point :)

I think what we really need is an additional sampling/throttling library that rounds out the missing behaviors in a set of well-designed, generalized operators. I mean, how many people have written their own "responsive" sampling operator? I.e., Take the first, then sample when there's chattiness and reset when there's silence. There are like dozens out there.

[glopesdev]

@RxDave guilty as charged...
source.Window(sampler).SelectMany(xs => xs.Take(1))

[danielcweber]

@glopesdev @RxDave @ghuntley
Should we fix it now and go for the breaking change? IMO the described behaviour is clearly unexpected, so the change might not be that breaking after all.

[glopesdev]

@danielcweber 4.0 is technically a major release, so we should definitely use it for behaviour changes of this kind. In fact, this is actually just a bug fix according to semantic versioning, since no APIs are actually breaking. As you say, there was a failure to comply with the documented behaviour of the operator, as also acknowledged above.

[clairernovotny]

I think we can take this fix for 4.0. Is there a PR for this against master?

[danielcweber]

274 deals with Sample but I think it actually implements a different behaviour... @glopesdev I'll review that, can you join?

[glopesdev]

@danielcweber sounds good, I don't mind contributing some tests if necessary

[danielcweber]

274 deals with periodic sampling by a timer. There may be a bug there as well but this issue deals with the scenario where you have a sampling sequence. @glopesdev A failing test for the desired behaviour would be awesome. I can implement a fix then.

[danielcweber]

@onovotny @glopesdev https://github.com/dotnet/reactive/pull/520