получение видео по idHttp. Зависший поток.

[olikke]

Здавствуйте!

Задача:
Есть несколько IP камер Axis (P1344) и бесконечное количество клиентов для них. Клиент может подключиться к любой из камер, а также к нескольким из них.
Поскольку к камерам напрямую подключаться нельзя, нужно создать что-то типа видеосервера. В задачу видеосервера входит получение от всех камер видеопотока по http и выдача их в сеть клиентам (UDP Broadcast - все клиенты в одной локальной сети).
Как делаю:
Для каждой камеры создаётся свой экземпляр TThread, внутри которого живёт idHTTP. Ну и естественно свой UDPClient (тоже использую Indy10).

Собственно код потока для HTTP:

PHP код:

unit THttp;

interface

uses
  Windows,Classes,SysUtils,IdHTTP,SyncObjs,UDP;

type
  TSt = class(TThread)
  private
    idHTTP1:TidHTTP;
    UDPClient1:TUDP;
    xStream: TStream;
  protected
    function LiveStreamWrite(const aBuffer; aCount: Longint): Longint;
    procedure Execute; override;
  public
    URL:String;
    Numb:byte;
    FEvent:TEvent;
  end;

implementation
uses
  LiveStreamer,Global;

procedure TSt.Execute;
begin
  FreeOnTerminate:=true;
  idHTTP1:=TidHTTP.Create(nil);
  idHTTP1.Request:=TidHTTPRequest.Create(idHTTP1);
  idHTTP1.Request.BasicAuthentication:=true;
  idHTTP1.Request.Password:='****';
  idHTTP1.Request.Username:='****';
  idHTTP1.HTTPOptions:=[hoForceEncodeParams];
  idHTTp1.Request.Connection:='Keep-Alive';
  UDPClient1:=TUDP.Create(True);
  UDPClient1.Priority:=tpNormal;
  UDPClient1.FreeOnTerminate:=True;
  UDPClient1.Port:=Table[Numb].Port;
  UDPClient1.Numb:=Numb;
  UDPClient1.Resume;
  xStream := TLiveStream.Create(nil, LiveStreamWrite);
  xStream.Size:=$8800;
  XStream.Position:=0;
  while (not fl_STOP) and (not fl_st[Numb]) do
  begin
    try idHTTP1.Get(URL, xStream);
    finally  end;
  end;
  FreeAndNil(xStream);
  UDPCLient1.Terminate;
  Event[numb].Setevent;
  UDPClient1:=nil;
end;



function TSt.LiveStreamWrite(const aBuffer; aCount: Integer): Longint;
begin
    Buffer[Numb].Lock;
  try
    Buffer[Numb].Size:=0;
    Buffer[Numb].Write(aBuffer,aCount);
    Buffer[Numb].Position:=0;
  finally
    Buffer[Numb].UnLock;
  end;
  Event[Numb].SetEvent;
end;

end. 


Код потока для UDP:

PHP код:

unit UDP;

interface

uses
  Classes,IdUDPClient,IdGlobal,SyncObjs,SysUtils;

type
  TUDP = class(TThread)
  private
    UDPClient:TidUDPClient;
  protected
    procedure Execute; override;
  public
    Port:integer;
    HOST:string;
    Numb:byte;
  end;


implementation

uses Global;


procedure TUDP.Execute;
var buff:string;
begin
  FreeOnTerminate:=true;
  //Создание клиента для передачи данных
  UDPClient:=TIdUDPClient.Create;
  UDPClient.BufferSize:=$8800;
  UDPClient.IPVersion:=Id_IPv4;
  UDPClient.Port:=Port;
  UDPClient.ReceiveTimeout:=-1;
  UDPClient.Tag:=0;
  UDPClient.BroadcastEnabled:=true;
  UDPClient.Host:=Table[Numb].Broadcast;
  try UDPClient.Active:=True;
  except end;
  while not Terminated do
  begin
    Event[Numb].WaitFor(INFINITE);
    Event[Numb].ResetEvent;
    if terminated then
    begin
      UDPClient.Active:=false;
      UDPClient.Free;
      buff:='';
      exit;
    end;
    buff:='';
      Buffer[Numb].Lock;
    try
      Buff:=Buffer[Numb].DataString;
    finally
      Buffer[Numb].UnLock;
    end;
    try UDPClient.Send(TRIM(Table[Numb].Broadcast),UDPClient.Port,buff);
    except
    end;
    buff:='';
  end;
