Broken Pipe while using custom service in threads

mohsenti · 2017-09-04 14:51:52

Hi ab,

I wrote a client/server RESTFul to serve image processing in server. To send files from client to server first read all file data and convert stream to RawByteString and pass data as Interface method parameter to service , because reading and writing files is too slow and cause UI freeze I create a thread to serve reading, converting , sending , receving and saving data, to determine when data are ready using another method of service (GetProgress) and if data are ready call another method (DownloadFile) , download file method return a rawbytestring and I write data to stream.

When I calling service methods in main thread all methods work fine but when running methods in separated thread randomly two exception occurs : 1 - Broken Pipe , 2 - File not found

1 - Broken Pipe : I don't now why this error occur when calling SendFile method of service.

2 - File not found : when calling DownloadFile method immediately after SendFile File not found exception occur while in SendFile method I only save file and send file id as result and in DownloadFile read that saved file and send it to client. It look like file bounded operators in service run after finished SendFile method.

Thread Execute method

procedure TFileThread.Execute;
var
  fs: TStream;
  Index: integer;
begin
  Synchronize(@SyncProgress);
  begin
    fs := TFileStream.Create(FFilePath, fmOpenRead);
    Index := FI.SendFile(StreamToRawByteString(fs));
    fs.Free;

    while True do
    begin
      FProgress := FI.GetProgress(Index);
      if (FProgress = 100) then
        break;
      Synchronize(@SyncProgress);
    end;

    FData := FI.DownloadFile(Index);
    Synchronize(@SyncFinish);
  end;
end;

Service Interface

IProcessor = interface(IInvokable)
    ['{8B1806EB-39CF-4ADC-B582-CF1C27F77B3E}']
    function SendFile(Stream: RawByteString): integer;
    function GetProgress(Index: integer): integer;
    function DownloadFile(Index: integer): RawByteString;
  end;

Service class

TProcessorService = class(TInterfacedObject, IProcessor)
  public
    function SendFile(StreamData: RawByteString): integer;
    function GetProgress(Index: integer): integer;
    function DownloadFile(Index: integer): RawByteString;
  end;

Service implementation

function TProcessorService.SendFile(StreamData: RawByteString): integer;
var
  fs: TFileStream;
  pm: TProcessModel;
  frs: TStream;
begin
  //Result := FLastIndex;
  //Save File
  fs := TFileStream.Create(IntToStr(FLastIndex) + '.jpg', fmCreate);
  frs := RawByteStringToStream(StreamData);
  frs.Position := 0;
  fs.CopyFrom(frs, frs.Size);
  fs.Free;
  
  Result := FLastIndex;
  Inc(FLastIndex);
end;

function TProcessorService.GetProgress(Index: integer): integer;
var
  I: integer;
  pm: TProcessModel;
begin
  Result := 100;
end;

function TProcessorService.DownloadFile(Index: integer): RawByteString;
var
  I: integer;
  fs: TFileStream;
begin
  fs := TFileStream.Create(inttostr(Index) + '.file', fmOpenRead);
  Result := StreamToRawByteString(fs);
  fs.Free;
end;

ab · 2017-09-04 16:30:34

Which communication method are you using?
Named pipes are not reliable when there is a lot of data: use sockets instead.

mohsenti · 2017-09-05 06:40:00

I use below code to create HttpServer.

var
  ApiMdl: TSQLModel;
  ApiSrv: TSQLRestServerDB;
  HTTPSrv: TSQLHttpServer;

