Ускоряем многопоточную программу в 150 раз за три простых шага

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

Вчера вечером я проверил первый способ (сегодня задокументировал-нарисовал графики) и вот что у меня получилось:

Исходный псевдокод собственно обработчика выглядит так:

while(job = nextJob()){ result = processJob(job); emit processed(result); }

Давайте накопим результаты в пачку и выдадим эту пачку принимающей стороне:

while(job = nextJob()){
  resultList.append(processJob(job));
  if(resultList.size() > batchsz) {
     emit processedList(resultList);
     resultList.clear();
   }
}
if(resultList.size() > 0)
     emit processedList(resultList);

Логика приложения несколько меняется: если у нас задачи поступают медленно, то выдача обработанных может тоже замедляться (пока не накопится batchsz), если это не /вполне/ подходит, то можно выдавать processedList или по накоплению или по таймеру, но вообще это все пока мелкие детали.

Берем и тестируем (я взял меньше чем вчера вариантов числа threads, а то задолбался с этими графиками):

Самый нижний график - тот же что вчера, только масштабы теперь логарифмические по обоим осям.

Да, батч дает некоторый результат (в полтора раза для 32 threads/32 batch size), но мы же ждали не этого? Мы ожидали ускорения раз эдак в 32 (для 32/32 случая)?

Лезем в профайлер и убеждаемся, что блокировка на QMutex на месте, просто это теперь другой QMutex, не на обработчике сигналов от threads, а на извлечении следующей задачи из очереди.

OK, а так же OK. Меняем QMutex на простейший спинлок, вот такой:

while (!atomic.testAndSetAcquire(0, 1)) YieldProcessor();

YieldProcessor() транслируется в nop; pause. Можно было просто _mm_pause() воткнуть, разницы никакой.

Смотрим результат:

Стало ну прям сильно лучше, для случая 32/32 - в 16 раз. Параллельная обработка на двух CPU уже не вредна (но и не полезна), ущерб от параллельной обработки на 32 ядрах - всего в 6 раз, а не в 30 как было раньше. Ну и для одного потока - батч в 32 результата в 4 раза лучше, чем по одному их фигачить.

Привычно лезем в профайлер, для случая 32/32 видим, что 86% CPU программа сидит в спинлоке.

OK, а так же ОК. Берем неблокирующую очередь. Я взял вот эту: github.com/cameron314/concurrentqueue потому что она header only, но судя по всему можно было бы взять любую. Компилируем, запускаем, меряем:

Стало ну не хорошо, но гораздо же лучше. Есть какая-то польза от многопоточности (32 потока работают в полтора раза быстрее, чем один, это УЖЕ РЕЗУЛЬТАТ), ну и на самом деле 2.6 миллиона rps меня, в общем, устраивают с огромным запасом (раз в 50-100).

Тем не менее, лезем в профайлер и находим там ТО С ЧЕГО НАЧИНАЛИ, опять тот же QMutex на обработке входящих сигналов от рабочих потоков. Он, конечно, в расчете на единицу работы жрет существенно меньше, но место подвигу еще, очевидно, есть.

В следующих сериях: мы же можем и выход от workers сунуть в такую же неблокирующую очередь. Это, правда, уже заметно поменяет логику: если сейчас у нас есть slot который позовут средствами Qt и сунут ему в зубы результат работы, то с очередью придется заводить consumer thread, как-то им управлять (ну, вероятно, он может сидеть в блокировке на queue.dequeue(), но нужно брать блокирующий вариант очереди) и вообще машинка сильно усложняется. Но я ее таки докую, интересно же.