Démon monotache "asynchrone"

Version imprimable

Bonjour,

j'aimerai savoir comment implémenter / réaliser un "démon" qui scrute un file d'attente (Queue) et traite les requêtes dans la file de manière asynchrone?

J'ai quelque chose du genre :
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 class CreateurDeTache1 { TacheDemon t = new TacheDemon(); // ... MonDemon.Enqueue(t); } class CreateurDeTache2 { TacheDemon t = new TacheDemon(); // ... MonDemon.Enqueue(t); } class MonDemon { private static Queue<TacheDemon> _listeTaches = new Queue<TacheDemon>(); public static void Enqueue(TacheDemon tache) { _listeTaches.Enqueue(tache); } }
Dès que je rajoute une tâche, je récupère donc la main de suite, par contre, là ou je bloque c'est de créer un "démon" qui scrute cette liste de taches (qui peut être vide comme contenir 10 taches, a tout moment) et exécute une tache après l'autre et renvoie le résultat à l'appelant.

J'avais pensé passer un AsyncCallback en paramètre de ma méthode Enqueue qui serait appelée lorsque la requête serait traitée, mais ça demande de générer un IAsyncResult.

:?

Comment vous y prendriez-vous? Quelle est la meilleure manière de procéder?

:merci:

Il y a une façon simple de le faire mais qui limite un peu le champs de ta demande parce qu'il manque des détails sur ton architecture (est-ce que ton démon est un service windows/wcf/une console ?)

Est-ce que tu entends par "appelant" soit un thread secondaire ou une méthode de classe ?

Si c'est une méthode de classe il y a probablement plusieurs solutions plus simple, cela dépend si tu peux étendre ou pas tes classes tacheDemon pour y ajouter par exemple une méthode à exécuter (on peut imaginer utiliser la reflexion par exemple ou les delegate/event)

Pour scruter la queue tu peux créer un thread/tâche/pool. Pour faire rapide et simpliste, et avoir une vue en gros, voici un code avec un thread (remarque que j'ai ajouté des lock sur la queue)
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 class CreateurDeTache1 { TacheDemon t = new TacheDemon(myCallBack); // ... MonDemon.Enqueue(t); public void MyCallBack() {} } class CreateurDeTache2 { TacheDemon t = new TacheDemon(MyCallBack); // ... MonDemon.Enqueue(t); public void MyCallBack() {} } class MonDemon { private static Queue<TacheDemon> _listeTaches = new Queue<TacheDemon>(); private Thread Scruteur ; public static void Enqueue(TacheDemon tache) { lock(_listeTaches) { _listeTaches.Enqueue(tache); } } //implémenter aussi une méthode stop/pause/ public void StartScrutation() { //verifier avant qu'il n'est pas déjà en cours d'execution Scruteur = new Thread(Scrute); Scruteur.Start(); } public void Scrute() { while(ConditionForQuitScrute) { lock(_listeTaches) { foreach(TacheDemon tachedemon in _listeTache) { tachedemon.Execute(); //par reflexion executer la méthode de la classe défini dans la propriété tachedemon.callback ou appeler le delegate } } } } }

'Soir,

je voulais justement éviter de faire un thread scruteur qui tourne en tache de fond, mais il me semble que je n'ai pas trop le choix de faire ça autrement que de faire de la sorte.

Mon projet est une DLL (un framework interne).

Merci pour les locks, je les avais déjà rajoutés dans mon code mais oublié de réécrire ici ;)

Je vais laisser le sujet ouvert pour voir si d'autres personnes auraient une autre manière de faire.

Cependant pour AsyncCallback, y aurait-il un exemple d'implémentation quelque part ? Ce qui me chiffonne c'est l'initialisation de IAsyncResult car de la même manière que pas mal de classe, j'aurai aimé faire : BeginEnqueue / EndQueue.

:merci:

[EDIT] J'ai oublié de le stipuler mais les créateurs de tâches sont des méthodes statiques...
Code:

1 2 3 4 5 6 7 8 9 10 11 static class CreateurDeTache1 { static int Execute(string task) { TacheDemon t = new TacheDemon(myCallBack); // ... MonDemon.Enqueue(t); } public void MyCallBack() {} }
du coup va falloir que je trouve un moyen de bloquer la aussi le retour de sorte de ne le renvoyer qu'une fois la tache exécutée.

Avec un semaphore voila ce que j'ia pu faire rapidement:

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
using System;
using System.Collections.Generic;
using System.Threading;
 
namespace ConsoleApplication2
{
	class Program
	{
		static void Main(string[] args)
		{
			var repository = Repository<Func<String>>.Instance;
			var c1 = new Consumer(repository);
			c1.Start();
 
			while (true)
			{
				Console.ReadLine();
				Producer.Enqueue();
			}
		}
	}
 
	public class Consumer
	{
		private readonly Repository<Func<String>> _instance;
 
		public Consumer(Repository<Func<String>> instance)
		{
			_instance = instance;
		}
 
		public void Start()
		{
			var thread = new Thread(o =>
										{
											if (!(o is Repository<Func<String>>))
												throw new ArgumentException("I want a repository to watch!");
 
											var repository = o as Repository<Func<String>>;
 
											while (true)
											{
												repository.Handle.WaitOne();
												var task = repository.Dequeue();
												if (task != null)
													Console.WriteLine(task());
											}
										});
			thread.Start(_instance);
		}
 
 
 
	}
 
	public class Repository<T>
	{
		#region Singleton
 
		private readonly static object Lock = new object();
		private static Repository<T> _instance;
		public static Repository<T> Instance
		{
			get
			{
				lock (Lock)
				{
					return _instance ?? (_instance = new Repository<T>());
				}
			}
		}
 
		#endregion
 
		private readonly Semaphore _syncEvent;
		private readonly Queue<T> _queue;
 
		private Repository()
		{
			_queue = new Queue<T>();
			_syncEvent = new Semaphore(0, int.MaxValue);
		}
 
		public WaitHandle Handle { get { return _syncEvent; } }
 
		public void Queue(T task)
		{
			lock (_queue)
			{
				_queue.Enqueue(task);
				_syncEvent.Release();
			}
		}
 
		public T Dequeue()
		{
			lock (_queue)
			{
				return _queue.Dequeue();
			}
		}
	}
 
	public static class Producer
	{
		public static void Enqueue()
		{
			Func<String> myTask = () =>
										{
											Thread.Sleep(5000);
											return DateTime.Now.ToString();
										};
			var instance = Repository<Func<String>>.Instance;
 
			instance.Queue(myTask);
		}
	}
 
}

Citation:

Envoyé par Arnaud F.

je voulais justement éviter de faire un thread scruteur qui tourne en tache de fond, mais il me semble que je n'ai pas trop le choix de faire ça autrement que de faire de la sorte.

c'est exactement ce qui est fait au minimum par le thread d'un service windows started.

Citation:

Mon projet est une DLL (un framework interne)

Tu peux préciser les interfaces que tu as avec les projets consommateurs ? C'est utilisé en multiprocessus ou multithread ?

Citation:

Cependant pour AsyncCallback, y aurait-il un exemple d'implémentation quelque part ? Ce qui me chiffonne c'est l'initialisation de IAsyncResult car de la même manière que pas mal de classe, j'aurai aimé faire : BeginEnqueue / EndQueue.

Il y a des façons élégantes de faire, il existe sûrement des patterns, je ne comprends pas ce qui té chiffonne avec le ASyncCallBack tu peux développer avec du code ce qui bloque ?
Citation:
J'ai oublié de le stipuler mais les créateurs de tâches sont des méthodes statiques...
Code:

1 2 3 4 5 6 7 8 9 10 11 static class CreateurDeTache1 { static int Execute(string task) { TacheDemon t = new TacheDemon(myCallBack); // ... MonDemon.Enqueue(t); } public void MyCallBack() {} }
du coup va falloir que je trouve un moyen de bloquer la aussi le retour de sorte de ne le renvoyer qu'une fois la tache exécutée.
Ah ok, tu as oublié aussi de réécire le dequeue non ?

Une idée serait là aussi d'utiliser par exemple le pattern command. Ainsi tu ajoutes dans ta queue la tache avec une command que tu exécutes une fois que tu as finis d'exécuter la tâche en queue

13/12/2010, 22h56
Arnaud F.

Citation:

Envoyé par hegros

c'est exactement ce qui est fait au minimum par le thread d'un service windows started.

Ok, au temps pour moi alors :), de toute façon faut bien réliaser le job d'une manière ou d'une autre.

Citation:

Envoyé par hegros

Tu peux préciser les interfaces que tu as avec les projets consommateurs ? C'est utilisé en multiprocessus ou multithread ?

Pour faire simple, voici ce que je réalise. J'ai un client TCP (une classe séparée, le démon en question) qui doit envoyer de manière synchrone les tâches au serveur (ça peut être du log, des traces, des exécutions de shellscript, ...).

Le client TCP se charge de reposter les requêtes jusqu'a ce qu'elles réussissent.

J'ai donc différentes classes (Log, Shellscript, Trace) qui rajoutent donc les taches à envoyer au serveur (et en récupérent le résultat).

C'est du multithread, monoprocessus.

Citation:

Envoyé par hegros

Il y a des façons élégantes de faire, il existe sûrement des patterns, je ne comprends pas ce qui té chiffonne avec le ASyncCallBack tu peux développer avec du code ce qui bloque ?

C'est pas que ça me chiffonne, c'est juste que je ne sais pas comment initialiser l'IAsyncResult que l'on doit passer en paramètre du IAsynCallback ;).

Citation:

Envoyé par hegros

Ah ok, tu as oublié aussi de réécire le dequeue non ?

Le Dequeue se fait dans le client TCP, une fois la requête correctement exécutée (dans le Thread consommateur).

Citation:

Envoyé par hegros

Une idée serait là aussi d'utiliser par exemple le pattern command. Ainsi tu ajoutes dans ta queue la tache avec une command que tu exécutes une fois que tu as finis d'exécuter la tâche en queue

Je vais voir ce pattern ;)

:merci: pour les réponses constructives et pour l'aide :)

Je ne sais pas trop où tu en es mais mon pifomètre de développeur me dit que tu as trouvé une autre solution :aie:

Pour information je pensais à utiliser le pattern commande de la sorte(on peut biensûr remanier pour créer les classes du pattern comme invokeur etc..)

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
 
 
public interface ICommand
{
   public void ExecuteTask();
   // ou public void ExecuteTask(UnParametre unParametre);
}
 
class CreateurDeTache1 : ICommand
{
    TacheDemon t = new TacheDemon(this);
    // ...
    MonDemon.Enqueue(t);
 
   public void ExecuteTask() {}
}
 
class CreateurDeTache2 : ICommand
{
    TacheDemon t = new TacheDemon(this);
    // ...
    MonDemon.Enqueue(t);
   public void ExecuteTask() {}
}
 
 
 
class MonDemon
{
    private static Queue<TacheDemon> _listeTaches = new Queue<TacheDemon>();
 
    private Thread Scruteur ;
    public static void Enqueue(TacheDemon tache)
    {
         lock(_listeTaches)
         {
           _listeTaches.Enqueue(tache);
         }
    }
    //implémenter aussi une méthode stop/pause/
    public void StartScrutation()
   {  //verifier avant qu'il n'est pas déjà en cours d'execution
       Scruteur = new Thread(Scrute);
      Scruteur.Start();
   }
 
 
public void Scrute()
{   
 while(ConditionForQuitScrute)
 {
    lock(_listeTaches)
    { 
       foreach(TacheDemon tachedemon in _listeTache)
       {
               tachedemon.ExecuteTask();
               // ou tachedemon.ExecuteTask(Resultat);
         }
       }
    }
}
 
}

tu peux aussi ajouter un paramètre à l'interface ExecuteTask si tu veux donner un résultat à l'appelant

T'as raison, j'ai fini trouver entre temps, même si je regrette un peu le mode opératoire :?

Voici ce que ça donne :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
    internal sealed class MySocketTask
    {
        internal MySocketTask(Guid guid,
            string serverName,
            XmlDocumentFragment task,
            MySocketResultEventHandler callback)
        {
            this.Guid = guid;
            this.ServerName = serverName;
            this.Task = task;
            this.Callback = callback;
        }
 
        internal Guid Guid { get; set; }
        internal string ServerName { get; set; }
        internal XmlDocumentFragment Task { get; set; }
        internal MySocketResultEventHandler Callback { get; set; }
    }
 
    internal sealed class MySocket : IDisposable
    {
        private static Queue<MySocketTask> _queue = new Queue<MySocketTask>();
 
        // Queue consumer
        private static Thread _queueConsumer = new Thread((ThreadStart) OnEnqueued);
        private static bool _isRunning;
 
        internal static void Enqueue(MySocketTask task)
        {
            if (!_isRunning)
            {
                _isRunning = true;
                _queueConsumer.Start();
            }
 
            if (task != null)
            {
                // Ensure tasks are enqueued in the right order
                lock (_queue)
                {
                    _queue.Enqueue(task);
                }
            }
        }
 
        private static void OnEnqueued()
        {
            MySocketTask task;
 
            while (true)
            {
                if (_queue.Count > 0)
                {
                    lock (_queue)
                    {
                        task = _queue.Dequeue();
                    }
 
                    using (MySocket socket = new MySocket(task))
                    {
                        socket.Open();
 
                        if (task.Callback != null)
                        {
                            task.Callback.Invoke(task.Guid, socket.Execute());
                        }
                        else
                        {
                            socket.Execute();
                        }
 
                        socket.Close();
                    }
                }
            }
        }
    }

Le seul "bémol", c'est que les créateurs de tâches ont une mini-usine a gaz pour attendre le retour de l'exécution de leur tache...

Car oui, je créé les tâches de manière asynchrone, mais certains créateurs de taches eux, peuvent être synchrones...

Un exemple tout simple de créateur de tache qui attend le retour (synchrone) :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
    public static class MyTaskCreator
    {
        private static Hashtable _guids = new Hashtable();
        private static Hashtable _results = new Hashtable();
 
        public static int Execute(string serverName, string task)
        {
            Guid guid = Guid.NewGuid();
            ManualResetEvent manualEvent = new ManualResetEvent(false);
            int result;
 
            _guids.Add(guid, manualEvent);
            _results.Add(guid, null);
 
            XmlDocumentFragment xmlTask = new XmlDocument().CreateDocumentFragment();
            xmlTask .InnerXml = "<mytask><![CDATA[" + task + "]]></mytask>";
 
            MySocket.Enqueue(new MySocketTask(guid, serverName, xmlTask, OnCompleted));
            manualEvent.WaitOne();
 
            result = (int)_results[guid];
 
            _guids.Remove(guid);
            _results.Remove(guid);
 
            return result;
        }
 
        private static void OnCompleted(Guid guid, XmlTextReader reader)
        {
            reader.ReadToFollowing("return");
            _results[guid] = int.Parse(reader.GetAttribute("value"));
 
            ((ManualResetEvent)_guids[guid]).Set();
        }
    }