C # Threading для конвейера

В настоящее время я создаю конвейер для моего симулятора asm. Я сделаю это с многопоточностью. В настоящее время у меня есть четыре потока, которые соответствуют каждой выборке, декодированию, выполнению и обратной записи. Я не понимаю, как правильно кодировать конвейер. Вот часть моего кода:

private void Fetch()
{
    while(_continue)
    {
        fetchEvent.WaitOne();
        if (!_continue) break;
        lock (irLock)
        {
            try
            {
                // Do stuff to fetch instruction...

                catch (IndexOutOfRangeException e) { Err = "Segmentation Fault at Line " + _pc.ToString(); }
                catch (Exception e) { Err = e.Message + "Line " + _pc.ToString(); }
                _pc++;

                GiveTurnTo(2); // used these 2 lines to handle mutual exclusion
                WaitTurn(1);
            }
        }
    }

    private void Decode()
    {
        while (_continue)
        {
            decodeEvent.WaitOne();
            if (!_continue) break;

            lock (irLock)
            {
                WaitTurn(2);
                Console.WriteLine("decode works");
                GiveTurnTo(1);
            }


        }
    }

    private void Execute()
    {
        while (_continue)
        {
            exeEvent.WaitOne();
            if (!_continue) break;

            lock (irLock)
            {
                //WaitTurn(3);
                Console.WriteLine("Am I, execute, getting a turn?");
                // GiveTurnTo(4);
            }

        }

    }

    private void WriteBack()
    {
        while (_continue)
        {
            wbEvent.WaitOne();
            if (!_continue) break;

            lock (irLock)
            {
                Console.WriteLine("Am I, Write Back, getting a turn?");
                //GiveTurnTo(1);
                // WaitTurn(4);
            }

        }

    }
}

Я использую этот метод для запуска цикла нажатием кнопки:

public void nextInstruction()
{
    fetchEvent.Set();
    decodeEvent.Set();
    exeEvent.Set();
    wbEvent.Set();
}
I was thinking of changing nextInstruction() to this:

public void nextInstruction()
{
    fetchEvent.Set();
}

При этом каждое нажатие кнопки всегда начинается с «Выбрать». После этого я подумал, что, может быть, я мог бы поместить событие set в методе Fetch для вызова следующей части цикла (Decode) и сделать то же самое для следующих методов. Я бы получил что-то вроде этого:

private void Fetch()
{
    while(_continue)
    {
        // Do stuff....
        decodeEvent.Set();
    }
}

private void Decode()
{
    while (_continue)
    {
        // Do stuff...
        exeEvent.Set();
    }
}

private void Execute()
{
    while (_continue)
    {
        // Do stuff...
        wbEvent.Set();
    }
}
private void WriteBack()
{
    while (_continue)
    {
        // Do stuff...
    }
}

Вместо того, чтобы оставлять эти установленные события открытыми, я думаю, что они должны активироваться какой-то логикой, но я не знаю, что это за логика. Может быть, их можно было активировать нажатием кнопки. Это позволило бы мне контролировать, когда инструкция передается по циклу. Теоретически я думаю, что это могло бы дать мне структуру конвейера. Может ли кто-нибудь внести свой вклад в это? Можно ли этого добиться с помощью только событий автоматического сброса?

Если вы посмотрите на первый блок кода, который я дал, вы увидите, что я пытался использовать блокировку, но это сделало так, что одновременно мог работать только один поток. Я хочу сделать так, чтобы он соответствовал формату Fetch0, {Decode0, Fetch1}, {Execute0, Decode1, Fetch3},... и так далее. Нужны ли в этом случае замки?


multithreading c# pipelining
person blutuu    schedule 16.11.2013    source источник

Ответы (2)


arrow_upward
1
arrow_downward

Нити для этого не подходят. Вы делаете слишком мало работы на шаг, прежде чем перевести поток в спящий режим. Лучше всего смоделировать это, если бы один поток выполнял каждый из шагов в обратном порядке, это повлияет на работу системы так же, как и в реальной системе.

public void nextInstruction()
{
    WriteBack();
    Execute();
    Decode();
    Fetch();
}

Чтобы понять, почему я обрабатываю их в обратном порядке, взгляните на то, как работает моя система, как мы ее постоянно называем.

  • Call 1
    1. WriteBack() does nothing, its input buffer is empty.
    2. Execute() ничего не делает, его входной буфер пуст.
    3. Decode() ничего не делает, его входной буфер пуст.
    4. Fetech() Берет 1-ю инструкцию и помещает ее во входной буфер Decode().
  • Call 2
    1. WriteBack() does nothing, its input buffer is empty.
    2. Execute() ничего не делает, его входной буфер пуст.
    3. Decode() Захватывает полученную 1-ю инструкцию из входного буфера, помещает декодированную инструкцию во входной буфер Execute().
    4. Fetech() Захватывает вторую инструкцию и помещает ее во входной буфер Decode().
  • Call 3
    1. WriteBack() does nothing, its input buffer is empty.
    2. Execute() Захватывает декодированную 1-ю инструкцию из входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    3. Decode() Захватывает полученную 2-ю инструкцию из своего входного буфера, обрабатывает ее, а затем помещает декодированную инструкцию во входной буфер Execute().
    4. Fetech() Берет третью инструкцию и помещает ее во входной буфер Decode().
  • Call 4
    1. WriteBack() Grabs the processed 1st value from its input buffer and writes it back to the memory store.
    2. Execute() Захватывает декодированную 2-ю инструкцию из своего входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    3. Decode() Захватывает полученную третью инструкцию из своего входного буфера, обрабатывает ее, затем помещает декодированную инструкцию во входной буфер Execute().
    4. Fetech() Захватывает 4-ю инструкцию и помещает ее во входной буфер Decode().
  • Call 5
    1. WriteBack() Grabs the processed 2nd value from its input buffer and writes it back to the memory store.
    2. Execute() Захватывает декодированную третью инструкцию из входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    3. Decode() Захватывает полученную 4-ю инструкцию из своего входного буфера, обрабатывает ее, затем помещает декодированную инструкцию во входной буфер Execute().
    4. Fetech() Берет 5-ю инструкцию и помещает ее во входной буфер Decode().
  • Call 6
    1. WriteBack() Grabs the processed 3rd value from its input buffer and writes it back to the memory store.
    2. Execute() Захватывает декодированную 4-ю инструкцию из входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    3. Decode() Захватывает полученную 5-ю инструкцию из своего входного буфера, обрабатывает ее, затем помещает декодированную инструкцию во входной буфер Execute().
    4. Fetech() Берет 6-ю инструкцию и помещает ее во входной буфер Decode().
  • т. д.

