| Новости | Поиск | FAQ | Статьи | Книги | Новости VCL | | Форумы ЧАТ | Орешник | KOL и MCK |
Форумы -» Основная («Начинающим»)/ Базы / WinAPI / Компоненты / Сети / Media / Игры / Corba и COM / KOL / FreePascal / .Net / Прочее / RSDN.ORG

	« предыдущая ветвь | форум | следующая ветвь »

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

Страницы:   < 1 2 >

Aleksandr. ©   (2005-02-11 15:12:00) 

Есть класс-наследник от 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 очень длительный.

---

Набережных С.   (2005-02-18 14:48:00) [1]

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

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

---

Digitman   (2005-02-18 09:12:00) [2]

да, забыл  ..

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   (2005-02-18 09:08:00) [3]

я бы даже так вот так сделал это :

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;

---

Набережных С.   (2005-02-18 08:13:00) [4]

А мож примерно так, а?

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;


---

Aleksandr.   (2005-02-17 16:44:00) [5]

Ну ничего у меня не выходит, млин! Какие странные ступоры выплывают. Мало того, по условию теперь в свойства нити необходимо добавить еще один 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;



---

Leonid Troyanovsky   (2005-02-14 23:19:00) [6]

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


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

--
Regards, LVT.

---

Alex Konshin   (2005-02-14 22:11:00) [7]

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

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

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

---

Leonid Troyanovsky   (2005-02-14 22:03:00) [8]

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

---

Aleksandr.   (2005-02-14 12:34:00) [9]

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

---

Verg   (2005-02-12 15:41:00) [10]

Я бы сделал объект очередь. У нее 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;

---

Eraser   (2005-02-12 15:18:00) [11]

Aleksandr.

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

---

Alexander Panov   (2005-02-12 14:57:00) [12]

Aleksandr.   (11.02.05 15:12)

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

---

Набережных С.   (2005-02-12 14:04:00) [13]

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

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

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

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

---

Leonid Troyanovsky   (2005-02-12 12:26:00) [14]

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

---

Alex Konshin   (2005-02-12 11:26:00) [15]

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

---

Leonid Troyanovsky   (2005-02-12 11:01:00) [16]

> 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 05:49:00) [17]

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

---

Alex Konshin   (2005-02-12 05:29:00) [18]

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

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

---

Leonid Troyanovsky   (2005-02-12 04:43:00) [19]

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

---

Aleksandr.   (2005-02-12 03:39:00) [20]

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

---

Страницы:   < 1 2 >

Наверх