Как разрулить работу потока в вызовах его методов?

← →
Aleksandr. (2005-02-11 15:12) [0]

Есть класс-наследник от TThread, имеющий свойство FList TList, в котором содержатся объекты с информацией для выполнения работы, метод AddQuery для добавления этих объектов в List, и метод,который работает внутри Execute:
type TQueryThread = class (TThread) private FList : TList; FListCS : TCriticalSection; procedure DoQuery(aQuery : TExecQuery); public procedure AddQuery(aQuery : TExecQuery); procedure Execute; override; ... end; procedure TQueryThread.AddQuery(aQuery : TExecQuery); begin FListCS.Acquire; try FList.Add(aQuery) finally FListCS.Release end end; procedure TQueryThread.Execute; var Q : TExecQuery; begin repeat while GetQueryCount>0 do begin // защищенный FListCS FList.Count FListCS.Acquire; try Q:=TExecQuery(FList.Items[0]); FList.Delete(0) finally FListCS.Release end; DoQuery(Q) end; Suspend until Terminated end;
Код DoQuery приводить не буду, он очень большой и длительный по исполнению.
Есть также менеджер потоков TQueryThread, который, получая список из TExecQuery, раздает его по потокам:
procedure TQueryManager.ProcessQueries(L : TList); var i, j : integer; begin for i:=0 to L.Count-1 do begin for j:=0 to FQueryThreadList.Count-1 do if FQueryThreadList.Items[j].ID=TExecQuery(L.Items[i]).ID then begin FQueryThreadList.Items[j].AddQuery(TExecQuery(L.Items[i])); if FQueryThreadList.Items[j].Suspended then FQueryThreadList.Items[j].Resume; Break end end end;

Вот вся эта схема время от времени дает длительные затыки. Лог выполнения показывает, что происходит примерно следующее:
Добавляется объект запроса в нить, она входит в метод DoQuery,
в это время добавляется еще один объект, и нить прекращает выполнять DoQuery. И обрабатывает все полученные объекты только тогда, когда перестанет вызываться AddQuery, возникнет пауза, а затем добавится еще один объект, добавление которого вызовет резюмирование нити. Может ли такое быть и как это можно разрулить? Догадываюсь, что есть вариант с блокировкой критической секцией между методами AddQuery и DoQuery, но он вызывает у меня сомнения, т.к. теряется смысл в FList, а метод DoQuery очень длительный.

← →
Digitman © (2005-02-11 15:23) [1]

первое же что коробит глаз - между GetQueryCount и FListCS.Acquire состояние списка FListCS может измениться

← →
Aleksandr. (2005-02-11 15:35) [2]

Digitman © :
То есть это все должно быть в критической секции? Но это же не критично - обращение идет к первому элементу FList а не последнему? Хотя на 2-процессорной машине, может, и имеет...

← →
Digitman © (2005-02-11 15:47) [3]

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

← →
Aleksandr. (2005-02-11 15:54) [4]

Digitman © :
Не понял... Если Вы о FList, то он же является свойством нити и используется только в ее методах? Кем же он еще используется?

← →
Aleksandr. (2005-02-11 18:10) [5]

Истчо вопрос на эту тему рождается: если нить находится в процессе выполнения Execute, а ей кто-нибудь вызывает resume - что произойдет с текущего выполняющегося метода? Он просто сбросится и нить вернется в начало цикла Execute?

← →
Eraser © (2005-02-11 19:42) [6]

Aleksandr
если нить находится в процессе выполнения Execute, а ей кто-нибудь вызывает resume - что произойдет с текущего выполняющегося метода?

Этот метод продолжет выполнение...

← →
Leonid Troyanovsky © (2005-02-11 20:54) [7]

> Aleksandr. (11.02.05 15:12)
> Есть класс-наследник от TThread, имеющий свойство FList
> TList, в котором содержатся объекты с информацией для

TThreadList.

--
Regards, LVT.

← →
Alexander Panov © (2005-02-11 21:51) [8]