end;

end. 


Так вот, вся эта зараза прекрасно работает. Клиенты смотрят картинки с камер.
Но где-то через час-два каринтка пропадает. Точнее от IP камеры пропадает поток (смотрела сниффером). А значит экземпояр потока TSt, привязанного к этой камере бесконечно ждёт продолжения кино (фактически висит).

У меня два вопроса:
1. Почему пропадает поток от IP камеры (мой косяк или её?)
2. Как мне грамотно прибить поток висящий? Кроме TerminateThread до него ничем не достучаться. Но в этом случае все мои ресурсы остаются незакрытыми. Да и неправильно это как-то....

СПАСИТЕ МЕНЯ!!!

[саша40]

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

[olikke]

саша40, как это обработать?

[raxp]

...а что это за LiveStreamer-юнит? Из какого SDK (в Axis SDK я такого не видел) или самописный модуль?

[саша40]

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

саша40, как это обработать?

не знаю. Как-то обрабатывается.

[Человек_Борща]

Я бы тоже завис после такого...

  while not Terminated do
  begin
    Event[Numb].WaitFor(INFINITE);
    Event[Numb].ResetEvent;
    if terminated then

Где конструкторы классов? Где try finally блоки?

Переработал ваш UDP трэд:

constructor TUDP.Create;
begin
  inhereited Create(True);
  FreeOnTerminate:=true;
  UDPClient:=TIdUDPClient.Create;
  UDPClient.BufferSize:=$8800;
  UDPClient.IPVersion:=Id_IPv4;
  UDPClient.Port:=Port;
  UDPClient.ReceiveTimeout:=-1;
  UDPClient.Tag:=0;
  UDPClient.BroadcastEnabled:=true;
  UDPClient.Host:=Table[Numb].Broadcast;
and;


destructor TUDP.Destroy; override;
begin
   UDPClient.Active:=false;
   UDPClient.Free;
inhereited;
and;


procedure TUDP.Execute;
var buff:string;
begin
  UDPClient.Active:=True; //Зачем try except если исключение не обрабатывается. Плохо так делать.. 
  while not Terminated do
  begin
   if  Event[Numb].WaitFor(INFINITE) = wrSignaled then
   begin
    if terminated then
     break;
         buff:='';
     
    try
      Buffer[Numb].Lock;
      Buff:=Buffer[Numb].DataString;
    finally
      Buffer[Numb].UnLock;
    end;
   UDPClient.Send(TRIM(Table[Numb].Broadcast),UDPClient.Port,buff); //Зачем try except если исключение не обрабатывается. Плохо так делать.. 
   Event[Numb].ResetEvent; //Wait for next signal state
   end; //Event end
  end;
end;

После увиденного кода, косяк ваш. И не один..

1. Почему пропадает поток от IP камеры (мой косяк или её?)

Где блоги try finally? Где конструкторы и деструкторы потоков?

Где обработка try except?(У вас она есть, НО вы же никак не знаете о том, что исключение поймано, и что за оно - вы тоже не знаете. Лог чтоль прикрутите...).
Кстати очевидно же, это есмъ причина сего:

Но где-то через час-два каринтка пропадает.

2. Как мне грамотно прибить поток висящий? Кроме TerminateThread до него ничем не достучаться. Но в этом случае все мои ресурсы остаются незакрытыми. Да и неправильно это как-то....

TMyThreead.Event[Numb].SetEvent;  //Поток получает сигнал
TMyThreead.Terminate;  //Terminated = true, поток выскочит из цикла while not terminated
TMyThreead.WaitFor;  //Основной поток ждет завершения дополнительного

