| Форумы
ЧАТ |
Орешник |
|
| Форумы -» | Основная («Начинающим»)/ Базы / WinAPI / Компоненты / Сети / Media / Игры / Corba и COM / KOL / FreePascal / .Net / Прочее / RSDN.ORG |
|
| ||
|
|
Новости |
Поиск |
FAQ |
Статьи |
Книги |
Новости VCL
| | Форумы ЧАТ |
Орешник |
|
| |||
| |||
| |||
« предыдущая ветвь | форум | следующая ветвь »
Организация многопоточности
Vitaliy_____ © (2009-07-03 08:24:00)Доброе время суток!
Озадачился вопросом распараллеливания алгоритма для повышения вычислительной производительности.
Есть модуль, реализующий искуственную нейросеть. В нем есть процедура DoTeach(), выполняющая обучение ИНС по модификации алгоритма обратного распространения ошибки.
В ней есть часто используемая процедура GetNetOut(), которая вычисляет значение выходов многослойной ИНС при заданных входах. Учитывая то, что коррекция весов происходит по завершении эпохи обучения, думаю будет достаточно распараллелить только GetNetOut, так как основная вычислительная нагрузка на ней.
Теперь ближе к вопросу.
Сама процедура сейчас выглядит так:
procedure TTeachNet.GetNetOut();
var
i,j,k,LN_1,LN_2: integer;
begin
LN_1:=LayersNum-1; LN_2:=LayersNum-2;
// Первый слой (входной), c активационной функции.
for i:=0 to LayerSize[0]-1 do
OutLayers[0,i]:=LayerActivateF(InpLayers[0,i],1);
// Рассчитываем все остальные слои кроме последнего.
for k:=1 to LN_2 do
for i:=0 to LayerSize[k]-1 do begin
InpLayers[k,i]:=0;
for j:=0 to LayerSize[k-1]-1 do
InpLayers[k,i]:=InpLayers[k,i]+OutLayers[k-1,j]*LayerWeights[k-1,i,j];
// Учет сдвига
InpLayers[k,i]:=InpLayers[k,i]+LayerWeights[k-1,i,LayerSize[k-1]]; // *1
OutLayers[k,i]:=LayerActivateF(InpLayers[k,i],1); // 1 - сигмоид
end;
// Рассчитываем последний слой c акт. ф. на выходе
for i:=0 to LayerSize[LN_1]-1 do begin
InpLayers[LN_1,i]:=0;
for j:=0 to LayerSize[LN_2]-1 do
InpLayers[LN_1,i]:=InpLayers[LN_1,i]+OutLayers[LN_2,j]*LayerWeights[LN_2,i,j];
// Учет сдвига
InpLayers[LN_1,i]:=InpLayers[LN_1,i]+LayerWeights[LN_2,i,LayerSize[LN_2]]; // *1
//OutLayers[LN_1,i]:=InpLayers[LN_1,i];
OutLayers[LN_1,i]:=LayerActivateF(InpLayers[LN_1,i],1);
end;
end;
Я вижу своей целью распараллелить циклы: каждый поток будет рассчитывать свой кусок цикла. По идее не сложно, но есть пара технических вопросов:
1) Нужно чтобы потоки не создавались при каждом вызове GetNetOut, а были созданы 1 раз при вызове DoTeach и по ее команде выполняли вычисление с новыми данными.
2) Массивы InpLayers, OutLayers, LayerWeights, LayerSize - это поля класса TTeachNet, и они необходимы для вычислений выхода ИНС. Как их лучше передавать, чтобы они были доступны, как глобальные переменные всем потокам? В принципе, есть экземпляр класса TTeachNet, и можно сделать эти переменные public (т.к. потоки планируется создавать и уничтожать при гарантированной жизни TTeachNet, его же методом, то ошибки не должно быть). Но, возможно, есть более хорошие идеи?
3) Также потребуется барьерная синхронизация между вычислениями различных слоев ИНС. Не силен я в этом, буклет или пример нужен...
4) Взаимодействия с VCL в этой части нет, с синхронизацией вроде проблем не будет по этому вопросу. Только распараллеливание вычислений, это должно облегчить задачу.
------------
Я понимаю, что мой вопрос в каком-то смысле звучит как "код давай", но что бы мне действительно хотелось:
- Ваше мнение, верны ли мои мысли на идейном уровне?
- пару примеров по техническим вопросам "а как это сделать", т.к. мне делфийских примеров на потоки показалось маловато, хотя бы барьерной синхронизации там нет, а она потребуется.
- подводные камни: тут нужен опыт, мало ли о чем не подумал, в многопоточности их много :)
------------
Прочитать если кто что напишет смогу не раньше вторника (7-го числа).
Vitaliy_____ (2009-09-10 05:41:00) [1]Я кажется наконец понял, что там могло быть:
InterlockedIncrement(Counter);
if Counter=NP then begin
ResetEvent(MBSEvent2);
Counter:=0;
State:=false;
SetEvent(MBSEvent1); // ПОСЛЕДНИМ шагом запустили все потоки
end else begin
repeat
key:=WaitForSingleObject(MBSEvent1,3000);
if key=WAIT_TIMEOUT then begin
MessageBox(0,PChar(WinAPIErrMsg4+#13+ThreadState),PChar(IntToStr(GetCurrentThreadID)),0);
DebugBreak;
end;
until key=WAIT_OBJECT_0;
end;
Похоже, ситуация такая:InterlockedIncrement(Counter);нам гарантирует, что Counter будет "честно" увеличен. А вот дальнейшее сравнениеif Counter=NPнебезопасно. Похоже получалось так: поток увеличивал счетчик и засыпал. После этого последний поток увеличивал счетчик и делал "свои дела", но до Counter:=0 не успевал еще дойти. Просыпался "уснувший" и, опа, он тоже считал себя последним, т.к. Counter=NP все еще истина. Вот и получаем ДВА последних потока. Уже ошибка, хотя событиям все равно сколько через них пройдет (кстати, когда делал на семафорах - там не все равно, потому и вылетало стабильно). Однако может быть так (хотя не спорю, достаточно редко):
У нас два "последних" потока. Один из них запускает все остальные, а второй спит. И если вдруг какой-то поток проснувшись дойдет до след. барьера (т.е. увеличит счетчик) и лишь после этого проснется "лишний последний поток" и выполнит counter:=0 - вот тогда и возникает ошибка. Похоже так. Естественно, вероятность ситуации повышается с увеличением числа потоков.
----------
Чувствую глубокое удовлетворения от познания дебрей ВинАПИ :)
Vitaliy_____ (2009-09-09 13:23:00) [2]Leonid Troyanovsky, [26]
> Могу заметить, чтобы написать и отладить не очень сложный
> составной объект синхронизации нужна масса времени и терпения.
Кажется придется согласиться :)
Код в [53] все еще не рабочий :) Как я понимаю проблему, у нас "защищен" Counter, но недостаточно защищена переменная state. Редко, но все-таки возникает ситуация "прорывания барьера". Долго не мог понять почему, т.к. возникает крайне редко - тяжело найти. Как раз та самая паршивая ситуация, когда 1 поток "засыпает на долго". Решил потестировать что будет при 4 потоках на двух ядрах, конкуренция возросла, стало появляться чаще. Ох уж ента штука... :)
Попробовал извернуться так:
...
MBSCrSect:TCriticalSection;
...
procedure TMyBarrierSinchronization.WaitBarrier(const ThreadState:string);
var key:cardinal;
WaitintEvent,ResetingEvent:THandle;
begin
MBSCrSect.Enter;
if state then begin // На нечетных барьерах ждем первое событие, выставляем второе
WaitintEvent:=MBSEvent1;
ResetingEvent:=MBSEvent2;
end else begin // На четных барьерах наоборот
WaitintEvent:=MBSEvent2;
ResetingEvent:=MBSEvent1;
end;
inc(Counter);
if Counter=NP then begin
if (ResetEvent(ResetingEvent)=false) then
raise Exception.Create(WinAPIErrMsg6);
Counter:=0;
State:=(state=false);
SetEvent(WaitintEvent); // запустили все остальные потоки
end;
MBSCrSect.Leave; // Все критичные данные в локальных переменных, гонка не возможна.
repeat
key:=WaitForSingleObject(WaitintEvent,7000);
if key=WAIT_TIMEOUT then begin
MessageBox(0,PChar(WinAPIErrMsg4+#13+ThreadState),PChar(IntToStr(GetCurrentThreadID)),0);
DebugBreak;
end;
until key=WAIT_OBJECT_0;
end;
Суслик_ (2009-08-28 14:17:00) [3]Если физический проц один, то только потеряешь производительность, а не получишь ее.
Vitaliy_____ (2009-08-27 09:05:00) [4]
> У WaitForSingleObject есть еще WAIT_FAILED.
Да. Обработки ошибки ожидания тут нет... Можно добавить.
> А для 2 процессов и InterlockedIncrement не годится.
Да точно, мне в принципе для разных не надо. Хотя чуток напильником поработать и можно будет для разных использовать (взять тот же оптекс у Рихтера за основу).
> MessageBox я б заменил на OutputDebugString
Вот тут не согласен - MessageBox удобен тем, что блокирует поток на время - я вижу на экране несколько сообщений от разных потоков, в нужном давлю ок и через DebugBreak попадаю в отладку именного этого потока (то есть хотя сообщения возникнуть могут не все сразу, я могу подождать и отлаживать именно интересующий меня поток).
Leonid Troyanovsky (2009-08-26 14:37:00) [5]
> Vitaliy_____ (26.08.09 12:59) [53]
> if GetLastError<>0 then raise Exception.Create(WinAPIErrMsg2);
Win32Check(MBSEvent1<>0);
MessageBox я б заменил на OutputDebugString, смотреть можно
в дельфийском EventLog.
У WaitForSingleObject есть еще WAIT_FAILED.
> т.к. в рамках 1 процесса можно и неименованные события
А для 2 процессов и InterlockedIncrement не годится.
--
Regards, LVT.
Vitaliy_____ (2009-08-26 12:59:00) [6]Переписал модуль, да обрадуется Leonid Troyanovsky, 2 события для синхронизации достаточно (ну еще счетчик с InterlockedIncrement) :) Пока работает, кому интересно можно потестировать.
Повторюсь, суть этой штуки - барьер с автосбросом - то есть несколько потоков могут "легко и непринужденно" собираться в контрольной точке (вызывая WaitBarrier), при этом когда они дождутся следующий барьер будет автоматически поставлен. Изначально необходимо только проинициализировать число потоков, на которые выставляется барьер.
Если кто протестирует - отпишитесь.
unit BarrierSinchronization;
interface
uses Windows, Utils, SysUtils, SyncObjs;
type TMyBarrierSinchronization=class(TObject)
private
NP:integer; // Число потоков для синхронизации
MBSEvent1,MBSEvent2:THandle; // 2 события для реализации барьера
EvName1,EvName2:String; // Имена событий
state:boolean; // Ключ, с каким событием работаем
Counter: integer; // Внутренний счетчик выполненности заданий
public
procedure Initialize(NumProcesses:integer);
procedure WaitBarrier(const ThreadState:string); overload; // Отладочная версия
procedure WaitBarrier(); overload;
constructor Create;
destructor destroy; override;
end;
var MyBarrierSinchronization:TMyBarrierSinchronization;
resourcestring
WinAPIErrMsg2='Can not create event 1 for synchronization!';
WinAPIErrMsg3='Can not create event 2 for synchronization!';
WinAPIErrMsg4='WaitTimeOut on Event 1!';
WinAPIErrMsg5='WaitTimeOut on Event 2!';
implementation
constructor TMyBarrierSinchronization.Create;
begin
EvName1:=MyUtils.GetUnicalName+'MBSMyEvent';
EvName2:=EvName1+'2';
EvName1:=EvName1+'1';
// Создаем события
MBSEvent1:=CreateEvent(nil,true,false,PChar(EvName1)); // Событие с ручным сбросом, изначально занято
if GetLastError<>0 then raise Exception.Create(WinAPIErrMsg2);
MBSEvent2:=CreateEvent(nil,true,false,PChar(EvName2)); // Событие с ручным сбросом, изначально занято
if GetLastError<>0 then raise Exception.Create(WinAPIErrMsg3);
end;
destructor TMyBarrierSinchronization.destroy;
begin
if MBSEvent1<>0 then
CloseHandle(MBSEvent1);
if MBSEvent2<>0 then
CloseHandle(MBSEvent2);
inherited;
end;
procedure TMyBarrierSinchronization.Initialize(NumProcesses:integer);
begin
NP:=NumProcesses;
Counter:=0;
ResetEvent(MBSEvent1);
ResetEvent(MBSEvent2);
State:=true;
end;
procedure TMyBarrierSinchronization.WaitBarrier(const ThreadState:string);
var key:cardinal;
begin
if state then begin // На нечетных барьерах ждем первое событие, выставляем второе
InterlockedIncrement(Counter);
if Counter=NP then begin
ResetEvent(MBSEvent2);
Counter:=0;
State:=false;
SetEvent(MBSEvent1); // ПОСЛЕДНИМ шагом запустили все потоки
end else begin
repeat
key:=WaitForSingleObject(MBSEvent1,3000);
if key=WAIT_TIMEOUT then begin
MessageBox(0,PChar(WinAPIErrMsg4+#13+ThreadState),PChar(IntToStr(GetCurrentThreadID)),0);
DebugBreak;
end;
until key=WAIT_OBJECT_0;
end;
end else begin // На четных барьерах наоборот
InterlockedIncrement(Counter);
if Counter=NP then begin
ResetEvent(MBSEvent1);
Counter:=0;
State:=true;
SetEvent(MBSEvent2);
end else begin
repeat
key:=WaitForSingleObject(MBSEvent2,3000);
if key=WAIT_TIMEOUT then begin
MessageBox(0,PChar(WinAPIErrMsg5+#13+ThreadState),PChar(IntToStr(GetCurrentThreadID)),0);
DebugBreak;
end;
until key=WAIT_OBJECT_0;
end;
end;
end;
procedure TMyBarrierSinchronization.WaitBarrier();
var key:cardinal;
begin
if state then begin // На нечетных барьерах ждем первое событие, выставляем второе
InterlockedIncrement(Counter);
if Counter=NP then begin
ResetEvent(MBSEvent2);
Counter:=0;
State:=false;
SetEvent(MBSEvent1); // ПОСЛЕДНИМ шагом запустили все потоки
end else begin
key:=WaitForSingleObject(MBSEvent1,INFINITE);
end;
end else begin // На четных барьерах наоборот
InterlockedIncrement(Counter);
if Counter=NP then begin
ResetEvent(MBSEvent1);
Counter:=0;
State:=true;
SetEvent(MBSEvent2);
end else begin
key:=WaitForSingleObject(MBSEvent2,INFINITE);
end;
end;
end;
initialization
MyBarrierSinchronization:=TMyBarrierSinchronization.Create;
end.
// Utils - модуль с утилитами, из него дергается только GetUnicalName - на основе времени создает уникальную текстовую прибавку к имени - можно убрать, т.к. в рамках 1 процесса можно и неименованные события использовать.
Vitaliy_____ (2009-08-19 06:09:00) [7]А я даже все статьи прочитал... Это было интересно - но если я не ошибаюсь, для корных процов проблем нет. У меня Core 2 duo e6750 @ 2.66GHz. Да и личный замер, как я писал в [33] более чем убеждает в этом. Хотя у меня в руках теперь хороший тестер, могу сам проверить на других :) Только ждать долго - хорошее разделение нагрузки только на реальной (читать большой) задаче - а ее потом для сравнения надо еще и на однопоточной версии проги прогнать.
Leonid Troyanovsky (2009-08-18 14:45:00) [8]
> Vitaliy_____ (18.08.09 13:51) [50]
> А если у нас появляется новый проц с 4 ядрами, то 4 event?
Я ж ссылочкой снабжал, насчет эффективной многопоточности.
--
Regards, LVT.
Vitaliy_____ (2009-08-18 13:51:00) [9]А если у нас появляется новый проц с 4 ядрами, то 4 event? Нет, не разубедил относительно семафоров.
Leonid Troyanovsky (2009-08-18 13:35:00) [10]
> Vitaliy_____ (18.08.09 12:12) [48]
> А если все же освободит и уснет в момент проверки while
> not Application.Terminated do?
Если вторичный поток вытеснен, то первичный узнает об этом
сравнив ID & calcid, ну и вторую часть ему придется считать самому.
> А по поводу спин-блокировки, у рихтера это "Худшее что можно
> сделать"...
Контекст непонятен. Ладно, посмотрим на досуге.
> Пусть попадают в режим ядра, у меня при большой задаче не
> такие частые синхронизации
Два Event с автосбросом, первичный поток сигналит одним для старта очередной итерации и ждет сигнал готовности от вторичного.
Ну и, конечно, внутри процесса имена объектам не нужны.
--
Regards, LVT.
Vitaliy_____ (2009-08-18 12:12:00) [11]
> Если вторичный поток будет, например, вытеснен не осводив
> КС,
> то первичный, досчитав свой цикл, будет ожидать ее освобождения.
>
> Для гуева потока это может и не очень, но я на гуи и не
> настаивал.
А если все же освободит и уснет в момент проверки while not Application.Terminated do?
Согласен, редкий случай, но возможный. В моем примере тоже по 10-20 итераций делает (то есть по 30 000 образов, в них по 4 барьера) и это * число итераций, а потом вдруг умирает. К счастью, стабильно :)
Нет, должно работать 100%.
А по поводу спин-блокировки, у рихтера это "Худшее что можно сделать"... стр. 195 (215 в моей электронной версии). Сам ее "придумал", потом прочитал, проникся и больше так не делаю :)
Пусть попадают в режим ядра, у меня при большой задаче не такие частые синхронизации, и можно пренебречь 1000 тактов ради того чтобы прога не мешала другим. А вот о "надежности" барьеров - придется самому разбираться, как время найду - хочу добить задачку, может потом статью накатаю :-[c]
Leonid Troyanovsky (2009-08-18 10:12:00) [12]
> Vitaliy_____ (18.08.09 08:29) [46]
> Идею твоей программы понял, там кстати целые числа - вполне
> подошел бы InterlockedExchange, твоя идея для вещественных
Да хоть для комплексных ;)
> Где гарантия, что порожденный поток не уснет надолго? (скажем,
> на час :) ) -
Эту гарантию должен дать разработчик.
> тогда "главный" просчитает часть цикла, отпишется через
> критическую секцию и начнет считать дальше, ведь так?
Если вторичный поток будет, например, вытеснен не осводив КС,
то первичный, досчитав свой цикл, будет ожидать ее освобождения.
Для гуева потока это может и не очень, но я на гуи и не настаивал.
> Только по сути это будет спин-блокировка, что крайне не
> рекомендуется тем же Рихтером.
Чего-то я такой рекомендации не помню.
Спин-блокировка - самый быстрый механизм синхронизации.
При использовании же КС потоки, все же, попадают в режим ядра.
--
Regards, LVT.
Vitaliy_____ (2009-08-18 08:29:00) [13]
> Leonid Troyanovsky ©
Идею твоей программы понял, там кстати целые числа - вполне подошел бы InterlockedExchange, твоя идея для вещественных лучше. Идея плохая :) Где гарантия, что порожденный поток не уснет надолго? (скажем, на час :) ) - тогда "главный" просчитает часть цикла, отпишется через критическую секцию и начнет считать дальше, ведь так? Просчитает ЕЩЕ ОДНУ часть цикла и снова увеличит ct.calcid := ct.calcid + 1;
При мелких вычислениях - очень реальная ситуация. Что при этом будет - а также как у меня где-то проскок через контрольную точку. Хотя, если еще одну контрольную ID-шку добавить, подход сработает. Только по сути это будет спин-блокировка, что крайне не рекомендуется тем же Рихтером.
Поправьте, если не прав...
Leonid Troyanovsky (2009-08-17 21:12:00) [14]
> Kolan © (17.08.09 16:15) [41]
> А чем принципиально исользование TRTLCriticalSection отличается
> от использования TCriticalSection?
Принципиально, наверное, не отличается.
Как-то обвык без объектов и Enter|Leave :(
--
Regards, LVT.
Дмитрий Белькевич (2009-08-17 18:03:00) [15]Ясно, буду разбираться.
Kolan (2009-08-17 16:29:00) [16]Ну наследования структур нет. А кроме того, как наследованием добавить методы в середину бибилиотечной иерархии? — Только исправив библиотеку. А бибилиотека может быть и без исходников, кроме того VCL какой-нибудь править — не самое безопасное занятие.
А тут раз:helper for TComponentнаписал и все.
Дмитрий Белькевич (2009-08-17 16:26:00) [17]
> record helper for
Тоже увидел. Вопрос, кстати, образовался. Чем эти хэлперы отличаются от обычного наследования?
Kolan (2009-08-17 16:15:00) [18]Благодарю.
А чем принципиально исользованиеTRTLCriticalSectionотличается от использованияTCriticalSection?
Кстати в 2009 Делфи для этого типа есть два хелпера:
TCriticalSectionHelper = record helper for TRTLCriticalSection
procedure Initialize; inline;
procedure Destroy; inline;
procedure Free; inline;
procedure Enter; inline;
procedure Leave; inline;
function TryEnter: Boolean; inline;
end;
TConditionVariableHelper = record helper for TRTLConditionVariable
public
class function Create: TRTLConditionVariable; static;
procedure Free; inline;
function SleepCS(var CriticalSection: TRTLCriticalSection; dwMilliseconds: DWORD): Boolean;
procedure Wake;
procedure WakeAll;
end;
Leonid Troyanovsky (2009-08-17 13:29:00) [19]
> Kolan © (17.08.09 13:15) [39]
> Что такое TRTLCriticalSection? В справке нет.
From Turbo Delphi for Win32 help:
TRTLCriticalSection stores information about a Windows critical section.
From Windows.pas (find declaration):
TRTLCriticalSection = _RTL_CRITICAL_SECTION; // &etc.
--
Regards, LVT.
Kolan (2009-08-17 13:15:00) [20]Что такое
T RTLCriticalSection? В справке нет.
| |
Страницы: 