Как параллелизировать небольшую чистую функцию?

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

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

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

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

Вы, возможно, должны поместить достаточно данных в "объекты", которые Вы обрабатываете, чтобы смочь сказать, что сделать с ними после того, как Вы сделаны. Они должны почти наверняка быть объектом, и можно хотеть, чтобы они содержали информацию состояния.

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

Править: Также посмотрите на часть параллельной библиотеки, как ThreadPoolExecutor. Легко забыть параллельную библиотеку (как, я просто сделал), это, вероятно, точно, что Вы искали (следовательно акцент)

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

void OuterFunction( Thingy inputData[N] )
{
  for ( int i = 0 ; i < N ; ++i )
    InnerFunction( inputData[i] );
}

Мы решили бы Вашу проблему (предположение, что система очереди заданий присутствует):

void JobFunc( Thingy inputData[], int start, int stop )
{
  for ( int i = start ; i < stop ; ++i )
    InnerFunction( inputData[i] );  
}
void OuterFunction( Thingy inputData[N], int numCores )
{
   int perCore = N / numCores; // assuming N%numCores=0 
                               // (omitting edge case for clarity)
   for ( int c = 0 ; c < numCores ; ++c )
     QueueJob( JobFunc, inputData, c * perCore, (c + 1) * perCore );
}

Пока Ваши входные данные абсолютно независимы, как Вы говорите в своем исходном вопросе, Вы не должны блокировать его; синхронизация только необходима, когда существует зависимость между потоками и здесь нет ни одного.

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

Затем рассмотрите возможность SIMD, можно векторизовать его для выполнения четырех точек ввода через единственный регистр одновременно. С четырьмя ядрами и 4-широким SIMD можно теоретически добраться 16x ускорение, но это предполагает, что работа, которую делает InnerFunction, является главным образом фиксированной математической функцией, так как ветвление имеет тенденцию стирать увеличение производительности SSE/VMX.

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

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

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

Первый блок проблемы должен обеспечить безопасность, правильность и параллелизуемость, и это кажется, что Вам покрыли это, потому что Ваша функция чиста.

Я думаю, что следующая самая сложная часть описывает параллелизм, конкретно Вы упоминаете, что эта функция вызвана многие много раз. Можно ли конвейерно обработать это и отдельное планирование функции от ее работы? Если Вы не можете конвейерно обработать это, делает это похоже на параллельный цикл, обход дерева, или это более неструктурированный, чем это. А именно, при повиновении Amdahl, если Вы не можете перекрыть работу и удостовериться, что существует несколько экземпляров его или что-то еще работающее одновременно, Вы эффективно последовательны даже при том, что Вы чисты. Что-либо, что можно сделать для рефакторинга работы в конвейер, рекурсивный обход дерева (или параллельный цикл) или если Вы должны более неструктурированная работа с explicity зависимостями между задачами, поможет здесь независимо от пользовавшейся библиотеки.

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

Я сожалею, если это не было супер конкретно, но я надеюсь, что это было полезно.

Вы смогли поворачивать цикл наизнанку с помощью Сравнивать-и-подкачивать для получения атомарной блокировки свободный инкремент:

void OuterFunction()
{
  for(int i = 0; i < N; i++)
    InnerFunction(i);
}

переходит в:

void OuterFunction()
{
   int i = 0, j = 0;

   void Go()
   {
      int k;
      while((k = atomicInc(*i)) < N)
      {
         InnerFunction(k);

         atomicInc(*j);
      }
   }

   for(int t = 0; t < ThreadCount - 1; t++) Thread.Start(&Go);

   Go(); // join in

   while(j < N) Wait(); // let everyone else catch up.
}

Править: моя поточная обработка ржава так, чтобы не компилировал, потому что имена неправильны

Это походит на что-то, где инструкции SIMD могут помочь. Если у Вас есть компилятор автовекторизации, необходимо смочь переписать функцию для работы на 4 значения одновременно, и компилятор может уплотнить это в соответствующие инструкции SSE. Это могло помочь сократить вызов функции наверху. Если Ваш компилятор не хорош в автовекторизации кода, то Вы можете использовать SSE intrinsics для получения почти вниз до уровня ассемблера для программирования тела функции.

Между вызовами нет зависимости данных, т. Е. Ни один вызов не использует результат любого другого вызова.

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

Если на эти вопросы ответ «да», то предыдущие предложения подходят, просто попробуйте максимизировать детализацию приложения, назначив выполнение функции как можно большему количеству исполнений на поток.

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