Moving from TThread.Synchronize to TThread.Queue

[Mr.Madguy]

CheckSynchronize is called from Application.ProcessMessages.

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

1. procedure TApplication.Run;
2. begin
3.   if (FMainForm <> nil) and FShowMainForm then FMainForm.Show;
4.   WidgetSet.AppRun(@RunLoop);
5. end;
6.  
7. procedure TApplication.RunLoop;
8. begin
9.   repeat
10.     if CaptureExceptions then
11.       try // run with try..except
12.         HandleMessage;
13.       except
14.         HandleException(Self);
15.       end
16.     else
17.       HandleMessage; // run without try..except
18.   until Terminated;
19. end;
20.  

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

1. procedure TApplication.HandleMessage;
2. begin
3.   WidgetSet.AppProcessMessages; // process all events
4.   if not Terminated then Idle(true);
5. end;  
6.  

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

1. procedure TWin32WidgetSet.AppProcessMessages;
2. var
3.   AMessage: TMsg;
4.   retVal, index: dword;
5.   pHandles: Windows.LPHANDLE;
6.  
7.     procedure CallWaitHandler;
8.     begin
9.       FWaitHandlers[index].OnEvent(FWaitHandlers[index].UserData, 0);
10.     end;
11.  
12. begin
13.   repeat
14.     if FPendingWaitHandlerIndex >= 0 then
15.     begin
16.       index := FPendingWaitHandlerIndex;
17.       FPendingWaitHandlerIndex := -1;
18.       CallWaitHandler;
19.     end;
20. {$ifdef DEBUG_ASYNCEVENTS}
21.     if Length(FWaitHandles) > 0 then
22.       DebugLn('[ProcessMessages] WaitHandleCount=', IntToStr(FWaitHandleCount),
23.         ', WaitHandle[0]=', IntToHex(FWaitHandles[0], 8));
24. {$endif}
25.     if FWaitHandleCount > 0 then
26.       pHandles := @FWaitHandles[0]
27.     else
28.       pHandles := nil;
29.     retVal := Windows.MsgWaitForMultipleObjects(FWaitHandleCount,
30.       pHandles, False, 0, QS_ALLINPUT);
31.     if (retVal < WAIT_OBJECT_0 + FWaitHandleCount) then
32.     begin
33.       index := retVal-WAIT_OBJECT_0;
34.       CallWaitHandler;
35.     end else
36.     if retVal = WAIT_OBJECT_0 + FWaitHandleCount then
37.     begin
38.       while PeekMessage(AMessage, HWnd(nil), 0, 0, PM_REMOVE) do
39.       begin
40.         if AMessage.message = WM_QUIT then
41.         begin
42.           PostQuitMessage(AMessage.wParam);
43.           break;
44.         end;
45.         TranslateMessage(@AMessage);
46.         if UnicodeEnabledOS then
47.           DispatchMessageW(@AMessage)
48.         else
49.           DispatchMessage(@AMessage);
50.       end;
51.     end else
52.     if retVal = WAIT_TIMEOUT then
53.     begin
54.       // check for pending to-be synchronized methods
55.       CheckSynchronize;
56.       CheckPipeEvents;
57.       break;
58.     end else
59.     if retVal = $FFFFFFFF then
60.     begin
61.       DebugLn('[TWin32WidgetSet.AppProcessMessages] MsgWaitForMultipleObjects returned: ', IntToStr(GetLastError));
62.       break;
63.     end;
64.   until false;
65. end;          
66.  

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