в дополнение:

 UDPClient.ReceiveTimeout:=-1;

Т.е. таймаута нет? Вы 100% уверены что не будет обрыва на линии, что не будет потери пакетов? Что сервер не перезагрузят? А то ведь, бесконечно ждать будите с таким свойством
Плохая идея TEvent глобальным делать, это не критическая секция. Еслион глобальный, то влияет на все TUDP одновременно.

[olikke]

...а что это за LiveStreamer-юнит? Из какого SDK (в Axis SDK я такого не видел) или самописный модуль?

К сожалению за давностью лет не помню автора:

unit LiveStreamer;

interface

uses
  Classes;

type
  TLSOnRead = function(var vBuffer; aCount: Longint): Longint of object;
  TLSOnWrite = function(const aBuffer; aCount: Longint): Longint of object;

  TLiveStream = class(TStream)
  protected
    fOnRead: TLSOnRead;
    fOnWrite: TLSOnWrite;
    // We dont override and just ignore SetSize. Some callers might try to preallocate
    // size. Because of this we just ignore instead of throwing an error.
    // procedure SetSize(NewSize: Longint); override;
    // procedure SetSize(const NewSize: Int64); override;
  public
    constructor Create(aOnRead: TLSOnRead; aOnWrite: TLSOnWrite);
    function Read(var vBuffer; aCount: Longint): Longint; override;
    function Write(const aBuffer; aCount: Longint): Longint; override;
    // We dont implement seek either, we catch all data...
    function Seek(const aOffset: Int64; aOrigin: TSeekOrigin): Int64; override;
  end;

implementation

uses
  SysUtils;

{ TLiveStream }

constructor TLiveStream.Create(aOnRead: TLSOnRead; aOnWrite: TLSOnWrite);
begin
  inherited Create;
  fOnRead := aOnRead;
  fOnWrite := aOnWrite;
end;

function TLiveStream.Read(var vBuffer; aCount: Integer): Longint;
begin
  if Assigned(fOnRead) then begin
    Result := fOnRead(vBuffer, aCount);
  end else begin
    // If not handled, we just ignore and say we handled it all so we dont err out caller
    Result := aCount;
  end;
end;

function TLiveStream.Seek(const aOffset: Int64; aOrigin: TSeekOrigin): Int64;
begin
  // Must return 0 - ie no position. If not stack overflow becuase how TStream.Seek is
  // implemented.
  Result := 0;
end;

function TLiveStream.Write(const aBuffer; aCount: Integer): Longint;
begin
  if Assigned(fOnWrite) then begin
    Result := fOnWrite(aBuffer, aCount);
  end else begin
    // If not handled, we just ignore and say we handled it all so we dont err out caller
    Result := aCount;
  end;
end;

end.

[olikke]

Где конструкторы классов?

Вы абсолютно правы. Впишу.

Где обработка try except?(У вас она есть, НО вы же никак не знаете о том, что исключение поймано, и что за оно - вы тоже не знаете. Лог чтоль прикрутите...).
Кстати очевидно же, это есмъ причина сего:

Есть в исходнике у меня обработка - лог. Выкинула из примера т.к. ничего интересного они мне не показали.

в дополнение:

Код:

UDPClient.ReceiveTimeout:=-1;

Т.е. таймаута нет? Вы 100% уверены что не будет обрыва на линии, что не будет потери пакетов? Что сервер не перезагрузят? А то ведь, бесконечно ждать будите с таким свойством

Я ничего не жду.

Плохая идея TEvent глобальным делать, это не критическая секция. Еслион глобальный, то влияет на все TUDP одновременно.

У меня массив TEventov - для каждого TUDP он свой.

Код:

TMyThreead.Event[Numb].SetEvent; //Поток получает сигнал
TMyThreead.Terminate; //Terminated = true, поток выскочит из цикла while not terminated
TMyThreead.WaitFor; //Основной поток ждет завершения дополнительного

Я обычно пишу наоборот