Как часто у тебя вызывается
procedure TQueryManager.ProcessQueries(L : TList);
?

← →
Aleksandr. (2005-02-11 23:17) [9]

Leonid Troyanovsky © :
Я, если честно, каких-то преимуществ TThreadList"a перед просто List"ом не увидел. Потому и не стал изгаляться.

Alexander Panov © :
Достаточно часто. В пиковые часы загрузки может и каждые 10 секунд вызываться.

← →
Leonid Troyanovsky © (2005-02-12 01:11) [10]

> Aleksandr. (11.02.05 23:17) [9]

> Я, если честно, каких-то преимуществ TThreadList"a перед
> просто List"ом не увидел. Потому и не стал изгаляться.

TThreadList защищает свой список критической секцией.
Так что, более ничего и не требуется.

--
Regards, LVT.

← →
Aleksandr. (2005-02-12 02:17) [11]

Leonid Troyanovsky © :
Ну, все равно какое-то разруливание требуется, наверное. Пусть я заменю код на
procedure TQueryThread.AddQuery(aQuery : TExecQuery); begin FList.LockList.Add(aQuery); FList.UnlockList end; procedure TQueryThread.Execute; var Q : TExecQuery; begin repeat while try Q:=TExecQuery(FList.LockList.Items[0]); FList.UnLockList; FList.LockList.Delete(0); FList.UnLockList DoQuery(Q) end; Suspend until Terminated end; Что это принципиально изменит?

← →
Aleksandr. (2005-02-12 02:18) [12]

Млин...

← →
Leonid Troyanovsky © (2005-02-12 03:29) [13]

> Aleksandr. (12.02.05 02:17) [11]

> procedure TQueryThread.AddQuery(aQuery : TExecQuery);
> begin
> FList.LockList.Add(aQuery);
> FList.UnlockList
> end;

Просто FList.Add(aQuery).
И

procedure TQueryThread.Execute;
var
Q : TExecQuery;
begin
while not Terminated do
begin
with FList.Lock do
try
if Count = 0 then
Break; // завершаемся или выход на ожидание(?)
Q:=TExecQuery(Items[0]);
Delete(0);
finally
FList.UnLockList;
end;
DoQuery(Q); // внешние try опущены
Q.Free; // Возможно, что требует Synchronize(..)
// Suspend;
end;
end;

И что за Suspend? Ecли нужна реакция на событие,
то WaitFor*Object, можно и без Terminated.

> Что это принципиально изменит?

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

--
Regards, LVT.

← →
Aleksandr. (2005-02-12 03:39) [14]

Leonid Troyanovsky © :
Спасибо за разъяснения! Только я со Suspend не понял. Мне не надо завершаться, мне надо переходить в ожидание, когда будет добавлен следующий запрос. Вы имеете в виду, что надо в констракторе создать эвент, а по выполнении DoQuery переходить в инфинитный WaitForSingleObject установки этого эвента?
А что до Q.Free, так этого вообще тут не надо, как я понимаю - объект создается в другом потоке, содержит в себе событие, которое надо установить по окончании DoQuery, и в нем же уничтожится, этого события (точнее, их комплекта от всех объектов переданного менеджеру в ProcessQueries списка) поток-создатель ожидает в WaitForMultipleObjects, за которым дестракторы и вызовет.

← →
Leonid Troyanovsky © (2005-02-12 04:43) [15]

> Aleksandr. (12.02.05 03:39) [14]
> Спасибо за разъяснения! Только я со Suspend не понял. Мне
> не надо завершаться, мне надо переходить в ожидание, когда
> будет добавлен следующий запрос. Вы имеете в виду, что надо
> в констракторе создать эвент, а по выполнении DoQuery переходить
> в инфинитный WaitForSingleObject установки этого эвента?

Например, делаем EventDoit с ручным сбросом и

