[Lazarus] Run 3 extrenal program, pipeline

[Lukasz Sokol]

On 02/02/2012 17:26, Lukasz Sokol wrote:
> On 02/02/2012 11:00, xmldom at seznam.cz wrote:
>> Hi, I need run 3 external program and redirected output: first | second | third
>>
>> For example: 
>>    df -hT | grep -v -E 'tmpfs|ecryptfs' | grep 'sda2'
>> it gives
>>    /dev/sda2     ext4     19G   11G  7,6G  58% /
>>
>> I use this http://wiki.lazarus.freepascal.org/Executing_External_Programs#How_to_redirect_output_with_TProcess sample code. It works fine for  first | second.
>>
>> For three programs I try edit it:
>>
>>
>> function threePipe(inputA, inputB, inputC: string): string;
>> var
>>  FirstProcess, SecondProcess, ThirdProcess: TProcess;
>>  Buffer: array[0..127] of char;
>>  ReadCount: Integer;
>>  ReadSize: Integer;
>> begin
>>  FirstProcess  := TProcess.Create(nil);
>>  SecondProcess := TProcess.Create(nil);
>>  ThirdProcess := TProcess.Create(nil);
>>
>>  FirstProcess.Options  := [poUsePipes];
>>  SecondProcess.Options := [poUsePipes];
>>  ThirdProcess.Options  := [poUsePipes,poStderrToOutPut];
>>
>>  FirstProcess.CommandLine  := inputA;
>>  SecondProcess.CommandLine := inputB;
>>  ThirdProcess.CommandLine := inputC;
>>
>>  FirstProcess.Execute;
>>  SecondProcess.Execute;
>>  ThirdProcess.Execute;
>>
> 
> while (Firstprocess.Running or SecondProcess.Running or ThirdProcess.Running) do
> begin
>   if Firstprocess.Running then
>     begin
>       if (FirstProcess.Output.NumbytesAvailable > 0) then 
>         begin								
>           ReadSize := FirstProcess.Output.NumBytesAvailable;
>           if ReadSize > SizeOf(Buffer) then 
>             ReadSize := SizeOf(Buffer); 
>           ReadCount := FirstProcess.Output.Read(Buffer[0], ReadSize);
>           SecondProcess.Input.Write(Buffer[0], ReadCount);
>           // at the end of this, we have written max. 128 bytes to SecondProcess' input
>         end;  // if no more bytes, we could terminate the FirstProcess but having it here
>               // may race when the lot starts: when FirstProcess just started but has no output yet
>               // or if for some reason it stops on the way; 
>               // some checking for first process's errors may be a good idea too, here.
>     end
>   else
>     SecondProcess.CloseInput;  // when first process ends, input to second also ends
> 
>   if SecondProcess.Running then
>     begin
>       if (SecondProcess.Output.NumBytesAvailable > 0) then
>         begin
>           ReadSize := SecondProcess.Output.NumBytesAvailable;
>           if ReadSize > SizeOf(Buffer) then  
>             ReadSize := SizeOf(Buffer);  
>           ReadCount := SecondProcess.Output.Read(Buffer[0], ReadSize);
>           ThirdProcess.Input.Write(Buffer[0], ReadCount);
>         end;
>     end
>   else
>     ThirdProcess.CloseInput;
> 	
>   if ThirdProcess.Running then
>     begin
>       if (ThirdProcess.Output.NumbytesAvailable >0) then
>         begin
>           ReadSize := ThirdProcess.Output.NumBytesAvailable;
>           if ReadSize > SizeOf(Buffer) then  
>             ReadSize := SizeOf(Buffer);  
>           if ReadSize > 0 then
>             begin
>               ReadCount := ThirdProcess.Output.Read(Buffer, ReadSize);
>               Write(Copy(Buffer,0, ReadCount));  // i think no writeln, but write (grep gives you ^M on each line IIRC)
>             end;
>         end;
>     end;
> 
> end;
> 
>>  while FirstProcess.Running or (FirstProcess.Output.NumBytesAvailable > 0) do
>>  begin
>>    if FirstProcess.Output.NumBytesAvailable > 0 then
>>    begin
>>      ReadSize := FirstProcess.Output.NumBytesAvailable;
>>      if ReadSize > SizeOf(Buffer) then
>>        ReadSize := SizeOf(Buffer);
>>      ReadCount := FirstProcess.Output.Read(Buffer[0], ReadSize);
>>      SecondProcess.Input.Write(Buffer[0], ReadCount);
>>    end;
>>  end;
>>  SecondProcess.CloseInput;
>>
>>  while SecondProcess.Running or (SecondProcess.Output.NumBytesAvailable > 0) do
>>  begin
>>    if SecondProcess.Output.NumBytesAvailable > 0 then
>>    begin
>>      ReadSize := SecondProcess.Output.NumBytesAvailable;
>>      if ReadSize > SizeOf(Buffer) then
>>        ReadSize := SizeOf(Buffer);
>>      ReadCount := SecondProcess.Output.Read(Buffer[0], ReadSize);
>>      ThirdProcess.Input.Write(Buffer[0], ReadCount);
>>    end;
>>  end;
>>  ThirdProcess.CloseInput;
>>
>>
>>  while ThirdProcess.Running do
>>    Sleep(1);
>>  ReadSize := ThirdProcess.Output.NumBytesAvailable;
>>  if ReadSize > SizeOf(Buffer) then
>>    ReadSize := SizeOf(Buffer);
>>  if ReadSize > 0 then
>>  begin
>>    ReadCount := ThirdProcess.Output.Read(Buffer, ReadSize);
>>    //WriteLn(Copy(Buffer,0, ReadCount));
>>  end
>>  else
>>    WriteLn('grep did not find what we searched for. ', ThirdProcess.ExitStatus);
>>
> 
> //  here;
>>  FirstProcess.Free;
>>  SecondProcess.Free;
>>  ThirdProcess.Free;
>>  threePipe := Copy(Buffer,0, ReadCount); // and this will be nonsense - better have another variable accounting all bytes out from third process
>> end;
>>
>>
>>
>> But after i call 
>>    Label1.Caption:=threePipe('df -hT', 'grep -v -E ''tmpfs|ecryptfs''', 'grep ''sda2''');
>> function my application end with infinite loop.
>>
>> Where I do an error?
>>
Sorry, I hit 'send' button before completing my thoughts.

With program as you wrote it, with your example chain, you wait for df to complete, and then write its 
complete output to grep; This df output can be quite large.
> 
> Thought compiled only.
> 
> Look what I think you need to do:
> - use same buffer to pass between stages, does not stop to read the whole buffer output from first 
> process or second process, just passes buffer
> as it comes
> 
Possible improvement of what I wrote : abstract the chunk where we pass chunks of output of one process 
to another process' input, possibly merge it with the writing to stdout case too (selectable/overload?),
let the whole procedure create as many tprocesses as need be, pass into it a TStringList containing one 
command per string, the associated Object can hold the TProcess reference, result : you can dynamically 
create pipelines as long as you need...

L.