Interaction d'application multithread avec un thread de journalisation

Question

Me revoilà avec des questions sur le multi-threading et un exercice de ma classe Programmation concurrente .

J'ai un serveur multi-thread - implémenté à l'aide de .NET Modèle de programmation asynchrone - avec GET ( téléchargement ) et PUT ( télécharger ) les services de fichiers. Cette partie est terminée et testée.

Il arrive que l'énoncé du problème indique que ce serveur doit avoir une activité de journalisation ayant un impact minimal sur le temps de réponse du serveur et qu'il doit être pris en charge par un thread de faible priorité - fil de journalisation - créé pour cet effet. Tous les messages de journalisation doivent être transmis par les threads qui les produisent vers ce thread de journalisation , en utilisant un mécanisme de communication qui ne peut pas verrouille le thread qui l'invoque (outre le verrouillage nécessaire pour assurer une exclusion mutuelle) et en supposant que certains messages de journalisation puissent être ignorés.

Voici ma solution actuelle. Aidez-nous à confirmer si cela constitue une solution au problème déclaré:

using System;
using System.IO;
using System.Threading;

// Multi-threaded Logger
public class Logger {
    // textwriter to use as logging output
    protected readonly TextWriter _output;
    // logger thread
    protected Thread _loggerThread;
    // logger thread wait timeout
    protected int _timeOut = 500; //500ms
    // amount of log requests attended
    protected volatile int reqNr = 0;
    // logging queue
    protected readonly object[] _queue;
    protected struct LogObj {
        public DateTime _start;
        public string _msg;
        public LogObj(string msg) {
            _start = DateTime.Now;
            _msg = msg;
        }
        public LogObj(DateTime start, string msg) {
            _start = start;
            _msg = msg;
        }
        public override string ToString() {
            return String.Format("{0}: {1}", _start, _msg);
        }
    }

    public Logger(int dimension,TextWriter output) {
        /// initialize queue with parameterized dimension
        this._queue = new object[dimension];
        // initialize logging output
        this._output = output;
        // initialize logger thread
        Start();
    }
    public Logger() {
        // initialize queue with 10 positions
        this._queue = new object[10];
        // initialize logging output to use console output
        this._output = Console.Out;
        // initialize logger thread
        Start();
    }

    public void Log(string msg) {
        lock (this) {
            for (int i = 0; i < _queue.Length; i++) {
                // seek for the first available position on queue
                if (_queue[i] == null) {
                    // insert pending log into queue position
                    _queue[i] = new LogObj(DateTime.Now, msg);
                    // notify logger thread for a pending log on the queue
                    Monitor.Pulse(this);
                    break;
                }
                // if there aren't any available positions on logging queue, this
                // log is not considered and the thread returns
            }
        }
    }

    public void GetLog() {
        lock (this) {
            while(true) {
                for (int i = 0; i < _queue.Length; i++) {
                    // seek all occupied positions on queue (those who have logs)
                    if (_queue[i] != null) {
                        // log
                        LogObj obj = (LogObj)_queue[i];
                        // makes this position available
                        _queue[i] = null;
                        // print log into output stream
                        _output.WriteLine(String.Format("[Thread #{0} | {1}ms] {2}",
                                                        Thread.CurrentThread.ManagedThreadId,
                                                        DateTime.Now.Subtract(obj._start).TotalMilliseconds,
                                                        obj.ToString()));
                    }
                }
                // after printing all pending log's (or if there aren't any pending log's),
                // the thread waits until another log arrives
                //Monitor.Wait(this, _timeOut);
                Monitor.Wait(this);
            }
        }
    }

    // Starts logger thread activity
    public void Start() {
        // Create the thread object, passing in the Logger.Start method
        // via a ThreadStart delegate. This does not start the thread.
        _loggerThread = new Thread(this.GetLog);
        _loggerThread.Priority = ThreadPriority.Lowest;
        _loggerThread.Start();
    }

    // Stops logger thread activity
    public void Stop() {
        _loggerThread.Abort();
        _loggerThread = null;
    }

    // Increments number of attended log requests
    public void IncReq() { reqNr++; }

}

En gros, voici les points principaux de ce code:

1. Démarrez un thread de faible priorité qui boucle la file d'attente de journalisation et imprime les journaux en attente vers la sortie. Ensuite, le thread est suspendu jusqu'à l'arrivée du nouveau journal ;
2. Quand un journal arrive, le thread de journalisation est réveillé et fait son travail.

Cette solution est-elle thread-safe ? Je lisais le problème et l’algorithme de solution Producteurs-Consommateurs , mais dans ce problème, même si j’ai plusieurs producteurs, je n’ai qu’un lecteur.

Merci d'avance pour toute votre attention.

Answer 1

Il semble que cela devrait fonctionner. Les producteurs-consommateurs ne devraient pas beaucoup changer dans le cas d'un consommateur unique. Petits nitpicks:

• L’acquisition d’un verrou peut être une opération coûteuse (comme le dit @ Vitaly Lipchinsky). Je vous recommande de comparer votre enregistreur à un enregistreur et à un enregistreur naïfs à écriture directe, en utilisant des opérations imbriquées. Une autre alternative consisterait à échanger la file d'attente existante avec une file vide dans GetLog et à laisser immédiatement la section critique. De cette façon, aucun des producteurs ne sera bloqué par de longues opérations chez les consommateurs.

• font le type de référence LogObj (classe). Il ne sert à rien de le structurer puisque vous le boxez quand même. ou bien rendre le champ _queue de type LogObj [] (c'est mieux de toute façon).

• faites votre arrière-plan de thread afin qu'il n'empêche pas la fermeture de votre programme si Stop ne sera pas appelé.

• Rincez votre TextWriter . Sinon, vous risquez de perdre même les enregistrements qui ont réussi à tenir dans la file d'attente (10 éléments, c'est un peu petit à mon humble avis).

