Synchronizing with multiple data requests
lazarus / 38-Networking and Web Programming

----------------------------------------------------------------------
fpcoder · 2015-01-07 19:00

Hi Everyone, how would you solve this:

Data is requested from two or more sources, using threads. All
received data is then merged together, but this must not happen until
*all* requests has finished, either by succeeding or failing.

Q: How to keep track of when all requests has finished? (So the
program can go on and merge available data?)

OOP solutions are fine. I was thinking of storing the requested data
in a "TInfoPack" object and then add all InfoPacks to a "TInfoSet"
when all requests are done.

----------------------------------------------------------------------
marcov · 2015-01-07 19:06

Generally you need a solution that can wait on multiple handles, like
waitformultiple.

Have a TEvent per thread, that they signal when they are ready, then
let the collecting thread block on an array of those tevents.

If a tevent becomes signalled, remove it from the array, and block
again, till all threads are ready.

On *nix this is harder.

----------------------------------------------------------------------
fpcoder · 2015-01-08 11:23

Thanks. Just found RTLEventCreate and its related routines, thats
probably what I should work with?

Quote from: marcov on January 07, 2015, 08:06:47 pmOn *nix this is
harder.

I had in fact planned to run the finished app on Linux Debian. Will
this not work there?

----------------------------------------------------------------------
marcov · 2015-01-08 11:48

Waitformultiple is a windows function.

In *nix this was never very clean cut. There are a lot of questions
for similar functionality (e.g.
http://stackoverflow.com/questions/7780749/waiting-on-multiple-
semaphores-without-busy-waiting-c-c-linux ), but few simple examples.

I never really needed it on *nix, so I haven't investigated. I assume
most programs simply poll using sleeps or fake with pipes and select.

----------------------------------------------------------------------
fpcoder · 2015-01-08 23:51

This appears to work under both linux/windows. But is it reliable?

1) Create and put all threads in an array, then run through the array
and call Start.

2) Then run through the array again and call WaitFor so execution
stops
for the calling thread until all threads has finished.


Code: [Select]procedure TThreadStarter.Execute;
var i1:integer;
begin
  //Start all threads
  for i1:=1 to WORK_THREAD_COUNT do ThreadArray[i1].Start;

  //Then issue a blocking command for each of them
  for i1:=1 to WORK_THREAD_COUNT do ThreadArray[i1].WaitFor;

  SendMessage('ThreadStarter: All WorkerThreads finished!');

  for i1:=1 to WORK_THREAD_COUNT do FreeAndNil(ThreadArray[i1]);

  SendMessage('ThreadStarter: All WorkerThreads destroyed!');

  Form1.btnStartThreads.Enabled:=True;  //Lazy..
end;

I tried it under Linux and it worked same as under windows. Just that
cthreads has to be added to uses.


Whole code:

Code: [Select]unit uMain;

{$mode objfpc}{$H+}

interface

uses
  Classes, SysUtils, FileUtil, Forms, Controls, Graphics, Dialogs,
StdCtrls;

const WORK_THREAD_COUNT = 7;

type
  TThreadMessage = procedure(Sender: TObject; AMessage: string) of
object;

  { TMessageThread }

  TMessageThread = class(TThread)
  private
    FOnMessage: TThreadMessage;  //Method Pointer
  protected
    procedure DoMessage;         //Dispatch Method
  public
    ThreadNo: integer;
    Message: string;
    procedure SendMessage(AMessage: string);
    property OnMessage: TThreadMessage read FOnMessage write
FOnMessage;  //Event
  end;

  { TMyThread }

  TWorkerThread = class(TMessageThread)
  public
    procedure Execute; override;
  end;

  { TThreadStarter }

  TThreadStarter = class(TMessageThread)
  public
    ThreadArray: array[1..WORK_THREAD_COUNT] of TWorkerThread;
    procedure Execute; override;
    constructor Create;
  end;


  { TForm1 }

  TForm1 = class(TForm)
    btnStartThreads: TButton;
    Label1: TLabel;
    lbMessages: TListBox;
    procedure btnStartThreadsClick(Sender: TObject);
  private
    { private declarations }
  public
    ThreadStarter: TThreadStarter;
    procedure OnThreadMessage(Sender: TObject; AMessage: string);
  end;

var
  Form1: TForm1;

implementation

{$R *.lfm}

{ TMessageThread }

procedure TMessageThread.DoMessage;
begin
  if Assigned(FOnMessage) then FOnMessage(Self, Message);
end;

procedure TMessageThread.SendMessage(AMessage: string);
begin
  Message:=AMessage;
  Synchronize(@DoMessage);
end;

{ TWorkerThread }

procedure TWorkerThread.Execute;
begin
  if (not Terminated) or (not Application.Terminated) then
    SendMessage('WorkerThread ' +IntToStr(ThreadNo) +': Executing');
end;


{ TForm1 }

procedure TForm1.btnStartThreadsClick(Sender: TObject);
begin
  btnStartThreads.Enabled:=False;

  lbMessages.Clear;
  lbMessages.Refresh;
  Sleep(70);
  ThreadStarter:= TThreadStarter.Create;
  ThreadStarter.Start;
end;

procedure TForm1.OnThreadMessage(Sender: TObject; AMessage: string);
begin
  lbMessages.Items.Add(AMessage);
end;

{ TThreadStarter }

procedure TThreadStarter.Execute;
var i1:integer;
begin
  //Start all threads
  for i1:=1 to WORK_THREAD_COUNT do ThreadArray[i1].Start;

  //Then issue a blocking command for each of them
  for i1:=1 to WORK_THREAD_COUNT do ThreadArray[i1].WaitFor;

  SendMessage('ThreadStarter: All WorkerThreads finished!');

  for i1:=1 to WORK_THREAD_COUNT do FreeAndNil(ThreadArray[i1]);

  SendMessage('ThreadStarter: All WorkerThreads destroyed!');

  Form1.btnStartThreads.Enabled:=True;  //Lazy..
end;

constructor TThreadStarter.Create;
var i1:integer;
begin
  inherited Create(True);
  FreeOnTerminate:=True;
  OnMessage:=@Form1.OnThreadMessage;

  for i1:=1 to WORK_THREAD_COUNT do
  begin
    ThreadArray[i1]:= TWorkerThread.Create(True, 131072);
    ThreadArray[i1].OnMessage:=@Form1.OnThreadMessage;

    ThreadArray[i1].ThreadNo:=i1;
    ThreadArray[i1].FreeOnTerminate:=False;

    SendMessage('ThreadStarter: WorkerThread ' +IntToStr(i1) +'
Created');
  end;
end;

end.