procedure TQueryThread.Execute;
..
begin
while WaitForSingleObject(EventDoit, INFINITE) = WAIT_OBJECT_0 do
begin
if Terminated then
Break;
Q := nil;
with FList.Lock do
try
if Count = 0 then
ResetEvent(EventDoit)
else
begin
Q:=TExecQuery(Items[0]);
Delete(0);
end;
finally
FList.UnLockList;
end;
if Assigned(Q) then
DoQuery(Q);
end;
end;

Тогда, при добавлении в список нужно делать SetEvent.
Также SetEvent в Terminate:

SetEvent(EventDoit);
inherited;

и

destructor TQueryThread.Destroy; // override;
begin
Terminate; // т.к. он не виртуальный
inherited;
end;

> А что до Q.Free, так этого вообще тут не надо, как я понимаю
> - объект создается в другом потоке, содержит в себе событие,
> которое надо установить по окончании DoQuery, и в нем же
> уничтожится, этого события (точнее, их комплекта от всех
> объектов переданного менеджеру в ProcessQueries списка)
> поток-создатель ожидает в WaitForMultipleObjects, за которым
> дестракторы и вызовет.

Проще всего, когда эти объекты созданы в первичном (VCL)
потоке.
Тогда уничтожать их можно передавая указатель в Post(Thread)
Message окну или в Application.OnMessage.
Т.е., специфические ожидания и не потребуются.

--
Regards, LVT.

← →
Alex Konshin © (2005-02-12 05:29) [16]

Leonid Troyanovsky © (12.02.05 04:43) [15]
Неверно.

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

← →
Alex Konshin © (2005-02-12 05:49) [17]

Хотя да... У тебя вместо простой проверки в цикле наличия запросов,
делается замысловатые пассы с event с ручным сбросом.
Совершенно излишнее усложнение. Лишние обращения к системе - лишние накладные расходы, к тому и сложнее получается. Достаточно простого event с автоматом и двойной цикл while.
Еще вместо TList я бы предложил использовать связный список.

← →
Leonid Troyanovsky © (2005-02-12 11:01) [18]

> Alex Konshin © (12.02.05 05:49) [17]

> Хотя да... У тебя вместо простой проверки в цикле наличия
> запросов,
> делается замысловатые пассы с event с ручным сбросом.
> Совершенно излишнее усложнение. Лишние обращения к системе
> - лишние накладные расходы, к тому и сложнее получается.

Просто такая схема обеспечит возможность добавления в список
после любого DoQuery. И реакцию на Terminate (не уверен, что
проверки Terminated внутри цикла достаточно для
мультипроцессорных систем).
Ведь, по условию, DoQuery длительная операция.

> Достаточно простого event с автоматом и двойной цикл while.
> Еще вместо TList я бы предложил использовать связный список.

Можно вообще без списка - хватит очереди APC потока.
Но, видимо, в оптимизации здесь нуждается тот самый DoQuery.

--
Regards, LVT.

← →
Alex Konshin © (2005-02-12 11:26) [19]

Вставлять в твоем случае можно в любой момент кроме try...finally.
Также, как и в моем случае.
Насчет APC - опять-таки ненужное усложнение.
Для организации списка в моем случае достаточно одного поля в TExecQuery и пару переменных-указателей на первый элемент и на последнее поле связи.
Просто не понимаю, зачем извращаться, когда все прямолинейно и ясно? Тем более, что это и эффективнее. Не, можно, конечно, и очередь Windows задействовать, и еще чего-нибудь, только зачем?

← →
Leonid Troyanovsky © (2005-02-12 12:26) [20]

> Alex Konshin © (12.02.05 11:26) [19]

> Вставлять в твоем случае можно в любой момент кроме try...finally.
> Также, как и в моем случае.

Да, конечно, со вставкой был не прав.
Еще не проснулся, sorry.

> Насчет APC - опять-таки ненужное усложнение.

