Вопрос # 624: …ксперты! Здравствуйте :) Зараннее прошу извинить за изложение - я нов… (Прочее)

Здравствуйте, уважаемые эксперты!
Здравствуйте :) Зараннее прошу извинить за изложение - я новичок как в программированиии так и в написаниях сообщ. на форумах.
Очередная (наверное :)) головоломка по потокам.
Для выполнения множества динамических долгих по выполнению (сервера удаленные) SQL-запросов, что формируются как колонка в результате другого SQL-запроса, решено было создать пул из одновременно выполняющих запросы потоков. Итак, основной поток организовывал прокрутку запросов, а пул подпотоков выполнял.

Вот основные моменты кода:

(приложение)

Итак, главные объекты
hmMain - хэндл на мютекс для синхронизации данных (в частности, счетчика в MySemaphore) между основным и подчиненными потоками
MySemaphore - для временного замораживания осн. потока в случае перегрузки пула подпотоков.

Где я ошибся? Почему счетчик семафора "переполняется" и вылетает в п. 2 (подписано - есть комментарий)

Приложение:

procedure TfrmMain.Run;
var
i, RecordCount : integer;
begin
//
SetLength(ActiveThreads, seThreadsCount.Value);
try
for i := 0 to seThreadsCount.Value-1 do
begin
ActiveThreads.thr := nil;
ActiveThreads.Status:=tsNil;
end;
if (hmMain<>0)then
try
try
sqlqOperators.First;
i:=0;
while not sqlqOperators.Eof do
begin
CreateQueryThread(i);
Application.ProcessMessages;
sqlqOperators.Next;
inc(i);
end;
WaitForAllQueryThreadsDone;
CheckForQueryResults;
finally
MySemaphore.Free;
end;
finally
CloseHandle(hmMain);
end;
finally
Finalize(ActiveThreads);
end;
end;
function TfrmMain.CreateQueryThread(Counter : Integer): Integer;
var
nth : TThreadQueryVRNTK;
i : Integer;
s:string;
begin
AddLog('Main before MySemaphore.CheckBeforeAdd');
AddLog('Main after MySemaphore.CheckBeforeAdd');
WaitForMainBlock;
AddLog('Main after WaitForMainBlock');
try
if (MySemaphore.AddRef>MySemaphore.MaxCount) then
begin
s:='MySemaphore.AddRef>MySemaphore.MaxCount';
AddLog(s);
end;
nth := nil;
Result:=-1;
for i := 0 to Length(ActiveThreads)-1 do
begin
case ActiveThreads.Status of
tsNil :
begin
nth := TThreadQueryVRNTK.Create(True);
nth.ConnectionString:=Copy(sqlcOperators.Params.Text, 1, Length(sqlcOperators.Params.Text));
ActiveThreads.thr:=nth;
nth.ThreadNumber:=i;
nth.FreeOnTerminate:=True;
end;
nth := ActiveThreads.thr;
tsRunning:
Continue;
end;
nth.SQL := Copy(sqlqOperators.FieldByName('SQL').AsString, 1, Length(sqlqOperators.FieldByName('SQL').AsString));
nth.QueryNumber:=Counter;
ActiveThreads.Status:=tsRunning;
Result:=i;
break;
end;
if (nth=nil) then
begin
s:='(';
for i := 0 to Length(ActiveThreads)-1 do
s:=s+Format('[PoolID=%d, QueryNumb=%d], ', [i, ActiveThreads.thr.QueryNumber]);
s:=s+')';
s:=Format('Error creating thread for query %d. Nth is nil. Result=%d. ActiveThreads are:#13#10%s', [Counter+1, Result, s]);
AddLog(s);
raise exception.Create(s);
end
else
NewQueryRunning(nth);
if (i>Length(ActiveThreads)) then
finally
ReleaseMainBlock(nil);
end;
end;
function TMySemaphore.AddRef: Integer;
begin
inc(FCounter);
Result := FCounter;
end;
procedure TMySemaphore.Release;
begin
dec(FCounter);
end;
procedure TMySemaphore.CheckBeforeAdd;
begin
if (FCounter>=FMaxCount) then
begin
if (WaitForSingleObject(FhEvent, INFINITE)<>WAIT_OBJECT_0) then
raise Exception.Create('WaitForSingleObject(FhEvent, INFINITE)<>WAIT_OBJECT_0');
ResetEvent(FhEvent);
end;
end;
procedure TMySemaphore.CheckAfterRelease;
begin
if (FCounter>=FMaxCount-1) then
SetEvent(FhEvent);
end;
constructor TMySemaphore.Create(Count : Integer);
begin
FMaxCount:=Count;
FCounter:=0;
FhEvent := CreateEvent(nil, True, False, nil);
end;
procedure TThreadQueryVRNTK.Execute;
begin
try
//SetName;
//raise(Exception.CreateFmt('Thread=%d', [GetCurrentThreadId]));
finally
end;
end;
procedure TfrmMain.ReleaseQueryThread(athq: TThreadQueryVRNTK);
begin
//athq.AddLog(Format('Thread %d before ReleaseQueryThread', [athq.ThreadID]));
WaitForMainBlock;
try
MySemaphore.Release;
finally
ReleaseMainBlock(athq);
MySemaphore.CheckAfterRelease;
end;
//athq.AddLog(Format('Thread %d after ReleaseQueryThread', [athq.ThreadID]));
end;
procedure TfrmMain.ReleaseMainBlock(Athr:TThread);
var
RelRes:BOOL;
begin
RelRes := ReleaseMutex(hmMain);
if (not RelRes) then
//if (Athr<>nil) then
Win32Check(RelRes);
end;
procedure TfrmMain.WaitForMainBlock;
var
WaitRes:integer;
begin
//while
WaitRes:=WaitForSingleObject(hmMain, INFINITE);
if (WaitRes<>WAIT_OBJECT_0) then //do;
raise(Exception.CreateFmt('res=%d', [WaitRes]));
end;

Здравствуйте, Иванов Сергей!
Вы сделали свою реализацию семафора, а она не потокобезопасная. Почему бы не использовать стандартную? А так как она не является безопасной, то она и не работает. Точнее работает, но не всегда.

Мини-форум вопроса

Чтобы оставлять сообщения в мини-форумах, Вы должны авторизироваться на сайте.

Экспертная система Delphi.int.ru

Сообщество программистов
Общение, помощь, обмен опытом

Delphi.int.ru Expert

Другие разделы портала

Переход к вопросу:

Статистика за сегодня:

Лучшие эксперты

Вопрос # 624

Мини-форум вопроса

Экспертная система Delphi.int.ru

Сообщество программистовОбщение, помощь, обмен опытом

Delphi.int.ru Expert

Другие разделы портала

Переход к вопросу:

Статистика за сегодня:

Лучшие эксперты

Вопрос # 624

Мини-форум вопроса

Сообщество программистов
Общение, помощь, обмен опытом