Вопрос про потоки

[genna29]

Добрый день. Не работал раньше с потоками, столкнулся с следующей задачей. Имеется stringlist,
в нем некоторое количество например ссылок. для каждой ссылки создаю поток(MyTh1), скачиваю в нем
информацию, разбитую на блоки. Каждый блок нужно обработать в потоках(MyTh2), количество которых
ограниченно threadscount. вопрос в том, как передать данные для обработки в виде массива из TMyRecord1 в поток MyTh2,
если он локален в первом потоке...
Заранее спасибо за советы

TMyRecord1=record
var1:string;
var2:integer;
end;

TMyRecord2=record
var1:string;
var2:integer;
end;



TMyTh1=class(TTHread)
..
TMyTh2=class(TThread)
private 
	arraylength:word;
..
var
	result:array of TMyRecord2;
	current1,current2:Word;
procedure start;
var 
	target:Tstringlist;
	MyThs1=array of TMyTh1;
begin
	current1:=0;
	for i:=0 to Target.count -1 do
	begin
		SetLength(MyThs1,i+1);
		MyThs1[i]:=Myths1.Create(false);
	end;	
end;

procedure MyTh1.Execute;
var
	cs:tcriticalsection;
	targetline:string;
	target2:array of TMyRecord1;
	myths2:array of Myth2;
begin
	cs:=tcriticalsection.create;
	cs.enter;
		targetline:=Target[i];
		inc(i);
	cs.leave;
	cs.free;
	while process(targetline) do
	begin
		l:=length(target2);
		setlength(target2,l+1);
		target2.var1:=...
		target2.var2:=...
	end;
	setlength(myth2,threadscount);	
	for i=0 to threadscount - 1 do
	begin
		myths2[i]:=Tmyth2.create(true);
		myths2[i].arraylength:=length(target2);	
	end;		
end;
Procedure MyTh2.execute;
var
	cs:tcriticalsection;
begin
	cs:=tcriticalsection.create;
	while i<arraylength do
	begin
		cs.enter;
		inc(i);
		cs.leave;
		//здесь надо обработать массив target2 
		//и записать результаты в массив result,
		//общий для всех потоков...
	end;
	cs.free;
end;

[eoln]

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

как передать данные для обработки в виде массива из TMyRecord1 в поток MyTh2,
если он локален в первом потоке...

Если правильно понял, то MyTh2 запускается из MyTh1. Надо переписать конструктор для потока MyTh2
Вот пример изменения конструктора http://programmersforum.ru/showpost....3&postcount=19
А вот тут тот же код, но оптимизированный http://programmersforum.ru/showpost....8&postcount=21
Там вместо url_name: string; memo_name: TMemo надо передать массив. Ну или в данном случае указатель на массив (в этом случае вызывающий поток обязательно должен дождаться завершения вызываемого, иначе локальный массив будет стёрт раньше времени)

P.S. Что касается ограничение числа потоков, то прекрасно подойдут семафоры.

[genna29]

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

unit mythread;

interface
uses windows,classes,syncobjs,forms,sysutils,dialogs;
type
  TMyRecord1=record
    var1:string;
    var2:integer;
  end;
  TMyRecord2=record
    var1:string;
    var2:integer;
  end;
  TMyTh1=class(TTHread)
    protected
    procedure Execute;override;
    procedure log;
    private
      text2:string;
  end;
   TMyTh2=class(TTHread)
    protected
    procedure log;
    procedure Execute;override;
    constructor Create(p:pointer;curnumber:integer;cmax:integer); overload;
    private
      text:string;
      parr:pointer;
      cnumber:integer;
      max:integer;
  end;
const
  threadscount=5;
var
  s:string;
  count,current2:array of integer;
	result:array of TMyRecord2;
	current,count2:integer;
  target:Tstringlist;

procedure start;

implementation

uses Unit1;
procedure start;
var
	MyThs1:array of TMyTh1;
  i:integer;