begin
  //Enable logging
  with TSQLLog.Family do
  begin
    Level := LOG_VERBOSE;
    EchoToConsole := LOG_VERBOSE;
  end;

  //Create SQLModel to serve database
  ApiMdl := TSQLModel.Create([], ROOT_NAME);

  ApiSrv := TSQLRestServerDB.Create(ApiMdl, {'1.db'}SQLITE_MEMORY_DATABASE_NAME, False);

  with ApiSrv do
  begin
    //Create missing tabales
    CreateMissingTables;

    //Register services
    ServiceDefine(TProcessService, [IProcess], sicSingle);
  end;

  //Start HttpServer
  HTTPSrv := TSQLHttpServer.Create(DEFAULT_PORT, [ApiSrv], '+', HTTP_DEFAULT_MODE);

  //Allow cross-site AJAX queries
  HTTPSrv.AccessControlAllowOrigin := '*';

  WriteLn('Processor is running. press enter to terminate Processor');
  ReadLn;

  WriteLn('Processor going to stop');

  //ApiCon.Free;
  HTTPSrv.Free;
  ApiSrv.Free;
  ApiMdl.Free;

  WriteLn('Processor stoped successfull');
end.

And I think server use socket because when change port to 80 it say port 80 is in use.

I build project with lazarus 1.9 and FPC stable 3.0.2 and my os is linux (Manjaro) .

Last edited by mohsenti (2017-09-05 06:55:49)

ab · 2017-09-05 07:10:04

Please follow the forum rules.
https://synopse.info/forum/misc.php?action=rules
Don't post such extensive source code in the forum.

About your question, to process binary content, a method-based service is the best option.
SOA interface-based services upload will use base64 encoding from the client to the server, which is not very efficient.

TFileStream is not needed. And your use of it is unsafe (the Free should be protected by a try ... finally).
Just use StringFromFile() or FileFromString() functions from SynCommons.pas.

mohsenti · 2017-09-05 09:42:22

Pardon me.My goal of the code is an example of my program.

ab wrote:

Please follow the forum rules.
https://synopse.info/forum/misc.php?action=rules
Don't post such extensive source code in the forum.

Thank you for attention.

I can't use StringFromFile function because I have process data before saving them in file.
After many tries I found a way to reproduce exception, if Calling server methods in a loop all process work fine but if I make a delay between request Broken pipe exception raised. Maybe it's occur because some threads closed.
Lazarus Thread window doesn't show any thread change activity except TProcessThread running and closing.

When I press continue button of Broken Pipe exception other process work fine and file send to server and receive by client.

Last edited by mohsenti (2017-09-05 09:44:10)

ab · 2017-09-05 14:29:28

You are doing something wrong in your code.
Did you at least ensure your server side code process is thread-safe?

mohsenti · 2017-09-05 15:04:45

Yes, server side codes are thread safe. Broken pipe occur in client side before sending and receiving data to/from server.

Last edited by mohsenti (2017-09-05 15:05:34)

ab · 2017-09-05 15:41:25

So your client side is broken.

mohsenti · 2017-09-07 13:57:58

Hi ab,

I install newpascal and lazarus with fpcdeluxe and rebuild mORMot and project with that , Broken pipe exception goes away and project works fine.
I compile project with delphi and fpc in windows and all tests passed successful.

Last edited by mohsenti (2017-09-07 13:58:08)

ab · 2017-09-07 15:07:45

Good news...

mORMot Open Source

#1 2017-09-04 14:51:52

Broken Pipe while using custom service in threads

#2 2017-09-04 16:30:34

Re: Broken Pipe while using custom service in threads

#3 2017-09-05 06:40:00

Re: Broken Pipe while using custom service in threads

#4 2017-09-05 07:10:04

Re: Broken Pipe while using custom service in threads

#5 2017-09-05 09:42:22

Re: Broken Pipe while using custom service in threads

#6 2017-09-05 14:29:28

Re: Broken Pipe while using custom service in threads

#7 2017-09-05 15:04:45

Re: Broken Pipe while using custom service in threads

#8 2017-09-05 15:41:25

Re: Broken Pipe while using custom service in threads

#9 2017-09-07 13:57:58

Re: Broken Pipe while using custom service in threads

#10 2017-09-07 15:07:45

Re: Broken Pipe while using custom service in threads

Board footer