1. function CheckSynchronize(timeout : longint=0) : boolean;
2.  
3. { assumes being called from GUI thread }
4. var
5.   ExceptObj: Exception;
6.   tmpentry: TThread.PThreadQueueEntry;
7.  
8. begin
9.   result:=false;
10.   { first sanity check }
11.   if Not IsMultiThread then
12.     Exit
13.   { second sanity check }
14.   else if GetCurrentThreadID<>MainThreadID then
15.     raise EThread.CreateFmt(SCheckSynchronizeError,[GetCurrentThreadID]);
16.   if timeout>0 then
17.     RtlEventWaitFor(SynchronizeTimeoutEvent,timeout)
18.   else
19.     RtlEventResetEvent(SynchronizeTimeoutEvent);
20.   tmpentry := PopThreadQueueHead;
21.   while Assigned(tmpentry) do
22.     begin
23.     { step 2: execute the method }
24.     exceptobj := Nil;
25.     try
26.       ExecuteThreadQueueEntry(tmpentry);
27.     except
28.       exceptobj := Exception(AcquireExceptionObject);
29.     end;
30.     { step 3: error handling and cleanup }
31.     if Assigned(tmpentry^.SyncEvent) then
32.       begin
33.       { for Synchronize entries we pass back the Exception and trigger
34.         the event that Synchronize waits in }
35.       tmpentry^.Exception := exceptobj;
36.       RtlEventSetEvent(tmpentry^.SyncEvent)
37.       end
38.     else
39.       begin
40.       { for Queue entries we dispose the entry and raise the exception }
41.       Dispose(tmpentry);
42.       if Assigned(exceptobj) then
43.         raise exceptobj;
44.       end;
45.     tmpentry := PopThreadQueueHead;
46.     end;
47. end;
48.  

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

1. Function PopThreadQueueHead : TThread.PThreadQueueEntry;
2.  
3. begin
4.   Result:=ThreadQueueHead;
5.   if (Result<>Nil) then
6.     begin
7.     System.EnterCriticalSection(ThreadQueueLock);
8.     try
9.       Result:=ThreadQueueHead;
10.       if Result<>Nil then
11.         ThreadQueueHead:=ThreadQueueHead^.Next;
12.       if Not Assigned(ThreadQueueHead) then
13.         ThreadQueueTail := Nil;
14.     finally
15.       System.LeaveCriticalSection(ThreadQueueLock);
16.     end;
17.     end;
18. end;  
19.  

[Mr.Madguy]

Also, yeah, single-linked list is used to implement queue. Err. Single-linked list and looped buffer both have their pros and cons. Single-linked list requires memory allocation per item and more memory for reference to next items, but it doesn't require data copying, when growing. Looped buffer has to be pre-allocated, requires indexed operations and data copying, when shrinking/growing.

[Remy Lebeau]

If aThread is nil then CurrentThread is used

The purpose of the AThread parameter is to associate the queued method with a specific TThread object.  For instance, if the queued method needs to access any non-static members of a TThread, it would be a bad thing if the queued method were called after the TThread was destroyed.  When a TThread is destroyed, it removes any queued methods that have been associated with it and have not been called by the main thread yet.

If FreePascal is substituting TThread.CurrentThread when AThread=nil, I would consider that a logic bug.  If TThread.Queue() is called inside of TThread.Execute() (or TThread.DoTerminate()) with AThread=nil, then the queued method will still be associated with the calling TThread, and thus subject to removal when that TThread is destroyed.

This would break compatibility with Delphi.  If AThread=nil, Delphi does not associate the queued method with any particular Thread object, as it should be.  When the calling TThread is destroyed, the queued method can still be called by the main thread.

Is CurrentThreadVar always the main thread?

No.  It is a 'threadvar' that points to whatever the calling thread actually is.  Every TThread has its own copy of CurrentThreadVar, and stores its Self pointer in it.  Thus, if GetCurrentThread() is called inside of TThread.Execute() or TThread.DoTerminate(), it can return that TThread object as-is.  But, if GetCurrentThread() is called in any non-TThread thread (like the main thread, but can be any thread not managed by TThread), There is no TThread object stored in CurrentThreadVar by default, so it returns a non-functional TThread object (so the caller can use things like TThread.ThreadID, at least).

[Remy Lebeau]

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.

Correct.

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!!

