Synchronize with MTProcs -> selfmade multithreading

[Ocye]

I'm using the great tool Multithreadproc. It scales perfectly and boost the calculation time significantly. My problem is that MTProcs has no Synchronize. During the operation no feedback can be given to the user and the program seems to hang. Any idea how to implement interaction into ProcThreadPool.DoParallel()?

[Ocye]

For those who are interested in my solution:

Code: [Select]

procedure TBruteforce.RunParallel(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
begin
  ProcThreadPool.EnterPoolCriticalSection;
  try
    Application.ProcessMessages; //e.g.
  finally
    ProcThreadPool.LeavePoolCriticalSection;
  end;
  TSingleOperation(FProcessing[index]).Run;
end;

FProcessing is a TList filled with items of TSingleOperation=class with parameters and a run method. Enter- and LeavePoolCriticalSection did the "trick" for interaction.

[Ocye]

Rejoyed too soon! With Linux no synch error occours. Under Windows I get an exception "CheckSynchronization called from non-main thread" or program hangs. Now I remember what the problem is . There is still no way for processing messages in the main thread.

[José Mejuto]

Rejoyed too soon! With Linux no synch error occours. Under Windows I get an exception "CheckSynchronization called from non-main thread" or program hangs. Now I remember what the problem is . There is still no way for processing messages in the main thread.

Hello,

You can not in anyway call ProcessMessages from something different than main thread, that's a task for Synchronize. I had not used it in MTProc but you should be able to call Synchronize or take the current thread object from "Item" parameter and synchronize using it.

[Ocye]

As I said, with Linux ProcessMessages can be called without problems, but, of course, it's not correct. I tried Item.Thread.Synchronize() yesterday at different positions but the application freezes completely. I created a callback method in mtprocs.pas (OnBeforeWork) and invoke it with Synchronize.

Code: [Select]

procedure TMT...SynchMethod;
begin
  if assigned(FOnBeforeWork) then FOnBeforeWork;
end;

  AThread.Sychronize(AThread.SynchMethod);
  AThread.Resume;
It works partially if I place it just before (or was it after?) Resume() but not if call it within a CriticalSection (any other position). Maybe it's not possible to Synchronize a locked thread, which would be reason for freeze at Item.Thread.Synchronize as well.
On the other hand, a beta tester reported no success with that implementation (Vista).

[José Mejuto]

As I said, with Linux ProcessMessages can be called without problems, but, of course, it's not correct. I tried Item.Thread.Synchronize() yesterday at different positions but the application freezes completely. I
[...]
On the other hand, a beta tester reported no success with that implementation (Vista).

Hello,

You are right, I was only able to perform some kind of "synchronization" using PostMessage as even SendMessage finally hangs.

I was calling it:

Code: [Select]

[....]
GetMem(DataPointer,1024);
pchar(DataPointer):=TimeToStr(Now);
Postmessage(frmMain.Handle,WM_USER,DataPointer,0);
[....]

and in frmMain:

Code: [Select]

Procedure CustomMessage(var MSG: TMessage); message WM_USER;
begin
  Self.Caption:=pchar(Msg.WParam);
  FreeMem(Msg.WParam);
end;

It is a dirty hack, it works but its far away from confortable. Only tested in WinXP Sp3.

[Ocye]

Thanks for your help, Joshy.
WParam is a PtrInt, no GetMem needed. If I send WM_USER with WPARAM=Now I get the message after calculation not during - Linux 64bit.
I don't understand why a user message should be preferred if even mouse events aren't evaluated. I need to show the calculation progress and want to give a chance to abort since calculation could take some minutes. Perhaps I should implement multithreading by myself with Waitfor().

[cdbc]

Hi Ocye

Have a look at Primoz' blog/pages...
http://17slon.com/blogs/gabr/2009/02/omnithreadlibrary-103.html

Omnithread library. Primoz is a very skilled programmer... :-)

HtH
Regards Benny

[José Mejuto]

Thanks for your help, Joshy.
WParam is a PtrInt, no GetMem needed. If I send WM_USER with WPARAM=Now I get the message after calculation not during - Linux 64bit.
I don't understand why a user message should be preferred if even mouse events aren't evaluated. I need to show the calculation progress and want to give a chance to abort since calculation could take some minutes. Perhaps I should implement multithreading by myself with Waitfor().

Hello,

It should be processed when the mainform queue is empty, which should happend almost of the time when running a process in a different thread, but for some reason seems that the main thread is not respawned after posting message, and sendmessage hangs waiting for completation (Windows).

I'm not 100% sure but I think that Syncronize and family is broken in FPC 2.5.1 I'll take a closer look.

If you have only one extra thread running it is better to use the TThread class.

For process notification and cancel threads I'm using always a pooling method, this means that the thread updates some variables every large round, or every X milliseconds and the main thread reads that values every 500 ms. or so.

[Ocye]

I have between 2 and 3375 single operations to run. Each takes some seconds depending on hardware and task. MTProcs blocks the application completely and no message is processed. Synchronize works since all network stuff is okay. I'm wondering why Item.Thread.Synchonize hangs. Maybe OmniThread fits better, I'll have a look.

[Ocye]

OmniThread seems to be a toolkit for Delphi only. Finally, I realized multithreading myself - quick and dirty but it works.

Code: [Select]

