VPF::И вновь о многопоточности. - Форум программистов

CynicRus

Дата 30.8.2013, 12:41 (ссылка)

(нет голосов)

Загрузка ...

Бывалый

Профиль
Группа: Участник
Сообщений: 248
Регистрация: 31.5.2012

Репутация: нет
Всего: 5

Приветствую уважаемых форумчан.

Имеется класс, создающий треды. И Класс треда, который обрабатывает в несколько экземпляров глобальную очередь, перекидывая полученное и обработанное из очереди в другую очередь. Затем необходимо, чтобы по завершению обработки первой очереди пулом тредов - запускался второй пул тредов, который обработает уже вторую очередь. Но, задача - дабы второй пул тредов запускался только после того как отработал первый. И вот как организовать метод execute потоков пула, чтобы дождаться завершения его работы - у меня чего-то ум за разум зашёл.

Код потока:

Код


TThreadUrlWorker = class(TThread)
private
  FBusy: boolean;
  FWriteLn: TStatusProc;
  FHttp: TIdHttp;
  procedure ProcessThreadUrl(Url: AnsiString);
public
  constructor Create(Status: TStatusProc);overload;
  destructor Destroy();override;
  procedure Execute; override;
  property busy: boolean read fbusy;
end;

//

destructor TThreadUrlWorker.Destroy;
begin
  FHttp.Free;
  inherited;
end;

procedure TThreadUrlWorker.Execute;
var
 cnt: integer;
 item: AnsiString;
begin
  //inherited;
  while not Terminated do
  begin
  Fbusy:=true;
  cnt:=PagesQueue.getCount;
  if (cnt > 0) then
   begin
   item:=PagesQueue.Pop;
   if not eq(Item,'Empty!') and not eq(Item,'') then
   ProcessThreadUrl(item);

   end else
      fbusy:=false;
  end;

end;



procedure TThreadUrlWorker.ProcessThreadUrl(Url: AnsiString);
var
 info: ansistring;
 GotUrls: TList<ansistring>;
 i: integer;
begin
 info:=FHttp.Get(Url);
 if Assigned(FWriteLn) then
   FWriteLn('Grab links from:'+#32+Url);
 try
 if (Length(info) > 0) then
  begin
    GotUrls:=GetPostUrls(info);
     if Assigned(GotUrls) then
     for I := 0 to GotUrls.Count-1 do
       begin
         FWriteLn('We add to post queue:' + GotUrls[i]);
         PostUrlQueue.Push(GotUrls[i]);
        end;
  end;
 finally
   if Assigned(GotUrls) then
    GotUrls.Free;
 end;

end;

Потоки то работают, но....работают постоянно, тоесть даже когда он ничего не делает по идее, он всё равно прогоняет execute.

Код пула:

Код


procedure TPagesThreadPool.AddThreads(n: Integer);
var
i: Integer;
begin
for i := 1 to n do
 begin
  inherited Add(TThreadUrlWorker.Create(FWriteLn));
 end;
 // if assigned(FWriteLn) then
   // FWriteLn('We add '+#32+'post workers');
end;

constructor TPagesThreadPool.Create(n: integer; Status: TStatusProc);
begin
 inherited Create;
 FWriteLn:=status;
 AddThreads(n);

end;

destructor TPagesThreadPool.Destroy;
begin
  Remove(Count);
 inherited;
end;

function TPagesThreadPool.GetBusy: integer;
 var
  i: integer;
begin
  result:=0;
  for i:=0 to count -1 do
   if TThreadUrlWorker(Items[i]).busy then
    inc(result);
end;

procedure TPagesThreadPool.Remove(n: Integer);
var
i: Integer;
begin
for i := n-1 downto 0 do begin
  with TThread(Items[i]) do begin
    Terminate;
    WaitFor;
  end;
  Delete(i);
end;
end;

PS: метод с получением количества занятых потоков не сработал. Собственно вопрос - как это организовать правильно?

MetalFan

Дата 2.9.2013, 11:34 (ссылка)

(нет голосов)

Загрузка ...

Аццкий Сотона

Профиль
Группа: Комодератор
Сообщений: 3815
Регистрация: 2.10.2006
Где: Moscow

Репутация: 62
Всего: 128

Мне кажется надо "более правильно" распределить задачи.
Пусть "рулением" выдаваемых потокам задач занимается отдельный поток-менеджер.
А в рабочем потоке добавить объект ядра (я, как правило, использую EVENT), для управления состоянием/определения состояния.
Т.е. алгоритм примерно такой:
1. Поток-менеджер (ПМ) следит за очередью задач на исполнение... появилась задача
2. ПМ проверяет пул потоков, берет первый "свободный" рабочий поток (РП) (у которого, к примеру, не "взеден" EVENT).
3. ПМ извлекает задачу из очередни на исполнение и назначает ее выбранному в п.2 РП. Событие "занятости" у РП взводится (SetEvent).
4. В execute РП в цикле проверяется состояние события (WaitForSingle/MultipleObject[s] /TEvent.WaitFor), если событие взведено - выполняет задачу, по окончанию "сбрасывает" состояние события.
5. ПМ проверяет состояние пула потоков, если какой-то из них "закончил" работу над задачей, "извлекает" ее из потока и помещает в очередь отработанных задач. Сигнализирует, при необходимости "наружу" о новой завершенной задаче.

Естественно, все действия с разделяемыми данными должны быть потокобезопасны.
Никаких небезопасных присвоений флагов (FBusy) внутри одного потока - проверки флагов другим и т.п.

Как-то так обычно у меня логика построена

Это сообщение отредактировал(а) MetalFan - 2.9.2013, 11:39

--------------------

There are always someone smarter than you...

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Delphi: Общие вопросы \| Следующая тема »