APC один из самых эффективных механизмов.
Широко исползуется в системе- для передачи одного параметра -
прямо, для большего - вместе с memory mapped file (секцией).
Да и зачем городить свою очередь, когда есть готовая.
И выглядит все очень понятно: посылаем запрос рабочему потоку
- выполнить DoQuery, DoQuit и т.д.
А рабочий поток просто сидит в SleepEx(INFINITE, True).
Ну, конечно, для потоков-объектов нужно сделать threadvar с
ссылкой на себя, или передавать эту ссылку в TExecQuery.

> Просто не понимаю, зачем извращаться, когда все прямолинейно
> и ясно? Тем более, что это и эффективнее. Не, можно, конечно,
> и очередь Windows задействовать, и еще чего-нибудь, только
> зачем?

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

--
Regards, LVT.

← →
Набережных С. © (2005-02-12 14:04) [21]

>Leonid Troyanovsky © (12.02.05 12:26) [20]
>APC один из самых эффективных механизмов.

Имхо, эффективность у него точно такая же, как и у очередей сообщений, а вот работать с ним менее удобнее.

>Alex Konshin © (12.02.05 11:26) [19]
>Не, можно, конечно, и очередь Windows задействовать, и еще чего-нибудь, только зачем?

Использование очереди сообщений, как и APC, позволит отказаться и от TList, и от критических секций, и от эвентов.

← →
Alexander Panov © (2005-02-12 14:57) [22]

Aleksandr. (11.02.05 15:12)

Может быть, натолкнет на какие-либо мысли вот эта реализация твоей задачи:
http://home.ural.ru/~panov/projects/threadspool/threadspool.html

← →
Eraser © (2005-02-12 15:18) [23]

Aleksandr.

Также в коллекции JEDI компонентов, есть целая вкладка для работы с потоками, по-моему то что надо.

Я бы сделал объект очередь. У нее public полем event с ручным сбросом.
Ее методы add, extractTop, getcount защищены критической секцией.

add
<Вход в КС>
добавляет новый элемент в конец очереди и взводит event (setEvent)
<Выход из КС>

extractTop
<Вход в КС>
Извлекает указатель на верхний элемент (Result:=), удаляет этот элемент и если очередь пуста, сбрасывает event (ResetEvent).
<Выход из КС>

Появления элементов в очереди ожидаем WaitFor... ф-циями

Queue : TQueue;

while WaitForSingleObject(Queue.Event, INFINITE) = WAIT_OBJECT_0 do begin Query := Queue.extractTop(); DoQuery(Query); end;

← →
Aleksandr. (2005-02-14 12:34) [25]

Ох... спасибо всем... Теперь буду пытаться понять все это... Особенно про ResetEvent. Я уже не первый год юзаю систему CreateEvent-SetEvent-CloseHandle, но впервые узнаю, что Event надо еще и сбрасывать. Что происходит в данном коде, если (не) вызывается ResetEvent?

> Aleksandr. (14.02.05 12:34) [25]
> Ох... спасибо всем... Теперь буду пытаться понять все это...
> Особенно про ResetEvent. Я уже не первый год юзаю систему
> CreateEvent-SetEvent-CloseHandle, но впервые узнаю, что
> Event надо еще и сбрасывать. Что происходит в данном коде,
> если (не) вызывается ResetEvent?

Чаще используется CreateEvent with bManualReset=False.
А в этом случае событие сбрасывается автоматически тогда,
когда его дождался любой поток (который его WaitFor*).

В упомянутом коде ручной сброс предлагался для обеспечения
быстрого отклика на Terminate (or Free) потока.
Иначе, поток завершится лишь после исчерпания очереди
(я не очень уверен в действенности проверки Terminated = true
в многопроцессорных системах).

--
Regards, LVT.

(я не очень уверен в действенности проверки Terminated = true
в многопроцессорных системах).

А какая разница? Это всего лишь переменная.

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

> Alex Konshin © (14.02.05 22:11) [27]
> (я не очень уверен в действенности проверки Terminated =
> true
> в многопроцессорных системах).
>
> А какая разница? Это всего лишь переменная.

А, действительно... Уговорил :)
Внешний - while & Event с автосбросом,
внутренний - while not Terminated.