type
  TOnAddBest = procedure(...) of Object;

  TSingleOperation=class(...)
    public
       procedure Run;
    end;

  TBruteForce=class
      FProcessing   : TList; //list of TSingleOperation
      FMaxThreads : integer;//assigned by GetSystemThreadCount() of MTProcs ->MTCPU
      procedure SetupProcessing;//fills FProcessing
      procedure RunThreaded(aStart:word=0);
      procedure AddBest(...);
    public
      function ComputeMove:boolean
    end;

  TSingleThread=class(TThread)
    private
      FOperation : TSingleOperation;
    protected
      procedure Execute; override;
    public
      constructor Create(aOperation:TSingleOperation);
    end;

implementation

var
  {$ifdef CPU32}
    aCriticalSection:LongWord;
  {$else}
    aCriticalSection:QWord;
  {$endif}
  ThreadsRunning : integer;

{ --- TBruteForce --- }

procedure TBruteForce.RunThreaded(aStart: word);
begin
  if aStart=0 then ThreadsRunning:=0;
  if aStart<FProcessing.Count then
  begin
    with TSingleThread.Create(TSingleOperation(FProcessing[aStart])) do
    try
      Resume;
      if (ThreadsRunning>FMaxThreads) or "no operation left = last thread" then Waitfor;
      OnProgress(nil,round((aStart/FProcessing.Count)*100));
      Application.ProcessMessages;
      RunThreaded(aStart+1);
    finally
      Free;
    end;
  end;
end;

function TBruteForce.ComputeMove:boolean;
begin
  FMaxThreads:=GetSystemThreadCount;
  SetupProcessing;
  ...
  if fmGameOptions.cbParallel.Checked then          //run parallel
  begin
    InitializeCriticalSection(aCriticalSection);
    try
      RunThreaded;
    finally
      DeleteCriticalSection(aCriticalSection);
    end;
  end else
  begin                                                           //run single threaded
    for i:=0 to FProcessing.Count-1 do
    begin
      OnProgress(self,round((i/FProcessing.Count)*100));
      Application.ProcessMessages;
      TSingleOperation(FProcessing[i]).Run(nil);
    end;
  end;
  ...
end;

{ --- TSingleOperation --- }

procedure TSingleOperation.Run(aThread:TThread);
begin
  ... //loops and calculations
  if aThread<>nil then EnterCriticalsection(aCriticalSection);
//  if aThread<>nil then aThread.Synchronize(aThread,@SynchMethod); //fails
  try
    if assigned(FOnAddBest) then FOnAddBest(...); //data exchange
  finally
    if aThread<>nil then LeaveCriticalSection(aCriticalSection);
  end;
  ...
end;

{ --- TSingleThread --- }

constructor TSingleThread.Create(aOperation: TSingleOperation);
begin
  inherited Create(true);
  FOperation:=aOperation;
end;

procedure TSingleThread.Execute;
begin
  inc(ThreadsRunning);
  FOperation.Run(self);
  dec(ThreadsRunning);
end;

If I call Synchronize program blocks; debugging, which should not be trusted, reports an runerror after Thread.Execute. Strange...

[mas steindorff]

It sounds like your far more advanced than I am with threading but my fix to a similar problem was to place a sleep(1) in the Execute procedure loop of the object.  that would release the CPU so it would check other processes including the main code. 
It did however slow down the thread.  I added a check to my port input thread to only sleep if the thread was not too far behind on it's work load and that worked for me.  In my case, I needed as much CPU clocks as I could get to handle the streeming data.
good luck.

[mattias]

Synchronize works for MTProcs just like for any other TThread.
Synchronize requires that the main thread still runs.
Solution:
Create a worker thread and execute the DoParallel there.

[Ocye]

Synchronize doesn't work as expected, see Joshy's posting. Example:

Code: [Select]

procedure TForm1.Button1Click(Sender: TObject);
begin
  test:=0;
  ProcThreadPool.DoParallel(@MyParallelProc,0,MaxInt,nil);
end;

procedure TForm1.MySynchMethod;
begin
  label1.caption:=inttostr(test);
end;

procedure TForm1.MyParallelProc(Index: PtrInt; Data: Pointer; Item: TMultiThreadProcItem);
begin
  inc(test);
//  Item.Thread.Synchronize(Item.Thread,@MySynchMethod); -> SIGSEV
end;

Maybe it's not a question of MTProcs rather than threading itself, I don't know.
BTW: Please add a short sentence to the wiki how Synchronize can be used with MTProcs (per Item.Thread). I was looking for that a long time  
Thanks for your work, Matthias!

[José Mejuto]

Synchronize doesn't work as expected, see Joshy's posting. Example:
[...]
Maybe it's not a question of MTProcs rather than threading itself, I don't know.
BTW: Please add a short sentence to the wiki how Synchronize can be used with MTProcs (per Item.Thread). I was looking for that a long time  
Thanks for your work, Matthias!

Hello,

I had understand the explanation, MTProcs dispatcher runs in main thread, well, in fact, in the current thread, so when you run "DoParallel" the main thread is blocked and syncronize will not happend. Solution run the DoParallel inside a Execute of a TThread. Something like:

MyParallelThread=class(TThread)
...
procedure MyParallelThread.Execute;
begin
  DoParallel(....)
end;
[......]
procedure Form1.Button1Click();
var
  Worker: TMyParallelThread;
begin
  Worker:=TMyParallelThread.Create(true,false,...); //Start paused
  Worker.FreeOnTerminate:=true;
  Worker.Resume;
end;

This way when synchronize should work.