Пишу качалку под Линух на fpc 2.2.0+Indy. Данные хранятся в СУБД Postgres. Использую потоки.
Ниже представлена главная ф-я работы потока. Алгоритм и процедура работает нормально, только вот происходит утечка памяти
вне зависимости есть ошибка при закачке или нет. Помогите разобраться!- Код: Выделить всё
procedure TGetFile.Execute;
var
pwd : pPasswd;
res: PPGResult;
OrderDate,FileSize: AnsiString;
Query: AnsiString;
AURI: TIdURI;
i: Byte;
HeapStat: TFPCHeapStatus;
LIO : TIdSSLIOHandlerSocketOpenSSL;
//FLIO: TIdIOHandlerStack;
LC : TIdCompressorZLib;
//LHE : EIdHTTPProtocolException;
finfo: stat;
begin
if ((fpGetUID = 0) and (fpGetGID = 0)) then
begin
pwd := Getpwnam(run_as_user);
fpSetGID(pwd^.pw_gid);
fpSetUID(pwd^.pw_uid);
end;
write_log('INF: Thread #'+IntToStr(TID)+' is starting.');
{Соединяемся с базой}
DBconn := PQsetdbLogin(pghost, pgport, Nil, Nil, dbName, login, passwd);
if (PQstatus(DBconn) = CONNECTION_BAD) then
begin
Writeln (stderr, PQERRORMessage(DBconn));
halt(100);
end;
while true do
begin
Query:=''; FOrderId:= ''; FNSHD:= ''; URL:=''; OrderDate:=''; FullName:=''; FileSize:='';
//writeln('Начали #',TID);
HeapStat:=GetFPCHeapStatus;
writeln('Начали #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
res := PQexec(DBconn, 'BEGIN');
PQclear(res);
Query := 'SELECT a."ID" FROM orders a WHERE a."STATUS" = 1 ORDER BY a."DATE" ASC LIMIT 1 FOR UPDATE NOWAIT';
res := PQexec(DBconn, PChar(Query));
if (PQresultStatus(res) <> PGRES_TUPLES_OK) then
begin
PQclear(res);
res := PQexec(DBconn, 'ROLLBACK');
PQclear(res);
sleep(2000);
continue;
end;
if (PQntuples(res) = 0) then
begin
PQclear(res);
res := PQexec(DBconn, 'ROLLBACK');
PQclear(res);
sleep(2000);
continue;
end
else
begin
FOrderId:=PQgetvalue(res,0,0);
//WriteLn(FOrderId);
PQclear(res);
Query:='UPDATE orders SET "STATUS" = 2 WHERE "ID" = '''+FOrderId+'''';
WriteLn(Query);
res := PQexec(DBconn, PChar(Query));
if (PQresultStatus(res) <> PGRES_COMMAND_OK) then
begin
PQclear(res);
PQfinish(DBconn);
Writeln (stderr, PQERRORMessage(DBconn));
halt(101);
end;
PQclear(res);
res := PQexec(DBconn, 'COMMIT');
PQclear(res);
Query:='select a."URL",a."PORT_ID",a."DATE",a."PATH_FILE",a."FILE_SIZE",a."CUR_POSITION" from orders a where a."ID" = '''+FOrderId+'''';
//writeln(Query);
res := PQexec(DBconn, PChar(Query));
if (PQresultStatus(res) <> PGRES_TUPLES_OK) then
begin
PQclear(res);
PQfinish(DBconn);
Writeln (stderr, PQERRORMessage(DBconn));
halt(101);
end;
URL := PQgetvalue(res, 0, 0);
FNSHD := PQgetvalue(res, 0, 1);
OrderDate := PQgetvalue(res, 0, 2);
FullName := PQgetvalue(res, 0, 3);
FileSize := PQgetvalue(res, 0, 4);
FilePosition := StrToInt(PQgetvalue(res, 0, 5));
PQclear(res);
writeln(URL,' ',FNSHD,' ',OrderDate,' ',FullName,' ',FileSize,' ',FilePosition);
AURI := TIdURI.Create(URL);
HeapStat:=GetFPCHeapStatus;
writeln('Create URL #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
if((UpperCase(AURI.Protocol) = 'HTTP') or (UpperCase(AURI.Protocol) = 'HTTPS')) then
begin
LIO := TIdSSLIOHandlerSocketOpenSSL.Create;
HeapStat:=GetFPCHeapStatus;
writeln('Create LIO #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
LC := TIdCompressorZLib.Create;
HeapStat:=GetFPCHeapStatus;
writeln('Create LC #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
try
IdHTTP1 := TIdHTTP.Create;
HeapStat:=GetFPCHeapStatus;
writeln('Create HTTP #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
try
IdHTTP1.Compressor := LC;
IdHTTP1.OnWork := WorkEventProc;
IdHTTP1.OnWorkEnd := WorkEndEventProc;
IdHTTP1.OnWorkBegin := WorkBegEventProc;
IdHTTP1.HandleRedirects := True;
IdHTTP1.Request.UserAgent := 'Mozilla/4.0 (compatible; MSIE 7.0; Windows NT 5.1; .NET CLR 2.0.50727; .NET CLR 3.0.04506.30; INFPath.2; .NET CLR 3.0.04506.648)';
IdHTTP1.Request.Referer := URL;
if (AURI.Username <> '') then IdHTTP1.Request.Username := AURI.Username;
if (AURI.Password <> '') then IdHTTP1.Request.Password := AURI.Password;
if FileExists(FullName) then
begin
DestinationObject := TFileStream.Create(FullName,fmOpenWrite);
DestinationObject.Seek(FilePosition, soFromBeginning);
NewFile := False;
IdHTTP1.Request.Range := IntToStr(FilePosition)+'-';
end
else
begin
if not DirectoryExists(dbstorage+'/'+FNSHD) then fpMkDir(dbstorage+'/'+FNSHD,511);
if not DirectoryExists(ExtractFileDir(FullName)) then fpMkDir(ExtractFileDir(FullName),511);
DestinationObject := TFileStream.Create(FullName,fmCreate);
NewFile := True;
end;
IdHTTP1.IOHandler := LIO;
IdHTTP1.ConnectTimeOut := 10000;
IdHTTP1.ReadTimeOut := 30000;
HeapStat:=GetFPCHeapStatus;
writeln('Create File #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
IdHTTP1.Get(URL,DestinationObject);
except
on E : Exception do
begin
if E is EIdHTTPProtocolException then
begin
//LHE := E as EIdHTTPProtocolException;
Query:='UPDATE orders SET "STATUS" = 3 WHERE "ID" = '''+FOrderId+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
if Assigned(DestinationObject) then
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+'+IntToStr(DestinationObject.Position)+' WHERE "PORT_ID"='''+FNSHD+''''
else
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+0 WHERE "PORT_ID"='''+FNSHD+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
write_log('ERR: #'+IntToStr(TID)+' '+FOrderId+' '+URL+'. '+IdHTTP1.ResponseText);
if FileExists(FullName) then
begin
fpStat(FullName,finfo);
//FreeAndNil(DestinationObject);
if (finfo.st_size = 0) then fpUnLink(FullName);
end;
end
else
begin
Query:='UPDATE orders SET "STATUS" = 3 WHERE "ID" = '''+FOrderId+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
if Assigned(DestinationObject) then
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+'+IntToStr(DestinationObject.Position)+' WHERE "PORT_ID"='''+FNSHD+''''
else
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+0 WHERE "PORT_ID"='''+FNSHD+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
write_log('ERR: #'+IntToStr(TID)+' '+FOrderId+' '+URL+'. '+E.Message);
if FileExists(FullName) then
begin
fpStat(FullName,finfo);
DestinationObject.Free;
if (finfo.st_size = 0) then fpUnLink(FullName);
end;
end
end;
end;
finally
FreeAndNil(DestinationObject);
HeapStat:=GetFPCHeapStatus;
writeln('Free file #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
FreeAndNil(IdHTTP1);
HeapStat:=GetFPCHeapStatus;
writeln('Free HTTP #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
FreeAndNil(LIO);
HeapStat:=GetFPCHeapStatus;
writeln('Free LIO #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
FreeAndNil(LC);
HeapStat:=GetFPCHeapStatus;
writeln('Free LC #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
{if Assigned(DestinationObject) then DestinationObject.Destroy;
LIO.Destroy;
LC.Destroy;
IdHTTP1.Destroy;}
write_log('DEBUG: #'+IntToStr(TID)+' Free.');
end;
end;
if((UpperCase(AURI.Protocol) = 'FTP') or (UpperCase(AURI.Protocol) = 'FTPS')) then
begin
LIO := TIdSSLIOHandlerSocketOpenSSL.Create;
LC := TIdCompressorZLib.Create;
IdFTP1 := TIdFTP.Create;
try
IdFTP1.Compressor := LC;
IdFTP1.IOHandler := LIO;
IdFTP1.OnWork := WorkEventProc;
IdFTP1.OnWorkBegin := WorkBegEventProc;
IdFTP1.OnAfterGet := FTPAfterGet;
IdFTP1.Passive := True;
IdFTP1.TransferType := ftBinary;
IdFTP1.ConnectTimeOut := 10000;
IdFTP1.TransferTimeout := 30000;
IdFTP1.Host := AURI.Host;
if(AURI.Username = '') then IdFTP1.Username := 'anonymous' else IdFTP1.Username := AURI.Username;
if(AURI.Password = '') then IdFTP1.Password := 'pass@give.me' else IdFTP1.Password := AURI.Password;
try
IdFTP1.Connect;
if IdFTP1.Connected then
begin
if (AURI.Path <> '') then IdFTP1.ChangeDir(AURI.Path);
if FileExists(FullName) then
begin
DestinationObject := TFileStream.Create(FullName,fmOpenWrite);
DestinationObject.Seek(FilePosition, soFromBeginning);
NewFile := False;
IdFTP1.Get(AURI.Document,DestinationObject, True);
end
else
begin
if not DirectoryExists(dbstorage+'/'+FNSHD) then fpMkDir(dbstorage+'/'+FNSHD,511);
if not DirectoryExists(ExtractFileDir(FullName)) then fpMkDir(ExtractFileDir(FullName),511);
DestinationObject := TFileStream.Create(FullName,fmCreate);
NewFile := True;
//writeln(stderr, AURI.Document);
IdFTP1.Get(AURI.Document,DestinationObject, True);
end;
IdFTP1.Quit;
end;
except
on E : Exception do
begin
Query:='UPDATE orders SET "STATUS" = 3 WHERE "ID" = '''+FOrderId+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
if Assigned(DestinationObject) then
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+'+IntToStr(DestinationObject.Position)+' WHERE "PORT_ID"='''+FNSHD+''''
else
Query:='UPDATE clients SET "COUNT_ERRORS" = "COUNT_ERRORS"+1,"COUNT_DOWN_SIZE" = "COUNT_DOWN_SIZE"+0 WHERE "PORT_ID"='''+FNSHD+'''';
res := PQexec(DBconn, PChar(Query));
PQclear(res);
write_log('ERR: #'+IntToStr(TID)+' '+FOrderId+' '+URL+'. '+E.Message);
if FileExists(FullName) then fpUnLink(FullName);
if IdFTP1.Connected then IdFTP1.Quit;
end
end;
finally
FreeAndNil(DestinationObject);
FreeAndNil(IdFTP1);
FreeAndNil(LC);
FreeAndNil(LIO);
write_log('DEBUG: #'+IntToStr(TID)+' Free.');
end;
end;
AURI.Free;
HeapStat:=GetFPCHeapStatus;
writeln('Free URL #',TID,' ',HeapStat.CurrHeapSize,' ',HeapStat.CurrHeapUsed,' ',HeapStat.CurrHeapFree);
write_log('DEBUG: #'+IntToStr(TID)+' URL free.');
end;
end;
end;
.
