Pourquoi obtiens-je des résultats non déterministes avec Repeat()?

Question

Je cherche à approfondir mes connaissances sur Rx. Je m'amuse donc à manipuler des flux et à essayer de les faire se comporter comme je m'y attendrais.

Alors que j'ai lu précédemment que l'opérateur Repeat() posait des difficultés en pratique car vous pourriez perdre des notifications entre OnCompleted et la nouvelle souscription, je ne parviens pas à comprendre pourquoi cela se produit.

        var subject = new Subject();

        var my = subject
            .Take(1)
            .Merge(Observable.Empty().Delay(TimeSpan.FromMilliseconds(2000)))
            .Repeat();
        my.Subscribe(Console.WriteLine);

        var stopwatch = new Stopwatch();
        stopwatch.Start();
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(1), () => subject.OnNext("1 à " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(2), () => subject.OnNext("2 à " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(3), () => subject.OnNext("3 à " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(4), () => subject.OnNext("4 à " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(5), () => subject.OnNext("5 à " + stopwatch.ElapsedMilliseconds));
        Scheduler.ThreadPool.Schedule(TimeSpan.FromSeconds(6), () => subject.OnNext("6 à " + stopwatch.ElapsedMilliseconds));

        Console.ReadLine();

Lorsque j'exécute cet exemple, les résultats sont totalement non déterministes :

Résultat 1 :

1 à 1006
3 à 3007
5 à 4995

C'est bien qu'il ait omis le 2 et le 4 mais même à l'intérieur de ce résultat, il y a quelque chose d'étrange car en réalité il n'y a pas un écart de 2 secondes entre le 3 et le 5.

Cependant, les résultats peuvent être encore pires. Voyez celui-ci :

1 à 1003
2 à 2003
4 à 4005
6 à 6004

Il n'y a pas d'écart de 2 secondes entre 1 et 2. C'est exactement une seconde. Pourquoi ne l'a-t-il pas exclu ?

Si quelqu'un pouvait éclaircir les choses pour moi, je serais plus que ravi !

ÉDITER

Je viens de remarquer que c'est peut-être le Merge qui pose problème ici. Si je remanie ma requête pour utiliser Concat, les choses semblent se passer comme elles devraient :

        var my = subject
            .Take(1)
            .Concat(Observable.Empty().Delay(TimeSpan.FromMilliseconds(2000)))
            .Repeat();

Answer 1

Windows (et d'autres systèmes d'exploitation de bureau) n'est pas un système d'exploitation en temps d'exécution et donc vous ne pouvez pas vous fier à ce que ses minuteries soient précises jusqu'aux millisecondes. Et surtout si vous avez plus de minuteries, cela peut entraîner un comportement non déterministe, ce qui est exactement votre cas.

Voici comment fonctionne votre séquence originale :

• temps ~ 0
  • Take(1) souscrit à sujet
  • la minuterie pour l'observable vide différé commence à fonctionner
• temps ~ 1
  • 1 est ajouté à sujet, 1 est écrit; après ce point, personne d'autre ne souscrit à sujet
• temps ~ 2
  • la minuterie pour l'observable vide différé se termine. En raison de cela, Take(1) souscrit à nouveau à sujet et une autre minuterie pour un observable vide différé démarre
  • environ au même moment, 2 est ajouté à sujet

En raison de légères différences de synchronisation, les deux actions au moment approximatif 2 peuvent se produire dans n'importe quel ordre. Et l'ordre est important, soit 2 est ajouté avant que Take() se réabonne, soit après. Ainsi, 2 peut être écrit, ou non.

Si ce que vous voulez est une séquence comme ceci :

1. attendre le premier élément et le retourner
2. attendre environ deux secondes
3. attendre le deuxième élément et le retourner (en ignorant ceux qui ont été ajoutés pendant l'attente de deux secondes)
4. …

alors je pense que le code dans votre édition est correct.

Mais cela ne garantit en aucun cas des résultats déterministes. Sur mon ordinateur, si je change le temps d'attente de l'observable vide différé à 1960 ms, j'obtiens des résultats non déterministes lors de l'utilisation de Concat().

Answer 2

Juste pour ajouter mon grain de sel en retard à la fête; si vous cherchez du déterminisme pour les tests, alors vous devriez utiliser le TestScheduler au lieu des planificateurs réels ThreadPool/TaskPool/NewThread. Purement pour les raisons que Svick souligne (la planification du système d'exploitation perturbe les choses).