Moving from TThread.Synchronize to TThread.Queue

[Remy Lebeau]

and that works?

Yes.  This is the exact same principle that Indy's TIdNotify uses, and it works fine.  Here is what TIdNotify looks like (this is a slimmed down representation of its actual code, which is otherwise full of IFDEFs to handle differences between Delphi/Win32, Delphi/Mobile, and FreePascal):

Code: Pascal  [Select][+][-]

1. type
2.   TIdNotify = class(TObject)
3.   protected
4.     ...
5.     //
6.     procedure DoNotify; virtual; abstract;
7.     procedure InternalDoNotify;
8.   public
9.     constructor Create; virtual; // here to make virtual
10.     procedure Notify;
11.     ...
12.   end;
13.  
14. constructor TIdNotify.Create;
15. begin
16.   inherited Create;
17. end;
18.  
19. procedure TIdNotify.Notify;
20. begin
21.   try
22.     TThread.Queue(nil, InternalDoNotify);
23.   except
24.     Free;
25.     raise;
26.   end;
27. end;
28.  
29. procedure TIdNotify.InternalDoNotify;
30. begin
31.   try
32.     DoNotify;
33.   finally
34.     Free;
35.   end;
36. end;
37.  

The code you posted looks to me that it will free the processor before it has finished processing

No, it will not.  If TThread.Queue() fails to add TDataProcessor.DoProcess() to the queue, it raises an exception, so the TDataProcessor object is freed immediately as its DoProcess() will not be called.  Otherwise, the TDataProcessor object stays in memory until the main thread calls TDataProcessor.DoProcess(), which then calls TDataProcessor.ProcessData() and then frees the TDataProcessor object.

For instance, my earlier example can be re-written to use TIdNotify like this:

Code: Pascal  [Select][+][-]

1. type
2.   TDataProcessor = class(TIdNotify)
3.   private
4.     FData : TStringList;
5.     procedure DoNotify; override;
6.   public
7.     constructor Create(var AData: TStringList); reintroduce;
8.     destructor Destroy; override;
9.   end;
10.  
11. constructor TDataProcessor.Create(var AData: TStringList);
12. begin
13.   inherited Create;
14.   FData := AData;
15.   AData := nil;
16. end;
17.  
18. destructor TDataProcessor.Destroy;
19. begin
20.   FData.Free;
21.   inherited;
22. end;
23.  
24. procedure TDataProcessor.DoNotify;
25. begin
26.   // do something with FData
27. end;
28.  
29. procedure MyThread.Execute;
30. var
31.   Data: TStringList;
32.   Processor: TDataProcessor;
33. begin
34.   while not Terminated do
35.   begin
36.     Data := TStringList.Create;
37.     try
38.       ReceiveData(Data); // ReceiveData will block until some data is received
39.       TDataProcessor.Create(Data).Notify;
40.     except
41.       Data.Free;
42.     end;
43.   end;
44. end;
45.  

What am I missing? Did you mend to use synchronise?

No, the whole point of my example was to demonstrate passing a TStringList to the main thread using TThread.Queue().

[taazz]

and that works?

Yes.  This is the exact same principle that Indy's TIdNotify uses, and it works fine, for example (this is a slimmed down representation of TIdNotify's actual code, which is full of IFDEFs):

The whole point of my example was to demonstrate passing a TStringList to the main thread using TThread.Queue().

Ok, I was reading try..finaly instead of try..except. I see it now, thank you.

[cpicanco]

Assuming that Queue would optimize data receivement, a problem you may face is that you must copy FData somehow to avoid overrides.

QUESTION

How to make such a copy???

For example, in Indy, its TIdNotify class (which wraps TThread.Queue()) sends queued data by implementing the queued method inside an object that gets freed when the queued method exits.  You can do something similar, eg:

Code: Pascal  [Select][+][-]

1. type
2.   TDataProcessor = class
3.   private
4.     FData : TStringList;
5.     procedure DoProcess;
6.     procedure ProcessData;
7.   public
8.     constructor Create;
9.     destructor Destroy; override;
10.     procedure Queue;
11.     property TStringList Data read FData;
12.   end;
13.  
14. constructor TDataProcessor.Create;
15. begin
16.   inherited
17.   FData := TStringList.Create;
18. end;
19.  
20. destructor TDataProcessor.Destroy;
21. begin
22.   FData.Free;
23.   inherited;
24. end;
25.  
26. procedure TDataProcessor.DoProcess;
27. begin
28.   try
29.     ProcessData;
30.   finally
31.     Free;
32.   end;
33. end;
34.  
35. procedure TDataProcessor.ProcessData;
36. begin
37.   // do something with FData
38. end;
39.  
40. procedure TDataProcess.Queue;
41. begin
42.   TThread.Queue(nil, @DoProcess);
43. end;
44.  
45. ...
46.  
47. procedure MyThread.Execute;
48. var
49.   Processor: TDataProcessor;
50. begin
51.   while not Terminated do
52.   begin
53.     Processor := TDataProcessor.Create;
54.     try
55.       ReceiveData(Processor.Data); // ReceiveData will block until some data is received
56.       Processor.Queue;
57.     except
58.       Processor.Free; // not queued, free now...
59.     end;
60.   end;
61. end;
62.  
63. procedure MyThread.ReceiveData(AData: TStringList);
64. begin
65.   AData.Add(...);
66. end;
67.  

