En décembre dernier, un nouveau concept a été introduit dans les Reactive Extensions. Ce concept est représenté par la nouvelle interface IScheduler.
Une des problématiques récurrentes avec la programmation asynchrone est le fait que certain code ne puisse s'executer que sur certains threads. Par exemple, avec WinForms ou Silverlight, seul le thread de l'UI peut mettre à jour l'UI. Si vous essayez de changer la couleur d'un Label depuis un thread du thread pool, vous obtiendrez une exception.
Dans un premier temps, les Reactive Extensions on résolues ce problème en utilisant un context global, représenté par une variable statique. Lorsque vous utilisiez une méthode du type Interval, les callbacks (appels a OnNext, OnError et OnCompleted) étaient appelés en utilisant un thread lié au context global. Cette approche n'est cependant pas très élégante, souvent peu pratique, et surtout peu composable.
Une des caractèristiques clés de Rx étant la composabilité, les créateurs de Rx ont complètement replacé cette approche dans une des dernières releases de la librairie.
La création de concurrence passe maintenant par des schedulers, implémentant l'interface IScheduler. Voici a quoi elle ressemble :
public interface IScheduler
{
IDisposable Schedule(Action action);
IDisposable Schedule(Action action, TimeSpan dueTime);
}
IScheduler represente un mechanisme pouvant executer du code. Il est possible de plannifier l'execution de code après un certain délai (méthode prenant un argument TimeSpan), ou bien de plannifier l'execution de code "dès que possible" (sans garantie qu'il s'execute immédiatement).
Dans Rx, on peut distinguer deux types d'opérateurs :
- Ceux qui relaient un callback, comme Select et Where. Ils observent la source, et lorsqu'une notification arrive, ils en produisent une autre sur le même thread. De la même façcon, Merge ou ForkJoin utilisent le thread d'une des sources pour produire une notification.
FromEvent est dans la même catégorie parcequ'il utilise le thread sur lequel l'évènement est levé pour produire la notification. Ce thread peut venir de l'UI dans le cas de Button.Click, ou du thread pool dans le cas de WebClient.DownloadStringCompleted, ou d'autres sources encore selon l'implémentation. L'opérateur FromEvent réutilise ce thread et n'en crée pas d'autre.
Ces opérateurs ne créent donc pas de concurrence, ils réutilisent les threads qui leur envoie la notification.
- D'autres opérateurs créent de la concurrence : Timer et Interval en sont des exemples. Return en est aussi un, même si la notification est censée arriver instantanément.
Tous les opérateurs créant de la concurrence prennent un IScheduler en paramètre, indiquant comment cette concurrence doit être créée.
Dans sa version actuelle, Rx fournit trois schedulers par défaut : Scheduler.Now, Scheduler.Later, et Scheduler.Dispatcher.
- Scheduler.Later crée de la concurrence en prenant un thread dans le thread pool. Dans .NET 4.0, il est implémenté en utilisant la TPL. La méthode Schedule(Action, TimeSpan) est implémentée en utilisant System.Threading.Timer.
- Scheduler.Now est particulier car il utilise le thread courant. La méthode Schedule(Action, TimeSpan) est donc implémentée en utilisant Thread.Sleep.
- Scheduler.Dispatcher permet de garantir que les threads utilisés sont compatibles avec le dispatcher en cours. Cela permet de mettre à jour l'UI depuis le thread produisant la notification.
Scheduler.Default pointe sur Scheduler.Later, sauf sur Silverlight ou il pointe sur Scheduler.Dispatcher.
Pour bien comprendre le fonctionnement des schedulers, il faut étudier quelques exemples. Utilisons Interval avec Scheduler.Later :
IScheduler scheduler = Scheduler.Later;
IObservable<long> interval = Observable.Interval(scheduler, TimeSpan.FromSeconds(1));
Console.WriteLine("Start Subscribe");
interval.Subscribe(Console.WriteLine);
Console.WriteLine("End Subscribe");
Le resultat sur la console est le suivant :
Start Subscribe
End Subscribe
0
1
2
...
Par contre, si on utilise Scheduler.Now a la place, on obtient :
Start Subscribe
0
1
2
...
Donc dans le cas de Scheduler.Later, l'appel a Subscribe plannifie les notifications en utilisant le thread pool, Subscribe n'est donc pas bloquant.
Au contraire, avec Scheduler.Now, les notifications sont plannifiées sur le thread en cours. Le thread est en someil pendant une seconde entre chaque notification. Subscribe est donc bloquant dans ce cas, et "End Subscribe" n'est jamais atteint.
Nous verrons dans une prochaine partie comment implémenter un opérateur en utilisant un scheduler.