• Implémentez IDisposable et / ou le finaliseur. Votre enregistreur possède le fil et le rédacteur de texte et ceux-ci doivent être libérés (et vidés - voir ci-dessus).

Answer 2

Hé là-bas. Si j’ai jeté un coup d’œil rapide, et même s’il semble sûr d’être thread-safe, je ne pense pas que ce soit particulièrement optimal. Je suggérerais une solution dans ce sens

REMARQUE: lisez simplement les autres réponses. Ce qui suit est une solution de verrouillage optimiste et plutôt optimale basée sur la vôtre. Les principales différences consistent à verrouiller une classe interne, à minimiser les «sections critiques» et à fournir une terminaison de thread élégante. Si vous souhaitez éviter le verrouillage complet, vous pouvez essayer certains de ces composants volatiles "non verrouillables". liste de liens trucs comme @Vitaliy Lipchinsky suggère.

using System.Collections.Generic;
using System.Linq;
using System.Threading;

...

public class Logger
{
    // BEST PRACTICE: private synchronization object. 
    // lock on _syncRoot - you should have one for each critical
    // section - to avoid locking on public 'this' instance
    private readonly object _syncRoot = new object ();

    // synchronization device for stopping our log thread.
    // initialized to unsignaled state - when set to signaled
    // we stop!
    private readonly AutoResetEvent _isStopping = 
        new AutoResetEvent (false);

    // use a Queue<>, cleaner and less error prone than
    // manipulating an array. btw, check your indexing
    // on your array queue, while starvation will not
    // occur in your full pass, ordering is not preserved
    private readonly Queue<LogObj> _queue = new Queue<LogObj>();

    ...

    public void Log (string message)
    {
        // you want to lock ONLY when absolutely necessary
        // which in this case is accessing the ONE resource
        // of _queue.
        lock (_syncRoot)
        {
            _queue.Enqueue (new LogObj (DateTime.Now, message));
        }
    }

    public void GetLog ()
    {
        // while not stopping
        // 
        // NOTE: _loggerThread is polling. to increase poll
        // interval, increase wait period. for a more event
        // driven approach, consider using another
        // AutoResetEvent at end of loop, and signal it
        // from Log() method above
        for (; !_isStopping.WaitOne(1); )
        {
            List<LogObj> logs = null;
            // again lock ONLY when you need to. because our log
            // operations may be time-intensive, we do not want
            // to block pessimistically. what we really want is 
            // to dequeue all available messages and release the
            // shared resource.
            lock (_syncRoot)
            {
                // copy messages for local scope processing!
                // 
                // NOTE: .Net3.5 extension method. if not available
                // logs = new List<LogObj> (_queue);
                logs = _queue.ToList ();
                // clear the queue for new messages
                _queue.Clear ();
                // release!
            }
            foreach (LogObj log in logs)
            {
                // do your thang
                ...
            }
        }
    }
}
...
public void Stop ()
{
    // graceful thread termination. give threads a chance!
    _isStopping.Set ();
    _loggerThread.Join (100);
    if (_loggerThread.IsAlive)
    {
        _loggerThread.Abort ();
    }
    _loggerThread = null;
}

Answer 3

En fait, vous introduisez le verrouillage ici. Vous avez le verrouillage en poussant une entrée de journal dans la file d'attente (méthode Log): si 10 threads poussent simultanément 10 éléments dans la file d'attente et réveillent le thread Logger, le 11ème thread attendra que le fil de journalisation consigne tous les éléments ...

Si vous voulez quelque chose de vraiment extensible, mettez en place une file d'attente sans verrouillage (l'exemple ci-dessous) Avec un mécanisme de synchronisation de file d’attente sans verrouillage, ce sera tout de suite (vous pouvez même utiliser une seule attente pour les notifications).

Si vous ne parvenez pas à trouver une implémentation de la file d'attente sans verrouillage sur le Web, voici une idée de la procédure à suivre: Utilisez la liste chaînée pour une implémentation. Chaque noeud de la liste liée contient une valeur et une référence volatile au noeud suivant. par conséquent, vous pouvez utiliser la méthode Interlocked.CompareExchange pour les opérations en file d'attente et en file d'attente. J'espère que l'idée est claire. Sinon, faites-le moi savoir et je vous fournirai plus de détails.

Answer 4

Je suis en train de faire une expérience de pensée ici, car je n'ai pas le temps d'essayer le code pour le moment, mais je pense que vous pouvez le faire sans verrou si vous êtes créatif.

Demandez à votre classe de journalisation de contenir une méthode qui alloue une file d’attente et un sémaphore à chaque appel (et une autre qui libère la file et le sémaphore lorsque le thread est terminé). Les threads souhaitant se connecter appellent cette méthode au démarrage. Lorsqu'ils veulent se connecter, ils envoient le message dans leur propre file d'attente et définissent le sémaphore. Le thread de journalisation a une grande boucle qui parcourt les files d'attente et vérifie les sémaphores associés. Si le sémaphore associé à la file d'attente est supérieur à zéro, elle disparaîtra et le sémaphore sera décrémenté.

Parce que vous n'essayez pas de supprimer des éléments de la file d'attente avant la définition du sémaphore et que vous ne le définissez pas avant d'avoir inséré des éléments dans la file d'attente, je pense ce sera en sécurité. Selon la documentation MSDN de la classe de files d'attente, si vous énumérez la file d'attente et qu'un autre thread modifie la collection, une exception est levée. Catch cette exception et vous devriez être bon.