I will try to write a fully working example based on this.

PS.: I am really enjoying the discussion!

[cpicanco]

TThread.Queue

class procedure
 TThread.Queue(
  aThread: TThread;
  aMethod: TThreadMethod
);

In the class procedure overloaded version of this call, the thread for which the method must be posted is the first argument. In the protected version of this call (used in the tthread instance), this argument is not there, and the thread instance is used.

Hi I just want to make something clear. If aThread is nil then CurrentThread is used:

// classes.inc
class procedure TThread.Queue(aThread: TThread; aMethod: TThreadMethod); static;
var
  queueentry: PThreadQueueEntry;
begin
  { ensure that we have a valid TThread instance }
  if not Assigned(aThread) then
    aThread := CurrentThread;
...
 

But it is not clear what CurrentThread is:

class function TThread.GetCurrentThread: TThread;
begin
  { if this is the first time GetCurrentThread is called for an external thread
    we need to create a corresponding TExternalThread instance }
  Result := CurrentThreadVar;
  if not Assigned(Result) then begin
    Result := TExternalThread.Create;
    CurrentThreadVar := Result;
  end;
end;   

 Is CurrentThreadVar always the main thread?

[taazz]

Is CurrentThreadVar always the main thread?

No from the fragments you posted it looks like a random thread created just to be used in the queue method for no apparent reason. I would have to look closer the code to formulate an opinion but so far looks like a leftover from a time before the queue method was a class method or something along those lines.

[molly]

@taazz:
Forgive me if i've read your post wrongly.

I think the trick there is that when you don't supply the nil (e.g. using self) parameter that the execution of the queue is linked to the thread itself. Meaning that if you terminate the thread before the queue is processed it will remove them from the queuelist.

But, you do not want that to happen because it _must_ be processed. Hence linking the execution of queue items to another (alive) thread.

You can also solve that with a call to synchronize after the termination loop btw, but it seems a nice trick of RL to have a separate processdata class.

In my test i used my own TQueue and initially waited until that list was empty before actually terminating the thread.

[taazz]

@taazz:
Forgive me if i've read your post wrongly.

It was a mistake the moment I post it with out checking the code and a double at that, adding conjecture with facts creating an unstable combination. So don't apologise for my mistakes.

I think the trick there is that when you don't supply the nil (e.g. using self) parameter that the execution of the queue is linked to the thread itself.

This one does not make sense to me. The execution of the Queue is linked to the thread? why?

Meaning that if you terminate the thread before the queue is processed it will remove them from the queuelist.

Really? what happens if I queued a method of an external class like the application's main form? The moment the thread is destroyed the method is removed from the queue? It can't be.

But, you do not want that to happen because it _must_ be processed. Hence linking the execution of queue items to another (alive) thread.

I agree with the first part, the code must be executed, I can't understand the logic behind the mechanism though. I do not see what it protects at this point.
Then again I haven't read the code yet.

You can also solve that with a call to synchronize after the termination loop btw, but it seems a nice trick of RL to have a separate processdata class.
In my test i used my own TQueue and initially waited until that list was empty before actually terminating the thread.

I'm guessing you assume that the method queued from the thread is part of the thread it self and not part of an other class.

Well just my ignorant point of view at this point. I'll take a closer look from home and add my findings.

[molly]

I'm guessing you assume that the method queued from the thread is part of the thread it self and not part of an other class.

i'm no expert but to my knowledge,

no not the method but the call itself is placed into a queuelist.

When you invoke TThread.Queue and afaik the call itself will be placed into a list owned by the thread. As long as the thread exist this list is being processed like fifo. Therefor this list grows.

The hidden self parameter in the call to tthread.queue refers to the thread itself. Passing nil ... well ... you saw the code posted.

As soon as you terminate the thread the calls places into the queuelist will be removed so that the responsible thread is able to terminate immediately.