--
Regards, LVT.

← →
Aleksandr. (2005-02-17 16:44) [29]

Ну ничего у меня не выходит, млин! Какие странные ступоры выплывают. Мало того, по условию теперь в свойства нити необходимо добавить еще один List для объектов другого типа, да еще и булевое свойство для третьего вида обработки. Я попытался сделать их как интерпретацию объектов первого вида:
в новом методе в FUpdList добавляются объекты другого типа, после чего создается TExecQuery с предопределенными свойствами и вызывается опять же AddQuery. Ну а в обработчике DoQuery в зависимости от свойств рассматриваемого TExecQuery вызывается тот или иной метод. Все это вроде бы и работает, но иногда все равно происходит нестыковка - менеджер добавляет запросы, а потоки их уже не обнаруживают. Удалось только снизить вероятность такого затыка защитой одной критической секцией методов AddQuery и всего содержимого участка while WaitForSingleObject...end, что опять же привело к неприятному эффекту - все потоки, размещающие через менеджер потоков свои запросы, вынуждены дожидаться разблокирования секций на период весьма не быстрой работы DoCurrentQuery. Код теперь выглядит так
В менеджере: // этот метод добавляет нитям в очередь запросы procedure TQueryManager.ProcessNewQuery(Q: TExecQueryList); var i : integer; A : TQueryThread; T : smallint; begin FResetCS.Enter; try T:=-1; try for i:=0 to Q.QueryCount-1 do begin try T:=Q.Items[i].qID; A:=Threads[T]; // метод возвращает нужную нить if Assigned(A) then A.AddQuery(Q.Items[i]) else begin DoLog(Format("Не удалось найти обработчик для %d, ручной сброс ожидания",[T])); Q.Items[i].qHandled:=true; SetEvent(Q.Items[i].qHandle) end except on E:Exception do begin DoLog(Format("Не удалось найти обработчик для %d "+E.Message,[T])); Q.Items[i].qHandled:=true; SetEvent(Q.Items[i].qHandle) end end end except on E:Exception do LogDBError(T,"Process new query: "+E.Message) end finally FResetCS.Leave end end; // этот метод добавляет нитям в очередь списки объектов для обработки другого типа procedure TQueryManager.ProcessATIUpdate(IDT: word; EvList: TATIUpdateList); var A : TQueryThread; begin FResetCS.Enter; try if EvList.Count>0 then begin A:=Threads[IDT]; if Assigned(A) then A.AddUpdates(EvList) else EvList.Clear end finally FResetCS.Leave end end; в нити: // этот метод добавляет в свой список элемент запроса procedure TQueryThread.AddQuery(Query: TExecQuery); begin DoLog(Format("Запрос добавлен для %d",[Query.qUserID])); FExecCS.Acquire; try DoLog("Добавление запроса"); FQueryList.Add(Query); SetEvent(FExecEvent) finally FExecCS.Release end end; // этот метод добавляет в список объектов пакет объектов для одного использования procedure TQueryThread.AddUpdates(EvList: TATIUpdateList); var Q : TExecQuery; begin if NOT ((NOT Assigned(EvList)) OR (EvList.Count=0) OR NOT Assigned(FUpdateList)) then begin FUpdateCS.Enter; try while EvList.Count>0 do begin FUpdateList.Add(EvList.Items[0].GetCopy); EvList.Delete(0) end finally FUpdateCS.Leave end; Q:=TExecQuery.Create(0,qiOtherInfo,0,0); AddQuery(Q) end end; procedure TQueryThread.Execute; var Q : TExecQuery; begin try Coinitialize(nil); try FSQLUpdator:=TATISQLUpdator.Create(FIDT); try LoadCashedData; CalculateMemBuffers; CalCulateIndexes; while WaitForSingleObject(FExecEvent,INFINITE)=WAIT_OBJECT_0 do try if Terminated then Break; FExecCS.Acquire; try CheckFileIndex; with FQueryList.LockList do try while Count>0 do begin Q:=Items[0]; if Q.qType=qiOtherInfo then begin case Q.qID of 0 : ProcessUpdates; 1 : ProcessClearing; end; Q.Free // убиваем запрос, потому как создан самой нитью end else DoQuery(Q); // иначе запрос будет убит потоком, который его создал и передал менеджеру Delete(0) end finally FQueryList.UnlockList end; // ReSetEvent(FExecEvent) // вроде как не надо, потому что создан =CreateEvent(nil, false, False, PChar(IntToStr(Self.ThreadID)+"thread")); finally FExecCS.Release end except on E:Exception do DoLog("Ошибка цикла, "+E.Message) end finally FreeAndNil(FSQLUpdator) end finally CoUninitialize end except on E:Exception do DoLog("Ошибка Execute "+E.Message) end end;