It is not actually a bug, just a design limitation.  TThread.Queue() was originally designed to follow the same semantics as TThread.Synchronize() (or even Windows, for that matter).  If the main thread calls TThread.Synchronize() or TThread.Queue() (or SendMessage()), the requested method (or message handler) is executed right away.  Only methods (or messages) requested (sent) by a worker thread are queued for execution in the main thread when it finds time to execute them.

For the record, the ability to have the main thread queue a delay-executed method was recently added to Delphi in 10.2 Tokyo, with the introduction of the TThread.ForceQueue() method.  FreePascal does not have an equivalent yet.  A close approximation would be to have the main thread use TThread.CreateAnonymousThread() to call TThread.Queue() (the same was true in Delphi until ForceQueue() was added).

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.

Yes, it is a global thread-safe list owned by the RTL, not TThread itself.  The main thread calls CheckSynchronize() periodically to execute pending methods that are waiting in the 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

They are not threadvars.  Just simple globals, protected by the critical section.

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

Correct.

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.

It doesn't care which thread is *adding* an item to the list.  It only cares which thread is *associated* with an item.  For instance, a queued method might need to access members of its associated thread object, so associating the method with a thread object ensures that thread object is still valid when the method is called.

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

If the queued method accesses its associated thread object, and the thread object is destroyed before the queued method is called, the method has to be removed from the list to avoid crashing.

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?

See above, and my earlier reply, too.

[taazz]

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.

Correct.

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!!

It is not actually a bug, just a design limitation.  TThread.Queue() was originally designed to follow the same semantics as TThread.Synchronize() (or even Windows, for that matter).  If the main thread calls TThread.Synchronize() or TThread.Queue() (or SendMessage()), the requested method (or message handler) is executed right away.  Only methods (or messages) requested (sent) by a worker thread are queued for execution in the main thread when it finds time to execute them.

Who ever design that needs to go back and work as a junior for a bit.

The synchronization started on the wrong foot by having a method of a class changing an other class instead of it self. if I did not know anything about synchronize and queue I would expect when I call MyThread.Synchronize(aMethod,SomeData) that the method will be executed in the context of mythread not the main thread. Then they introduce queue based on the same backwards principle and violate the logic behind queue for no apparent reason. I get it that for what ever reason the thread has finished execution and needs to terminate, I'll even go as far as assume that most of threads are badly designed with a bunch of method calls queued in the main thread that need a thread instance to function properly, instead of a hard crash to raise awareness of the wrong design you remove the queued methods creating (possible) impossible to truck data corruption in the process. I get it it is not as straight forward to choose between a hard to track crash and a hard to track data corruption but data corruption should never be the default choice, or is it just me that values the data above application stability?

As for the windows and sendMessage it does that to avoid dead locking it self, as far as I know it does not execute a call to postmessage immediately as far as I know postmessage was used for delayed execution on the same thread from its inception. And yes I do expect that queue exhibits the same behaviour (in general) as postmessage (although it is still backwards) and synchronize the same behaviour as sendmessage. You do not stack one exception on top of an other.

For the record, the ability to have the main thread queue a delay-executed method was recently added to Delphi in 10.2 Tokyo, with the introduction of the TThread.ForceQueue() method.  FreePascal does not have an equivalent yet.  A close approximation would be to have the main thread use TThread.CreateAnonymousThread() to call TThread.Queue() (the same was true in Delphi until ForceQueue() was added).

Lazarus already has it, it is called Application.QueueAsyncCall at least that was designed properly as for fpc lets hope they'll redesign it.

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.

Yes, it is a global thread-safe list owned by the RTL, not TThread itself.  The main thread calls CheckSynchronize() periodically to execute pending methods that are waiting in the list.

agreed

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

They are not threadvars.  Just simple globals, protected by the critical section.

agreed.

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

Correct.

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.

It doesn't care which thread is *adding* an item to the list.  It only cares which thread is *associated* with an item.  For instance, a queued method might need to access members of its associated thread object, so associating the method with a thread object ensures that thread object is still valid when the method is called.