[cpicanco]

How to get a TThread from MainThreadID, so I can ensure that Queue will be processed by the main thread?

edit:

Wharever, I will test like this:

Code: Pascal  [Select][+][-]

1. function TDataProcessor.Queue: TThreadMethod;
2. begin
3.   Result := @DoProcess;
4. end;
5.              
6. // TThread.Execute
7. var
8.    Processor : TDataProcessor;
9. ...
10. Queue(Processor.Queue);
11. ...
12.  

[molly]

How to get a TThread from MainThreadID, so I can ensure that Queue will be processed by the main thread?

By passing nil to the call to queue, as shown by RL's code:

Code: [Select]

procedure TDataProcess.Queue;
begin
  TThread.Queue(nil, @DoProcess);
end;


Although i am not 100% sure if that will be the main thread. to my knowledge it is.

You can easily test this situation yourself. Make a long delay inside your DoProcess(), and test with terminating your thread prematurely. Do it once with nil and once without nil parameter to notice the difference.

[Thaddy]

Although i am not 100% sure if that will be the main thread. to my knowledge it is.

Afaik it is the creating thread (almost but not always the main thread in light or noob scenario's) that obtains it. So the owning thread of the thread. the thread in which you created your thread...).
Even that only holds true with all defaults.

The example by Remy is really the simplest I can think of that cuts wood. Simpler is neigh impossible.

[molly]

Ah, ok. that makes perfect sense.

Thank you clarifying that Thaddy.

[cpicanco]

1

unit DataProcessor;

{$mode objfpc}{$H+}

interface

uses
  Classes, SysUtils;

type

  { TDataProcessor }

  TDataProcessor = class
  private
    FData : TStringList;
    procedure DoProcess;
  protected
    procedure ProcessData; virtual; abstract;
  public
    constructor Create;
    destructor Destroy; override;
    procedure Queue;
    property Data : TStringList read FData;
  end;

implementation

procedure TDataProcessor.DoProcess;
begin
  try
    ProcessData;
  finally
    Free;
  end;
end;

constructor TDataProcessor.Create;
begin
  FData := TStringList.Create;
end;

destructor TDataProcessor.Destroy;
begin
  FData.Free;
  inherited Destroy;
end;

procedure TDataProcessor.Queue;
begin
  TThread.Queue(nil, @DoProcess);
end;


end.

2

unit zmq_network;

{$mode objfpc}{$H+}

interface

uses Classes, SysUtils, DataProcessor;

type     

  TMessRecvProc = procedure(AResponse: TStringList) of object;

  // right now you must create a class for each event you want to queue
  TMessageProcessor = class sealed(TDataProcessor)
  private
    FOnProcessData: TMessRecvProc;
  protected
    procedure ProcessData; override;
  public
    property OnProcessData : TMessRecvProc read FOnProcessData write FOnProcessData;
  end;       

  TZMQMessagesThread = class sealed(TThread)
  private
    FContext : Pointer;
    FSubscriber,
    FOnMessageReceived: TMessRecvProc;
  protected
    procedure Execute; override;
  public
    constructor Create(CreateSuspended: Boolean = True); overload;
    destructor Destroy; override;
    property OnMessageReceived : TMessRecvProc read FOnMessageReceived write FOnMessageReceived;
  end;   
           
implementation

uses zmq, zmq.helpers;   

procedure TMessageProcessor.ProcessData;
begin
  if Assigned(OnProcessData) then OnProcessData(Data);
end;
       
procedure TZMQMessagesThread.Execute;
var
  rc: integer;
  LMessage : TMessageProcessor;
  item : zmq_pollitem_t;
begin
  with item do
  begin
    socket := FSubscriber;
    fd := 0;
    events := ZMQ_POLLIN;
    revents := 0;
  end;

  while not Terminated do
  begin
    rc := zmq_poll(item, 1, -1); // block thread until some message arrive, deprecated, you should use the new poller api
    if rc = 0 then continue;
    if (item.revents and ZMQ_POLLIN) > 0 then
      begin
        LMessage := TMessageProcessor.Create;
        LMessage.OnProcessData:=FOnMessageReceived;
        RecvMultiPartString(FSubscriber, LMessage.Data);
        try
          LMessage.Queue;
        except
          LMessage.Free;
        end;
      end;
  end;
end;


constructor TZMQMessagesThread.Create(CreateSuspended: Boolean);
begin
  FreeOnTerminate := True;
  FContext := zmq_ctx_new;

  // zmq subscriber
  FSubscriber := zmq_socket(FContext, ZMQ_SUB);
  zmq_connect(FSubscriber, PChar('<zmq_bind host goes here>'));

  // subscribe to all messages
  zmq_setsockopt(FSubscriber, ZMQ_SUBSCRIBE, nil, 0);

  inherited Create(CreateSuspended);
end;

 

2 is not complete, but ilustrates a test I have done with a program here. It worked.

Queue(Processor.Queue) and Queue(@Someprocedure) both did not work in the context of the mentioned program, it generates unsynchronized behavior. But may work in a different context. A self contained example would be great, I will do it eventually.

[Mr.Madguy]

@Mr.Madguy,

def yes and def no. I still have 2 TRS-80, the onley interrupt handler is used by random number generator, formal "IM 1" in Z80 CPU assembly language( interrupt masked 1), all other in/out from hardware are performed by cycling ports data. on my amstrad cpc, OK, there is plenty interrupts around ASIC component

under L-DOS (198~) there is a type ahead capability so you can hit keystrokes while diskettes are turning BUT the buffer is 10 keys long, if you type 11 or more keys,the 11th and the others are simply trashed into intergalactic blackbyte holes

SO whatever you write as container EVEN NOWADAYS can not be general nor generic on big amount of data EXCEPT IF you tell the sender (computer or hardware component) to slow down when buffers are flooded. It must be able to understand this order to not hang your container. In anycase you need a mechanism that is well known for centuries
===> Waitfor or Synchronize (that is one and one only concept)

OTHERWISE you write unsafe code happily easy to attack with DDOS virus concept

You write about something without even researching and understanding, how it works. Again, Synchronize blocks calling thread and that turns it essentially into just an event waiting thread. I.e. it exists only to avoid main thread blocking by waiting cycle - all data processing is being performed in main thread anyway, so data processing is essentially serial. This isn't, what threads are intended to be used for. Threads are intended to be used for PARALLEL processing.

Have you ever heard about conveyor or pipeline design? This is design, that is intended to minimize idling, i.e. ineffective working. When your main thread processes data, your calling thread is blocked, i.e. essentially idles, while it can possibly do something useful - for example fetch more data from hardware. Buffering - is the only way of achieving this goal. Even in real life buffer warehouses are used in order to avoid production process interruption and idling.

Buffers and queues are actually everywhere. Window message queue - is greatest example.

Also, there is no risk of DDOS attack, when using queue. Queue has limited size and when this limit is reached - calling thread starts being blocked by default as in Synchronize case. Queue size is usually compromise between memory consumption and effectivity. It's size is picked according to peak load. There is an option to set read/write timeout though, if you need to do something else, while waiting.

[taazz]

I'm guessing you assume that the method queued from the thread is part of the thread it self and not part of an other class.

i'm no expert but to my knowledge,

no not the method but the call itself is placed into a queuelist.

lets see what the code tells us.

Code: Pascal  [Select][+][-]

1. class procedure TThread.Queue(aThread: TThread; aMethod: TThreadMethod); static;
2. var
3.   queueentry: PThreadQueueEntry;
4. begin
5.   { ensure that we have a valid TThread instance }
6.   if not Assigned(aThread) then
7.     aThread := CurrentThread;
8.  
9.   New(queueentry);
10.   FillChar(queueentry^, SizeOf(TThreadQueueEntry), 0);
11.   queueentry^.Thread := aThread;
12.   queueentry^.Method := aMethod;
13.  
14.   { the queueentry is freed by CheckSynchronize (or by RemoveQueuedEvents) }
15.   ThreadQueueAppend(queueentry);
16. end;
17.  

Well my pascal might not be that good, but to me it looks like that, only the method is attached to the queue. Not the call, the method to be called, is marked with the thread and is placed in the list.

When you invoke TThread.Queue and afaik the call itself will be placed into a list owned by the thread. As long as the thread exist this list is being processed like fifo. Therefor this list grows.

Well that makes no sense to me so lets look at the ThreadQueueAppend code to see what goes on.

Code: Pascal  [Select][+][-]

1. procedure ThreadQueueAppend(aEntry: TThread.PThreadQueueEntry);
2. begin
3.   { do we really need a synchronized call? }
4.   if GetCurrentThreadID = MainThreadID then begin
5.     ExecuteThreadQueueEntry(aEntry);
6.     if not Assigned(aEntry^.SyncEvent) then
7.       Dispose(aEntry);
8.   end else begin
9.     System.EnterCriticalSection(ThreadQueueLock);
10.     try
11.       { add the entry to the thread queue }
12.       if Assigned(ThreadQueueTail) then begin
13.         ThreadQueueTail^.Next := aEntry;
14.       end else
15.         ThreadQueueHead := aEntry;
16.       ThreadQueueTail := aEntry;
17.     finally
18.       System.LeaveCriticalSection(ThreadQueueLock);
19.     end;
20.  
21.     { ensure that the main thread knows that something awaits }
22.     RtlEventSetEvent(SynchronizeTimeoutEvent);
23.     if assigned(WakeMainThread) then
24.       WakeMainThread(aEntry^.Thread);
25.  
26.     { is this a Synchronize or Queue entry? }
27.     if Assigned(aEntry^.SyncEvent) then begin
28.       RtlEventWaitFor(aEntry^.SyncEvent);
29.       if Assigned(aEntry^.Exception) then
30.         raise aEntry^.Exception;
31.     end;
32.   end;
33. end;
34.  

Well alarm bell No 1. If the thread that is queuing a method is the main thread the method is executed on the spot and never queued. That is a huge bug. The only reason one will queue a method from the main thread is to be executed on a later time, after the current process has finished and the code forces it to be executed now. Wow!! Noted but that is not todays concern, on to the problem at hand.
the code first locks access to the thread with a critical section then uses the variable ThreadQueueTail that is not declared in the procedure it self to append the queued method to what it looks like a single linked list. hmm the only way this could be thread bound is if the variables threadQueueHead and ThreadQueueTails are some how re declared for each thread the only construct that I know of that can do that is a threadvar lets find out the declaration of those variables

Code: Pascal  [Select][+][-]

1. var
2.   { event executed by SychronizeInternal to wake main thread if it sleeps in
3.     CheckSynchronize }
4.   SynchronizeTimeoutEvent: PRtlEvent;
5.   { the head of the queue containing the entries to be Synchronized - Nil if the
6.     queue is empty }
7.   ThreadQueueHead: TThread.PThreadQueueEntry;
8.   { the tail of the queue containing the entries to be Synchronized - Nil if the
9.     queue is empty }
10.   ThreadQueueTail: TThread.PThreadQueueEntry;
11.   { used for serialized access to the queue }
12.   ThreadQueueLock: TRtlCriticalSection;
13.   { this list holds all instances of external threads that need to be freed at
14.     the end of the program }
15.   ExternalThreads: TThreadList;
16.   { this list signals that the ExternalThreads list is cleared and thus the
17.     thread instances don't need to remove themselves }
18.   ExternalThreadsCleanup: Boolean = False;
19.  
20.   { this must be a global var, otherwise unwanted optimizations might happen in
21.     TThread.SpinWait() }
22.   SpinWaitDummy: LongWord;
23.  
24. threadvar
25.   { the instance of the current thread; in case of an external thread this is
26.     Nil until TThread.GetCurrentThread was called once (the RTLs need to ensure
27.     that threadvars are initialized with 0!) }
28.   CurrentThreadVar: TThread;
29.  
30.  
31.  

It looks like a single list for all threads to me,l as expected.

The hidden self parameter in the call to tthread.queue refers to the thread itself. Passing nil ... well ... you saw the code posted.

So far we found out that when the queue method exits, the method queued is already ready for execution, and plays no farther role in the list or execution of the application. So the main question now is why does the list needs to know which thread added which method to the list so I looked at the checkSynchronise procedure and found nothing that uses that information. So there mast be something else that uses it and after a bit of searching I found a couple of methods that use that information and guess what they are all methods of TThread that remove methods from the queue.
I quess that is helpfull, there might be a reason you want a previously queue method not to be executed, although in its current implementation it will remove all entries of that method from the queue not only the first or the last or the one with data = X which makes it a bit limited.

As soon as you terminate the thread the calls places into the queuelist will be removed so that the responsible thread is able to terminate immediately.

Yeap you are write here is the destructor of TThread.

Code: Pascal  [Select][+][-]

1. destructor TThread.Destroy;
2. begin
3.   if not FExternalThread then begin
4.     SysDestroy;
5.     if FHandle <> TThreadID(0) then
6.       CloseThread(FHandle);
7.   end;
8.   RemoveQueuedEvents(Self);
9.   DoneSynchronizeEvent;
10.   { set CurrentThreadVar to Nil? }
11.   inherited Destroy;
12. end;
13.  

Now knowing that the queue method has no role what so ever in the list after it finishes executing, knowing that the queued method might be anything from any class or worst no class, knowing that the only peace of code that checks who added the method in the queue is in TThread it self, can some one please explain to me what is this suppose to protect? Oh and since you are on it, please explain to me it how is this queue processed from a lazarus application too, I did not find any thing in the TApplication class to follow and TApplication.QueueAsyncCall has its own list which might introduce a different behavior at the end.