begin
  s:='';
  setlength(count,0);
  setlength(current2,0);
	current:=0;
  target:=tstringlist.Create;
  for i := 0 to 9 do
    target.Add('test_abcdef'+inttostr(i));
  count2:=0;
  for i:=0 to Target.count -1 do
	begin
		SetLength(MyThs1,i+1);
		MyThs1[i]:=TMyth1.Create(false);
	end;
  while count2<target.count do
  begin
    application.ProcessMessages;
    sleep(100);
  end;
  target.Free;
  showmessage(inttostr(length(s)));
end;
procedure TMyTh1.log;
begin
  form1.memo1.lines.add(text2);
end;
procedure TMyTh1.Execute;
var
  l,i:integer;
	cs:tcriticalsection;
	targetline:string;
	target2:array of TMyRecord2;
	myths2:array of TMyth2;
  curnumber:integer;
begin
	cs:=tcriticalsection.create;
	cs.enter;
		targetline:=Target[current];
    curnumber:=current;
		InterlockedIncrement(current);
    i:=current;
	cs.leave;
  i:=0;
	while i<length(targetline) do
	begin
		l:=length(target2);
		setlength(target2,l+1);
		target2[l].var1:=targetline[i+1];
		target2[l].var2:=i+1;
    InterlockedIncrement(i);
	end;
	setlength(myths2,threadscount);
  setlength(count,length(count)+1);
  setlength(current2,length(current2)+1);
  current2[curnumber]:=0;
  count[curnumber]:=0;
	for i:=0 to threadscount - 1 do
	begin
		myths2[i]:=Tmyth2.create(@target2[Low(target2)],curnumber,length(target2));
	end;
  while count[curnumber]<length(target2) do
  begin
    application.ProcessMessages;
    sleep(100);
  end;
  cs.Enter;
    InterlockedIncrement(count2);
  cs.Leave;
  cs.Free;
end;
constructor TMyTh2.Create(p:pointer;curnumber:integer;cmax:integer);
begin
  inherited Create(false);
  parr:=p;
  cnumber:=curnumber;
  max:=cmax;
end;
procedure TMyTh2.Execute;
var
  cs:tcriticalsection;
  var1:string;
  var2:integer;
  arr2:array of TMyRecord2;
begin
  cs:=tcriticalsection.Create;
  while current2[cnumber]< max do
  begin
    cs.Enter;
      pointer(arr2):=parr;
      var1:=arr2[current2[cnumber]].var1;
      var2:=arr2[current2[cnumber]].var2;
      InterlockedIncrement(current2[cnumber]);
    cs.Leave;
    text:=var1;
    cs.enter;
      s:=s+var1;
    cs.Leave;
    synchronize(log);
    InterlockedIncrement(count[cnumber]);
  end;
  cs.Free;
end;
procedure TMyTh2.log;
begin
  form1.Memo1.Lines.Add(text);
end;

end.

может кто подкинет пару идей или ссылок?

[eoln]

1)

  while count2<target.count do
  begin
    application.ProcessMessages;
    sleep(100);
  end;

Надо заменить на WaitForSingleObject (см. MSDN), контролировать количество работающих потоков семафорами, а чтобы не было зависания сделать процедуру start в отдельном потоке. Но можно и так оставить - работать будет

2)

  for i:=0 to Target.count -1 do
	begin
		SetLength(MyThs1,i+1);
		MyThs1[i]:=TMyth1.Create(false);
	end;

можно заменить на

  SetLength(MyThs1,Target.count);
  for i:=0 to Target.count -1 do
		MyThs1[i]:=TMyth1.Create(false);

3) Если первые два пункта необязательно исправлять, но следующая часть ошибочна. Чередование синхронизаций странное. Например,

    cs.enter;
      s:=s+var1;
    cs.Leave;
    synchronize(log);

можно всё и в synchronize засунуть. А сами критические секции должны создаваться вне потока, иначе смысла от них нет. Вчастности в процедуре procedure TMyTh2.Execute; локально объявленная критическая секция cs блокирует доступ для повторного доступа от себя самой, что никогда не произойдёт в силу её последовательности. Аналогично для первого потока.

Ссылки, например, http://www.rsdn.ru/ http://www.msdn.ru/