А мож примерно так, а?
const SN_QUERY = WM_USER + 400; SN_QUIT = SN_QUERY + 1; type TExecQuery = class // Stub end; TQueryThread = class(TThread) private procedure DoQuery(aQuery : TExecQuery); protected procedure Execute; override; public procedure AddQuery(aQuery : TExecQuery); procedure Terminate; end; implementation { TQueryThread } procedure TQueryThread.AddQuery(aQuery: TExecQuery); begin PostThreadMessage(ThreadID, SN_QUERY, integer(aQuery), 0); end; procedure TQueryThread.DoQuery(aQuery: TExecQuery); begin // Stub end; procedure TQueryThread.Execute; var Msg: TMsg; begin while GetMessage(Msg, 0, 0, 0) do begin case Msg.message of SN_QUERY: DoQuery(TExecQuery(Msg.wParam)); SN_QUIT: begin while PeekMessage(Msg, 0, SN_QUERY, SN_QUERY, PM_REMOVE) do TExecQuery(Msg.wParam).Free; Break; end; end; end; end; procedure TQueryThread.Terminate; begin PostThreadMessage(ThreadID, SN_QUIT, 0, 0); inherited; end;

я бы даже так вот так сделал это :
TQueryThread = class(TThread) private procedure DoQuery(aQuery : TExecQuery); protected procedure Execute; override; public destructor Destroy; override; procedure AddQuery(aQuery : TExecQuery); end; .. destructor TQueryThread.Destroy; begin if GetCurrentThreadId <> MainThreadId then PostThreadMessage(ThreadId, WM_QUIT, 0, 0); inherited; end; procedure TQueryThread.AddQuery(aQuery: TExecQuery); begin PostThreadMessage(ThreadID, SN_QUERY, integer(aQuery), 0); end; procedure TQueryThread.DoQuery(aQuery: TExecQuery); begin // Stub end; procedure TQueryThread.Execute; var Msg: TMsg; begin try try while not Terminated and GetMessage(Msg, 0, 0, 0) do Dispatch(Msg.Message); finally while PeekMessage(Msg, INVALID_HANDLE_VALUE, SN_QUERY, SN_QUERY, PM_REMOVE) do try TExecQuery(Msg.wParam).Free; except end; end; except end; end;

да, забыл ..

TQueryThread = class(TThread) private procedure MsgQuery(var Message: TMessage); message SN_QUERY; procedure DoQuery(aQuery : TExecQuery); protected procedure Execute; override; public destructor Destroy; override; procedure AddQuery(aQuery : TExecQuery); end; .. procedure TQueryThread.MsgQuery(var Message: TMessage); var ExecQuery: TExecQuery; begin ExecQuery := TExecQuery(Message.wParam); try DoQuery(ExecQuery); finally ExecQuery.Free; end; end;

> Digitman © (18.02.05 09:08) [31]

Конечно нужно делать так, как у тебя:) Мой пример и кодом-то назвать нельзя:) Эскиз, набросанный за 5 минут, в котором полно ошибок, как логических, так и технических

Как разрулить работу потока в вызовах его методов? Найти похожие ветки