That is good only if the user requests it. I can see the wisdom of adding self as the associated thread on the queue overload with no thread parameter but I see no wisdom on overriding the users nil with what ever they think is appropriate, that is inappropriate behavior. You never override the end user's parameters for any reason.

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

If the queued method accesses its associated thread object, and the thread object is destroyed before the queued method is called, the method has to be removed from the list to avoid crashing.

I disagree, it is better to crash and raise awareness, instead of having seemingly random corruptions with no indication that the application has a design flaw.

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?

See above, and my earlier reply, too.

Thank you for your time, it looks like that queue is badly designed and can't be trusted, never the less your insight was interesting and invaluable.

[taazz]

CheckSynchronize is called from Application.ProcessMessages.

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

1. procedure TWin32WidgetSet.AppProcessMessages;
2. var
3.   AMessage: TMsg;
4.   retVal, index: dword;
5.   pHandles: Windows.LPHANDLE;
6.  
7.     procedure CallWaitHandler;
8.     begin
9.       FWaitHandlers[index].OnEvent(FWaitHandlers[index].UserData, 0);
10.     end;
11.  
12. begin
13.   repeat
14.     if FPendingWaitHandlerIndex >= 0 then
15.     begin
16.       index := FPendingWaitHandlerIndex;
17.       FPendingWaitHandlerIndex := -1;
18.       CallWaitHandler;
19.     end;
20. {$ifdef DEBUG_ASYNCEVENTS}
21.     if Length(FWaitHandles) > 0 then
22.       DebugLn('[ProcessMessages] WaitHandleCount=', IntToStr(FWaitHandleCount),
23.         ', WaitHandle[0]=', IntToHex(FWaitHandles[0],);
24. {$endif}
25.     if FWaitHandleCount > 0 then
26.       pHandles := @FWaitHandles[0]
27.     else
28.       pHandles := nil;
29.     retVal := Windows.MsgWaitForMultipleObjects(FWaitHandleCount,
30.       pHandles, False, 0, QS_ALLINPUT);
31.     if (retVal < WAIT_OBJECT_0 + FWaitHandleCount) then
32.     begin
33.       index := retVal-WAIT_OBJECT_0;
34.       CallWaitHandler;
35.     end else
36.     if retVal = WAIT_OBJECT_0 + FWaitHandleCount then
37.     begin
38.       while PeekMessage(AMessage, HWnd(nil), 0, 0, PM_REMOVE) do
39.       begin
40.         if AMessage.message = WM_QUIT then
41.         begin
42.           PostQuitMessage(AMessage.wParam);
43.           break;
44.         end;
45.         TranslateMessage(@AMessage);
46.         if UnicodeEnabledOS then
47.           DispatchMessageW(@AMessage)
48.         else
49.           DispatchMessage(@AMessage);
50.       end;
51.     end else
52.     if retVal = WAIT_TIMEOUT then
53.     begin
54.       // check for pending to-be synchronized methods
55.       CheckSynchronize;
56.       CheckPipeEvents;
57.       break;
58.     end else
59.     if retVal = $FFFFFFFF then
60.     begin
61.       DebugLn('[TWin32WidgetSet.AppProcessMessages] MsgWaitForMultipleObjects returned: ', IntToStr(GetLastError));
62.       break;
63.     end;
64.   until false;
65. end;          
66.  

They made it widget set bound? that is an other head scratcher.

Thank you for digging that up, I wasn't looking forward to dig in the widget set code and all that inc file nonsense.

[Remy Lebeau]

Lazarus already has it, it is called Application.QueueAsyncCall at least that was designed properly as for fpc lets hope they'll redesign it.

Thanks for that. Application.QueueAsyncCall() does sounds useful.  Unfortunately, it is in the Forms unit, so I can't use it in Indy (no Forms unit allowed! Except in the TIdAntiFreeze component and design-time editors).  Is there another option available in FreePascal's RTL, except for spawning a thread to call TThread.Queue()?

It doesn't care which thread is *adding* an item to the list.  It only cares which thread is *associated* with an item.  For instance, a queued method might need to access members of its associated thread object, so associating the method with a thread object ensures that thread object is still valid when the method is called.

That is good only if the user requests it. I can see the wisdom of adding self as the associated thread on the queue overload with no thread parameter but I see no wisdom on overriding the users nil with what ever they think is appropriate, that is inappropriate behavior. You never override the end user's parameters for any reason.

Agreed.

[taazz]

Lazarus already has it, it is called Application.QueueAsyncCall at least that was designed properly as for fpc lets hope they'll redesign it.

Thanks for that. Application.QueueAsyncCall() does sounds useful.  Unfortunately, it is in the Forms unit, so I can't use it in Indy (no Forms unit allowed! Except in the TIdAntiFreeze component and design-time editors).  Is there another option available in FreePascal's RTL, except for spawning a thread to call TThread.Queue()?

Not that I know of, no. In an end user application I would suggest to write your own queue mechanism and add an application onidle handler to run a "checksynchronize" type of call but it requires access to the forms unit.
You could install the handler from the TIdAntiFreeze unit on initialization though, the good thing is, that lazarus provides event lists so your handler will not interfere with the end users code.

[molly]

Thank you guys for the additional information, and taaz for digging into the implementation and setting things straight. tbh, i never bothered to dig into it that deep.

[cpicanco]

No.  It is a 'threadvar' that points to whatever the calling thread actually is.

So, I conclude that "CurrentCallingThreadVar" would be a more readable name. I will need some time to digest this information. Hopefully this discussion will end up in good examples in the wiki.

[PascalDragon]

If aThread is nil then CurrentThread is used

The purpose of the AThread parameter is to associate the queued method with a specific TThread object.  For instance, if the queued method needs to access any non-static members of a TThread, it would be a bad thing if the queued method were called after the TThread was destroyed.  When a TThread is destroyed, it removes any queued methods that have been associated with it and have not been called by the main thread yet.

If FreePascal is substituting TThread.CurrentThread when AThread=nil, I would consider that a logic bug.  If TThread.Queue() is called inside of TThread.Execute() (or TThread.DoTerminate()) with AThread=nil, then the queued method will still be associated with the calling TThread, and thus subject to removal when that TThread is destroyed.

This would break compatibility with Delphi.  If AThread=nil, Delphi does not associate the queued method with any particular Thread object, as it should be.  When the calling TThread is destroyed, the queued method can still be called by the main thread.

That was already fixed in Trunk in revision 33863. Sadly I forgot to get that revision merged into 3.0.x... oh well... 

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!!

It is not actually a bug, just a design limitation.  TThread.Queue() was originally designed to follow the same semantics as TThread.Synchronize() (or even Windows, for that matter).  If the main thread calls TThread.Synchronize() or TThread.Queue() (or SendMessage()), the requested method (or message handler) is executed right away.  Only methods (or messages) requested (sent) by a worker thread are queued for execution in the main thread when it finds time to execute them.

For the record, the ability to have the main thread queue a delay-executed method was recently added to Delphi in 10.2 Tokyo, with the introduction of the TThread.ForceQueue() method.  FreePascal does not have an equivalent yet.  A close approximation would be to have the main thread use TThread.CreateAnonymousThread() to call TThread.Queue() (the same was true in Delphi until ForceQueue() was added).

Trunk revision 37359 just added TThread.ForceQueue() as well

[Remy Lebeau]

Trunk revision 37359 just added TThread.ForceQueue() as well

Thanks for that.  I have updated Indy to use it.

[snorkel]

Why don't you just use Postmessage to send data to the main form/thread, works way better than synchronize
All the different OSes (Linux,Windows etc) now support Postmessage.

You could even combine a queue with using postmessage.