2

問題は次のとおりです。各スレッドで順番に使用する必要がある5つのメソッドを言いました

Method1(); 次に Method2(); 次に Method3(); 次に Method4(); 次に Method5();

1から5までの番号が付けられた5つのスレッドも実行しています

次のシナリオを実装したいと思います。

スレッド 1 で method1 の使用を開始してから、method2 に移動する [並行して、現在使用されていない method1 の使用をスレッド 2 で開始する]

次に、スレッド 1 がメソッド 3 に移動し、スレッド 2 がメソッド 2 に進むと、スレッド 3 は解放されたメソッド 1 の使用を開始する必要があります。

public void Execute(object OPCounter)
{
    //Method 1
    lock (thisLock)
    {
    FetchedInstructionQueue[PCounter] = Stager.Stage1(InstructionsMemory);
    }

    //Method 2
    lock (thisLock)
    {
    DecordedInstructionQueue[PCounter] = Stager.Stage2(FetchedInstructionQueue, regMem);
    }

    //Method 3
    lock (thisLock)
    {
    ALUResultQueue[PCounter] = Stager.Stage3(DecordedInstructionQueue);
    }

    lock (thisLock)
    {
    MemoryQueue[PCounter] = Stager.Stage4(DecordedInstructionQueue, memory, ALUResultQueue);
    }

    lock (thisLock)
    {
    object obj = Stager.Stage5(DecordedInstructionQueue, ALUResultQueue, regMem, memory, MemoryQueue);
    InternalWriter(PCounter, obj);
    }

}

///This is the initiator of threads

private void ExecuteBtn_Click(object sender, EventArgs e)
{
    InstructionsMemory = InstructionsTextBox.Text.Split('\n');
    FetchedInstructionQueue = new string[InstructionsMemory.Length];
    DecordedInstructionQueue = new Instruction[InstructionsMemory.Length];
    ALUResultQueue = new int[InstructionsMemory.Length];
    MemoryQueue = new int[InstructionsMemory.Length];
    Thread[] threads = new Thread[InstructionsMemory.Length];

    for (APCounter = 0; APCounter < InstructionsMemory.Length; APCounter = 5 + APCounter)
    {
        if (APCounter + 5 < InstructionsMemory.Length)
        {
            object s1 = APCounter;
            object s2 = APCounter + 1;
            object s3 = APCounter + 2;
            object s4 = APCounter + 3;
            object s5 = APCounter + 4;

            threads[APCounter] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 1] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 2] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 3] = new Thread(new ParameterizedThreadStart(Execute));
            threads[APCounter + 4] = new Thread(new ParameterizedThreadStart(Execute));

            threads[APCounter].Start(s1);
            threads[APCounter + 1].Start(s2);
            threads[APCounter + 2].Start(s3);
            threads[APCounter + 3].Start(s4);
            threads[APCounter + 4].Start(s5);
        }
    }
4

1 に答える 1

2

必要な説明がコメントに収まらないため、これを回答として書いています。

実行する作業のパイプラインがあるようです (特定のオブジェクトで、変更される場合と変更されない場合があります)。このパイプラインを実行するためのスレッドもいくつかあります。パイプラインは 5 つのステージで構成されています。

一般に、パイプラインでは、パイプラインのステップごとに 1 つのスレッドが必要です (つまり、ステップ 1 に 1 つのスレッド、ステップ 2 に 1 つのスレッド、ステップ 3 に 1 つのスレッドなど)。これをオプション A としましょう。

作業中のオブジェクトにスレッドが追従するように設定したいようです。したがって、スレッド 1 はオブジェクト 1 を 5 つのステージすべてでカバーし、スレッド 2 はオブジェクト 2 をカバーするというように続きます。なぜこれを行う必要があるのか​​ は明確ではありませんが、とにかく実行しましょう. これをオプション B と呼びましょう。

簡単にするために、3 つのスレッドと 3 つのステージを使用するオプションを示します。

オプション A: 従来のパイプライン

3 ステージ、ステージごとに 1 つのスレッド、オブジェクトはステージ間を移動します。

void Main()
{
    var stage1Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());
    var stage2Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());
    var stage3Queue = new BlockingCollection<object>(new ConcurrentQueue<object>());

    var threads = new Thread[] {new Thread(() => Stage1Worker(stage1Queue, stage2Queue)),
                                new Thread(() => Stage2Worker(stage2Queue, stage3Queue)),
                                new Thread(() => Stage3Worker(stage3Queue))
                               };

    foreach (var thread in threads) thread.Start();

    stage1Queue.Add("*");
    stage1Queue.Add("*");
    stage1Queue.Add("*");

    Console.ReadKey();
}

public void Stage1Worker(BlockingCollection<object> queue, BlockingCollection<object> next)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        next.TryAdd(task.ToString() + "*"); // will always succeed for a ConcurrentQueue
    }
}

public void Stage2Worker(BlockingCollection<object> queue, BlockingCollection<object> next)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        next.TryAdd(task.ToString() + "*"); // will always succeed for a ConcurrentQueue
    }
}

public void Stage3Worker(BlockingCollection<object> queue)
{
    foreach (var task in queue.GetConsumingEnumerable())
    {
        Console.WriteLine(task); // do work here, even mutating task if needed
        // no more work!
    }
}

オプション B: 同期されたメソッド アクセス パイプライン

これは非常に奇妙で、この「理由」を知らずに適切な解決策を見つけるのは困難です。以下は、1 つのタスクが 1 つのスレッドによって実行され、スレッドが各メソッドへのアクセスを待機することを保証します。ただし、スレッド 1 がタスク 1、スレッド 2 がタスク 2 などを実行するとは限りません。準備ができているスレッドが「次の」タスクを取得します。

object stage1Lock = new object();
object stage2Lock = new object();
object stage3Lock = new object();

void Main()
{
    var tasks = new BlockingCollection<object>(new ConcurrentQueue<object>());

    var threads = new Thread[] {new Thread(() => Worker(1, tasks)),
                                new Thread(() => Worker(2, tasks)),
                                new Thread(() => Worker(3, tasks))
                               };

    foreach (var thread in threads) thread.Start();

    tasks.Add("*");
    tasks.Add("**");
    tasks.Add("***");
    tasks.Add("****");
    tasks.Add("*****");

    LINQPad.Util.ReadLine();
}

public void Worker(int id, BlockingCollection<object> tasks)
{
    foreach (var task in tasks.GetConsumingEnumerable())
    {   
        Console.WriteLine(id + " got task: " + task);

        lock (stage1Lock){
            Console.WriteLine(id + " - Stage 1: " + task);
        }

        lock (stage2Lock){
            Console.WriteLine(id + " - Stage 2: " + task);
        }

        lock (stage3Lock){
            Console.WriteLine(id + " - Stage 3: " + task);
        }
    }
}
于 2012-04-30T03:20:35.393 に答える