ロガースレッドとのマルチスレッドアプリケーションの相互作用

Question

ここで、マルチスレッドに関する質問と、同時プログラミングクラスの演習について再度説明します。

.NET 非同期プログラミングモデルを使用して実装されたマルチスレッドサーバーがあり、 GET （ download ）および PUT （アップロード）ファイルサービス。この部分は完了してテストされています。

このサーバーには、サーバーの応答時間への影響が最小限のロギングアクティビティが必要であり、低優先度のスレッド-ロガースレッド-この効果のために作成されました。すべてのロギングメッセージは、できない通信メカニズムを使用して、このロガースレッドに生成するスレッドによって渡されるものとします。 strong>呼び出したスレッドをロックし（相互排除を確実にするために必要なロックに加えて）、一部のロギングメッセージが無視されると仮定します。

これが私の現在の解決策です。これが上記の問題の解決策であるかどうかの検証にご協力ください：

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++; }

}

基本的に、このコードの主なポイントは次のとおりです。

1. 低優先度のスレッドを開始して、ロギングキューをループし、保留中のログを出力に出力します。この後、新しい log が到着するまで thread は中断されます;
2. ログが到着すると、ロガースレッドが起動され、動作します。

このソリューションはスレッドセーフですか？ 生産者-消費者の問題と解決アルゴリズムを読んでいますが、この問題では複数の生産者がいますが、読者は1人だけです。

ご清聴ありがとうございました。

Answer 1

動作しているようです。生産者と消費者は、単一の消費者の場合に大きく変わるべきではありません。ちょっとした工夫：

• ロックの取得は、高価な操作になる可能性があります（@Vitaliy Lipchinskyが言うように）。単純な「ライトスルー」ロガーおよびインターロック操作を使用するロガーに対してロガーのベンチマークを行うことをお勧めします。もう1つの方法は、既存のキューを GetLog の空のキューと交換し、クリティカルセクションをすぐに残すことです。これにより、プロデューサーはコンシューマーでの長時間の操作によってブロックされなくなります。

• LogObj参照型（クラス）を作成します。とにかくそれをボクシングしているので、構造化する意味はありません。または、 _queue フィールドを LogObj [] タイプにします（とにかく良い）。

• Stop が呼び出されない場合にプログラムを閉じないように、スレッドの背景を作成します。

• TextWriter をフラッシュします。それ以外の場合は、キューに収まるように管理されたレコードでさえ失う危険があります（10アイテムは少し小さい私見です）

• IDisposableおよび/またはファイナライザを実装します。ロガーはスレッドとテキストライターを所有しているので、それらを解放（およびフラッシュ-上記参照）する必要があります。

Answer 2

おい。ざっと見てみると、スレッドセーフのように見えますが、特に最適だとは思いません。これらの線に沿って解決策を提案します

注：他の応答を読んでください。以下は、あなた自身に基づいた、かなり最適な、楽観的なロックソリューションです。主な違いは、内部クラスのロック、「クリティカルセクション」の最小化、および正常なスレッド終了の提供です。ロックを完全に回避したい場合は、その揮発性の「非ロック」を試してください。 @Vitaliy Lipchinskyが示唆するように、リンクされたリストのもの。

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

実際には、ここでロックを導入しています。ログエントリをキューにプッシュする際にロックします（Logメソッド）：10個のスレッドが同時に10個のアイテムをキューにプッシュしてロガースレッドを起動した場合、11番目のスレッドはロガースレッドがすべてのアイテムを記録するまで待機します...

本当にスケーラブルなものが必要な場合は、ロックフリーキューを実装します（例を以下に示します）。ロックフリーのキュー同期メカニズムを使用すると、すぐに実行できます（通知に単一の待機ハンドルを使用することもできます）。

Webでロックフリーキューの実装を見つけることができない場合、これを行う方法を以下に示します。 実装にリンクリストを使用します。リンクリストの各ノードには、値と次のノードへの揮発性参照が含まれます。したがって、エンキューおよびデキューの操作には、Interlocked.CompareExchangeメソッドを使用できます。アイデアが明確であることを願っています。そうでない場合はお知らせください。詳細をお知らせします。

Answer 4

コードを実際に試す時間がないので、ここで思考実験をしていますが、創造性があればロックなしでこれを行うことができると思います。

ログクラスには、呼び出されるたびにキューとセマフォを割り当てるメソッド（および、スレッドの完了時にキューとセマフォの割り当てを解除するメソッド）が含まれています。ロギングを実行するスレッドは、起動時にこのメソッドを呼び出します。ログを記録する場合、メッセージを独自のキューにプッシュし、セマフォを設定します。ロガースレッドには、キューを実行して関連するセマフォをチェックする大きなループがあります。キューに関連付けられたセマフォがゼロより大きい場合、キューはポップオフされ、セマフォはデクリメントされます。

セマフォが設定されるまでキューからオブジェクトをポップしようとせず、キューにプッシュするまでセマフォを設定しないため、考えるこれは安全です。キュークラスのMSDNドキュメントによると、キューを列挙しているときに別のスレッドがコレクションを変更すると、例外がスローされます。その例外をキャッチすると、あなたはうまくいくはずです。