Видите, требуется 4 вызова, прежде чем конвейер заполнится, это то, что происходит в реальной жизни, требуется больше циклов для получения первой инструкции от Fetch до WriteBack, однако период времени между инструкциями, выходящими из конвейера, намного короче по сравнению.

Теперь, глядя на свою старую систему, вы можете увидеть, что на самом деле вы никогда не заполняете свой конвейер, одна инструкция может пройти все этапы за один шаг моделирования.

  • Call 1
    1. Fetech() Grabs the 1st instruction and puts it in to Decode()'s input buffer.
    2. Decode() Захватывает полученную 1-ю инструкцию из своего входного буфера, обрабатывает ее, затем помещает декодированную инструкцию во входной буфер Execute().
    3. Execute() Захватывает декодированную 1-ю инструкцию из входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    4. WriteBack() Захватывает обработанное 1-е значение из входного буфера и записывает его обратно в хранилище памяти.
  • Call 2
    1. Fetech() Grabs the 2nd instruction and puts it in to Decode()'s input buffer.
    2. Decode() Захватывает извлеченную 2-ю инструкцию из своего входного буфера, обрабатывает ее, а затем помещает декодированную инструкцию во входной буфер Execute().
    3. Execute() Захватывает декодированную 2-ю инструкцию из входного буфера, обрабатывает ее, а затем помещает результат во входной буфер WriteBack().
    4. WriteBack() Захватывает обработанное 2-е значение из входного буфера и записывает его обратно в хранилище памяти.

За один вызов наша инструкция перешла от полученной к записанной, прежде чем мы даже смогли получить вторую инструкцию.

person Scott Chamberlain    schedule 16.11.2013
comment
Я понимаю, о чем вы говорите, но это для задания, и каждый цикл должен выполняться с собственным потоком. У меня возникла идея, чтобы инструкции не выполнялись за один звонок. Для этого я хочу, чтобы каждый цикл вызывал следующее (кроме обратной записи) AutoResetEvent каждый раз, когда нажимается кнопка. Если эти циклы не готовы к вызову, их не будет. Что вы думаете? - person blutuu; 16.11.2013
comment
Вы дали слишком общее описание, чтобы я мог об этом думать. Попробуйте, если не работает, задайте новый вопрос о том, что не сработало. - person Scott Chamberlain; 16.11.2013

arrow_upward
1
arrow_downward

Конвейер - это процесс, который вызывает синхронизацию или асинхронность событий. Наблюдатели регистрируются в конвейере, чтобы получать уведомления об этих событиях. Конвейер хранит состояние, чтобы наблюдатели могли работать с ним. В случае асинхронных событий вам нужно будет обеспечить синхронизацию состояния, а также обратные вызовы, чтобы узнать, когда поток завершен или готов к работе. Вот грубый пример высокоуровневого использования конвейера:

public class OnFetchEvent : AsyncEvent {}
public class OnDecodeEvent : AsyncEvent {}
public class OnExecuteEvent : AsyncEvent {}
public class OnWritebackEvent : AsyncEvent {}

public class FetchObserver : Observer,
   IObserve<OnFetchEvent>
{
   public void OnEvent(OnFetchEvent @event)
   {
      ....do some stuff

      // raise the next event
      RaiseEvent<OnDecodeEvent>(); 
   }
}

public class Pipeline
{
   public void RaiseEvent<TEvent>()
   {
      if (typeof(TEvent) is AsyncEvent)
        ...create thread and raise the event which will notify the appropriate 
           observers of the event in the newly created thread
   }

  }

Использование:

pipeline.RegisterObserver<FetchObserver>()
  .AndObserver<DecodeObserver>()
  .AndObserver<ExecuteObserver>()
  .AndObserver<WriteBackObserver>();

pipeline.RaiseEvent<OnFetchEvent>();

Это будет обрабатывать один цикл. События вызываются асинхронно (что означает, что наблюдатели будут выполняться для каждого потока). Чтобы иметь несколько циклов, вам нужно запустить асинхронный конвейер (собственный поток). И при необходимости вызовите конвейер RaiseEvent. Я надеюсь, что это обеспечивает некоторый высокоуровневый подход к использованию конвейера с использованием потоков.

person Community    schedule 16.11.2013
comment
В описанной вами ситуации вы хотите создать 4 потока для задачи. Это не то же самое, что работать с конвейером. В этом сценарии вам потребуется 4 очереди для перемещения работы из одного потока в другой с помощью постановки в очередь и удаления из очереди, где потоки отслеживают очереди на работу и синхронно обрабатывают каждую задачу в очереди. - person ; 16.11.2013