TMyThreead.Terminate;  //Terminated = true, поток выскочит из цикла
TMyThreead.Event[Numb].SetEvent;  //Поток получает сигнал
 while not terminated
TMyThreead.WaitFor;  //Основной поток ждет завершения дополнительного

Но в данном случае мне это не поможет. Ведь зависает не TUDP, а TSt, в котором происходит следующее:
В TSt.Execute я дохожу до кода

while (not fl_STOP) and (not fl_st[Numb]) do
begin
try idHTTP1.Get(URL, xStream);
finally ... end;
end;

и остаюсь там навсегда, т.к. xStream бесконечен.
Единственная возможность завершиться (не уверена что хорошо так делать), это в функции считывания данных из буфера сделать так:

function TSt.LiveStreamWrite(const aBuffer; aCount: Integer): Longint;
var str:string;
begin

if fl_STOP then
begin
idHTTP1.Disconnect;
exit;
end;
Buffer[Numb].Lock;
try
Buffer[Numb].Size:=0;
Buffer[Numb].Write(aBuffer,aCount);
Buffer[Numb].Position:=0;
finally
Buffer[Numb].UnLock;
end;
Event[Numb].SetEvent;
INC(HTTPCount[Numb]);
end;

По крайней мере, когда данные по http приходят - это помогает.
Но данные от камеры просто перестают приходить.
Просматривая сниффером пакеты от камеры я видела, как в один прекрасный момент CheckSum сетевого и транспортного протокола (IP и TCP) становятся incorrect. При этом idHTTP всё также ожидает данных

idHTTP1.Get(URL, xStream);

а их нет - функция

function TSt.LiveStreamWrite(const aBuffer; aCount: Integer): Longint;

больше не вызывается никогда.
Всё что, как я понимаю, можно сделать - поставить следящий поток, и отлавливая зависания idHTTP делать TerminateThread. Но это ведь не корректно.
Завтра внесу правки, указанные Человек_Борща, но боюсь это мне не поможет....

[Человек_Борща]

Я ничего не жду.

В этом ваша проблема.
ReciveTimeout - указывает кол-во секунд в течении которых клиент ждет хотя бы 1 байт информации от сервера, в противном случае соединение размыкается. В вашем случае клиент ждет отдачи сервера бесконечно, итог - поток висит.

У меня массив TEventov - для каждого TUDP он свой.

Не практично. Если TEvent для TThread, то пусть он будет инкапсулирован в TThread и доступен через общедоступное свойство или метод.
Какова гарантия что завтро кол-во TThread'ов не станет больше кол-ва TEvent'ов в массиве?

try idHTTP1.Get(URL, xStream);
finally ... end;

...клиент ждет хотя бы 1 байт информации от сервера, в противном случае соединение размыкается.

[olikke]

Если TEvent для TThread, то пусть он будет инкапсулирован в TThread и доступен через общедоступное свойство или метод.
Какова гарантия что завтро кол-во TThread'ов не станет больше кол-ва TEvent'ов в массиве?

Можно, но не критично и к решению проблемы не приблизит. Количество TThread'ов определяется кол-вом камер, а оно оговорено в ТЗ.

В этом ваша проблема.
ReciveTimeout - указывает кол-во секунд в течении которых клиент ждет хотя бы 1 байт информации от сервера, в противном случае соединение размыкается. В вашем случае клиент ждет отдачи сервера бесконечно, итог - поток висит.

Да не висит он. У меня есть уже готовые клиенты для UDP потока, которые на локальных машинах прекрасно принимают, декодируют и смотрят поток от tUDP. И речь идёт не о паре-тройке посылок. За один час передаётся их порядка 324000 от каждой камеры (порядка 0,2% пропадают по дороге).
К тому же даже полное зависание tUDP (если б оно было) никак не может подвесить tST, поскольку TEvent скинут сразу, и даже из буфера Buffer[Numb] данные считываются заранее.
Просто из принципа поставлю

UDPClient.ReceiveTimeout:=0;