Потоки. Оптимальный вариант реализации

[veniside]

> что сотню потоков процессор держит без сколько-нибудь заметного ущерба

поток потоку рознь. Из этой сотни потоков почти все спят, иначе проц постоянно был бы загружен на 100%. И нет смысла запускать 3 тяжелых потока на 2-х ядерном проце, это просто ненужная трата ресурсов на переключения между ними, без повышения производительности.


> Затраты на поток по сравнению с секундой совершенно ничтожны.

мы говорим об оптимальном варианте. Сегодня надо создать 30 потоков, завтра 300, послезавтра 3000. Пять бабушек - уже рубь!

[s-andriano]

Сообщение от veniside

> что сотню потоков процессор держит без сколько-нибудь заметного ущерба

поток потоку рознь. Из этой сотни потоков почти все спят, иначе проц постоянно был бы загружен на 100%. И нет смысла запускать 3 тяжелых потока на 2-х ядерном проце, это просто ненужная трата ресурсов на переключения между ними, без повышения производительности.

Нет, я как раз говорю о "тяжелых" потоках, каждый из которых способен обеспечить 100% загрузки ядра.
Не нужно бояться запускать потоков больше, чем есть ядер.

[veniside]

хм, ну вот простой тест:

program
  heavyThreads;

{$APPTYPE CONSOLE}

uses
  Windows;

var
  tc: integer;

procedure mark();
begin
  tc := GetTickCount();
end;

procedure elapsed(const msg: string);
begin
  writeln(msg, GetTickCount() - tc, ' ms');
end;

procedure job();
var
  i: integer;
  f: double;
begin
  f := 121.2;
  for i := 1 to 100000000 do
    f := f / f;
end;

function doJob(param: pointer): DWORD; stdcall;
begin
  job();
  //
  result := 0;
end;


var
  h: TWOHandleArray;
  id: cardinal;
begin
  mark();
  job();  job();  job();  job();
  job();  job();  job();  job();
  elapsed('One thread: ');
  //
  mark();
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(2, @h, true, INFINITE);
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(2, @h, true, INFINITE);
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(2, @h, true, INFINITE);
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(2, @h, true, INFINITE);
  elapsed('Two threads: ');
  //
  mark();
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[2] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[3] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(4, @h, true, INFINITE);
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[2] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[3] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(4, @h, true, INFINITE);
  elapsed('Four threads: ');
  //
  mark();
  h[0] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[1] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[2] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[3] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[4] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[5] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[6] := CreateThread(nil, 0, @doJob, nil, 0, id);
  h[7] := CreateThread(nil, 0, @doJob, nil, 0, id);
  WaitForMultipleObjects(8, @h, true, INFINITE);
  elapsed('Eight threads: ');
  //
end.

Результат:

One thread: 11516 ms
Two threads: 3703 ms
Four threads: 3735 ms
Eight threads: 3765 ms

Угадайте с одного раза, сколько ядер у моего процессора и почему мне не следует запускать более 2-х тяжелых потоков одновременно, чтобы оптимально выполнить одну и ту же работу.


> Не нужно бояться запускать потоков больше, чем есть ядер.

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

[s-andriano]

Запустил Ваш код.

One thread: 4219 ms
Two threads: 3047 ms
Four threads: 3047 ms
Eight threads: 3047 ms

И какие будут выводы?

ничего страшного не произойдёт. Просто работа будет выполнена медленней, ну, это не так страшно

Неочевидно.
Это в простейшем случае можно определить объем работы, приходящийся на каждый поток. А когда неопределенность составляет скажем 2-3 раза, то в случае двух ядер/двух потоков:
- первое ядро будет делать 25% работы, второе - 75%. По сравнению с однопоточным режимом выигрыш составит 25%.
Делим на 100 потоков с тем же разбросом, т.е. от от 0.58 до 1.7% на каждый. В худшем случае ядро в одиночестве будет работать 1.7% времени. Оставшиеся 98.3% будут выполнены за 49.2% времени. Суммарное время составит 50.8% от однопоточного варианта.
Т.е. задача будет посчитана примерно в 1.5 раза быстрее при 100 потоках про сравнению с двумя.

[veniside]

выводы: 2 ядра, и тест слишком грубый и слишком рано заканчивается, чтобы почувствовать разницу )
Добавьте нолик в цикле по i, разница должна стать заметной.

[s-andriano]

Сообщение от veniside

выводы: 2 ядра, и тест слишком грубый и слишком рано заканчивается, чтобы почувствовать разницу )
Добавьте нолик в цикле по i, разница должна стать заметной.

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

Внес некоторые коррективы в Вашу программу:

procedure job;
var
  i,j : integer;
  f: double;
begin
  f := 121.2;
  j := random(10000) + 1;
  for i := 1 to 1000*j do
    f := f / f;
end;

[gusluk]

Чтобы был понятней обсуждаемый вопрос, опишу суть задачи (нужно было это сделать в первом посте).
Это задача на моделирование состояние газовых и жидких сред.
Алгоритм расчетов состояния примерно такой:

Есть начальное состояние среды (начальные условия), загружаем его в исходный массив данных. В зависимости от необходимой точности и разрешения он может представлять собой матрицу от 1000Х1000 и более элементов. Используя этот массив рассчитываем вклад влияния отдельных факторов (сила трение, гравитация и др.) на состояние среды (Fx, Fy, Fz). Факторов может быть от трех до семи (редко больше), зависит от задачи.
Затем влияние отдельных факторов суммируется и рассчитывается новое состояние среды (Fxyz). Новое состояние мы (если это необходимо) как то отображаем графически, записываем его на место начального состояния среды (в исходный массив данных) и рассчитываем следующее состояние. Это все повторяется какое то количество раз (пока не получим интересующий нас результат).
Получается что потоков нам нужно от трех до семи, и пока не рассчитано новое состояние (Fxyz) запускать повторно Fx, Fy, Fz мы не можем. Ф-ции Fx, Fy, Fz используют для расчета весь массив данных, нужно синхронизировать доступ к нему или скопировать массив в каждый поток.
П.С. С синхронизацией разобрался, понял что она при чтении данных не обязательна.

[veniside]

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

кстати да, я тоже это заметил ) Оказалось, что в Дельфи настолько криво сделана работа с плавающей точкой, что random() в одном потоке выполняется быстрее, чем в двух и более. И код с оптимизацией работает медленней, чем код без неё. Поэтому я исключил работу с плавающей точкой, добавил работу с памятью и расширил тест до 64 потоков. Исходник ниже, а результат вот:

1 threads: 3031 ms / 64 jobs.
2 threads: 1563 ms / 64 jobs.
4 threads: 1578 ms / 64 jobs.
8 threads: 1672 ms / 64 jobs.
16 threads: 1968 ms / 64 jobs.
32 threads: 2688 ms / 64 jobs.
64 threads: 3703 ms / 64 jobs.

> первое ядро будет делать 25% работы, второе - 75%

стоп, мы говорим, что все потоки одинаково загружены. Не важно чем, но они не простаивают. Тезис простой: если потоки не простаивают, то их не должно быть больше, чем ядер. Если 100% работы можно разбить на 100 частей, то оптимальное разбиение по ядрам:

1 ядро: 1 поток, 100 частей
2 ядра: 2 потока, по 50 частей каждый
4 ядра: 4 потока, по 25 частей каждый
и т.д.

Ф-ции Fx, Fy, Fz используют для расчета весь массив данных, нужно синхронизировать доступ к нему или скопировать массив в каждый поток.

Так вы и не сказали, меняют ли массив Fx, Fy, Fz или нет. Если меняют, то есть очередность их запуска (т.е. очевидно, что результат работы одной из F будет зависеть от данных, изменнёных в другой F), если нет, то никакой синхронизации не нужно и копирования тоже.

program
  heavyThreads2;

{$APPTYPE CONSOLE}

uses
  Windows, SysUtils;

var
  tc: DWORD;
  nj: integer;

// --  --
procedure mark();
begin
  tc := GetTickCount();
  nj := 0;
end;

// --  --
procedure elapsed(const msg: string);
begin
  writeln(msg, GetTickCount() - tc, ' ms / ', nj, ' jobs.');
end;

var
  A: array[word] of byte;

// --  --
function job(): int64;
var
  i: integer;
begin
  result := 10000;
  for i := 1 to 1000000 do result := result + result div i + A[i mod 65535];
  //
  InterlockedIncrement(nj);
end;

// --  --
function doJob(param: pointer): DWORD; stdcall;
var
  i: integer;
  c: ^integer;
begin
  c := param;
  for i := 1 to c^ do job();	// do job several times
  //
  result := 0;
end;


// -- main --

var
  c, i, t: integer;
  h: TWOHandleArray;
  id: cardinal;
begin
  for c := 0 to 6 do begin
    //
    mark();	// mark time
    //
    i := 1 shl (6 - c);
    for t := 0 to 1 shl c - 1 do
      h[t] := CreateThread(nil, 0, @doJob, @i, 0, id);
    //
    WaitForMultipleObjects(1 shl c, @h, true, INFINITE);
    //
    elapsed(IntToStr(1 shl c) + ' threads: ');	// calculate time difference and print message
  end;
end.

[gusluk]

Сообщение от veniside

> Так вы и не сказали, меняют ли массив Fx, Fy, Fz или нет.

Не меняют.

[veniside]

> Не меняют.

тогда можно взять за основу исходник из поста #18, там есть и работа с массивом, и ожидание завершения работы всех потоков, и запуск от 1 до 64 потоков сразу. Код из главного тела программы выносим в отдельный поток, запускаем его в гуе и всё. Когда он пошлёт нам сообщение, отрисовываем график и уходим на новую итерацию, если нужно. При желании потом можно ограничить число одновременно работающих потоков, не создавать потоки заново. Но, думаю, на ваших объемах это даст не более 1% выигрыша, так что это скорее академический, чем практический интерес.

Update: в коде не хватает закрытия хендлов после остановки потоков:

for t := 0 to 1 shl c - 1 do
  CloseHandle(h[t]);