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

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

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

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

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

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

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

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

Изменить: также посмотрите некоторые параллельные библиотеки, например 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, которую вы можете векторизовать для одновременного запуска четырех входных точек через один регистр. С четырьмя ядрами и четырехъядерным SIMD вы можете теоретически получить 16-кратное ускорение, но это предполагает, что работа, которую выполняет InnerFunction, в основном является фиксированной математической функцией, поскольку ветвление имеет тенденцию сводить на нет прирост производительности SSE / VMX.

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

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

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

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

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

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

вы могли бы вывернуть цикл наизнанку, используя Compare-and-Swap, чтобы получить приращение без атомарной блокировки:

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.
}

Изменить: мои потоки ржавые, поэтому они не компилируются из-за неправильных имен

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

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

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

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