Made burst and blocking thresholds configurable, fixed deadlock issue.

Author: vincentparrett

VSoft.Messaging.dspec

@@ -1,7 +1,7 @@
 {
   "metadata": {
     "id": "VSoft.Messaging",
-    "version": "0.1.0",
+    "version": "0.2.0",
     "description": "VSoft.Messaging is a libary that provides an internal synchronous/asynchronous publish/subscribe messaging system for Delphi applications.",
     "authors": "Vincent Parrett",
     "projectUrl": "https://github.com/VSoftTechnologies/VSoft.SemanticVersion",
@@ -104,6 +104,13 @@
           "id": "Runtime",
           "project": ".\\packages\\Rad Studio $compilerWithCodeName$\\VSoft.MessagingR.dproj"
         }
+      ],
+      "runtime" : [
+       {
+         "buildId" : "Runtime",
+         "src" : "bin\\VSoft.MessagingR$LibSuffix$.bpl",
+         "copyLocal" : true
+       }
       ]
     }
   ]

demos/Vcl/MsgDemo.dproj

@@ -986,8 +986,8 @@
         </BorlandProject>
         <ProjectFileVersion>12</ProjectFileVersion>
         <DPM>
-            <PackageReference id="VSoft.WeakReference" platform="Win32" version="0.0.1"/>
-            <PackageReference id="VSoft.WeakReference" platform="Win64" version="0.0.1"/>
+            <PackageReference id="VSoft.WeakReference" platform="Win64" version="0.1.0"/>
+            <PackageReference id="VSoft.WeakReference" platform="Win32" version="0.1.0"/>
         </DPM>
     </ProjectExtensions>
     <Import Project="$(BDS)\Bin\CodeGear.Delphi.Targets" Condition="Exists('$(BDS)\Bin\CodeGear.Delphi.Targets')"/>

src/VSoft.Messaging.Dispatchers.pas

@@ -202,7 +202,7 @@ procedure TMessageDispatcherThread.Execute;
     end;
     while (not Self.Terminated) and (FDispatcher.FTarget <> nil) and (FDispatcher.FQueue.Count > 0) and FDispatcher.Enabled do
     begin
-      msgs := FDispatcher.DequeueAtMost(cMaxBurst);
+      msgs := FDispatcher.DequeueAtMost(TMessagingOptions.MaxBurst);
       for i := 0 to Length(msgs) -1 do
       begin
         try
@@ -427,12 +427,11 @@ procedure TMessageDispatcherUIThread.Execute;
                   i : integer;
                   msgs : TArray<IMessage>;
                 begin
-                  msgs := FDispatcher.DequeueAtMost(cMaxBurst);
+                  msgs := FDispatcher.DequeueAtMost(TMessagingOptions.MaxBurst);
                   if Length(msgs) > 0 then
                   begin
                     for i := 0 to Length(msgs) -1 do
                     begin
-
                       if not FDispatcher.Enabled or Self.Terminated then
                         break;
                       try
@@ -441,8 +440,12 @@ procedure TMessageDispatcherUIThread.Execute;
                         //not much we can do here!
                       end;
                       msgs[i] := nil;
-                      //potential tight loop.
-                      TThread.Yield;
+                      //potential tight loop, yield occasionally
+                      if TMessagingOptions.MaxBurst > 10 then
+                      begin
+                        if (i mod 10) = 0 then
+                          TThread.Yield;
+                      end;
                     end;
                   end;
                 end;
@@ -458,8 +461,10 @@ procedure TMessageDispatcherUIThread.Execute;
     end;
     while (not Self.Terminated) and (FDispatcher.FTarget <> nil) and (FDispatcher.FQueue.Count > 0) and FDispatcher.Enabled do
     begin
-      TThread.Queue(nil,threadProc);
-      //potential tight loop, so allow other threads to run.
+      //since we are queuing messages and want them to complete processing before the next one is dispatched, we cannot use TThread.Queue here.
+      TThread.Synchronize(nil,threadProc);
+      //calls SwitchToThread - Causes the calling thread to yield execution to another thread that is ready to run on the current processor.
+      //The operating system selects the next thread to be executed.
       TThread.Yield;
     end;
   end;

src/VSoft.Messaging.Internal.pas

@@ -30,17 +30,12 @@ interface
 uses
   System.SyncObjs;
 
-const
-  cMaxBurst           = 20;  // Maxium number of message to dequeue and process in one go.
-  cBlockedThreshold   = 100; // Start waiting for queues to unblock at this global queue depth
-  cUnblockedThreshold = 50;  // Stop waiting at this queue depth  (must be at least one less than block threshold)
-
 type
   //used to control pushback etc
   TMessagingControl = class
   private
     class var
-      FGlobalQueueDepth : Integer;
+      FGlobalQueueDepth : integer;
       FUnblockedSignal : TEvent;
   public
     class constructor Create;
@@ -78,7 +73,8 @@   TVSMessageWrapper<T> = class(TInterfacedObject,IMessage)
 implementation
 
 uses
-  System.Classes;
+  System.Classes,
+  VSoft.Messaging;
 
 { TMessagingControl }
 
@@ -89,7 +85,7 @@ implementation
 
 class procedure TMessagingControl.DecrementGlobalQueueDepth;
 begin
-  if TInterlocked.Decrement(FGlobalQueueDepth) = cUnblockedThreshold then
+  if TInterlocked.Decrement(FGlobalQueueDepth) = TMessagingOptions.UnblockThreshold then
     FUnblockedSignal.SetEvent; // Unblock "PushbackIfNeeded"
 end;
 
@@ -105,15 +101,15 @@ class function TMessagingControl.GetGlobalQueueDepth: integer;
 
 class procedure TMessagingControl.IncrementGlobalQueueDepth;
 begin
-  if TInterlocked.Increment(FGlobalQueueDepth) = cBlockedThreshold then
+  if TInterlocked.Increment(FGlobalQueueDepth) = TMessagingOptions.BlockedThreshold then
     FUnblockedSignal.ResetEvent; // Block "PushbackIfNeeded"
 end;
 
 class procedure TMessagingControl.PushbackIfNeeded;
 var
   res : TWaitResult;
 begin
-  if FGlobalQueueDepth < cBlockedThreshold then
+  if FGlobalQueueDepth < TMessagingOptions.BlockedThreshold then
     exit;
 
   if MainThreadID = TThread.CurrentThread.ThreadID then

src/VSoft.Messaging.pas

@@ -44,11 +44,10 @@   TChannelHelper = record
     FPostMessageProc : TMessageProc;
     FSendMessageProc : TMessageProc;
   public
-    //10.4 update 2 broke record constraints, commenting out for now.
     //async
-    procedure PostMessage<T {: record}>(const message : T);
+    procedure PostMessage<T : record>(const message : T);
     //sync
-    procedure SendMessage<T {: record}>(const message : T);
+    procedure SendMessage<T : record>(const message : T);
     constructor Create(const postMessageProc : TMessageProc; const sendMessageProc : TMessageProc);
   end;
 
@@ -140,13 +139,35 @@   TMessageDispatcherFactory = class
     class function CreateUIDispatcher(const target : TObject = nil) : IMessageDispatcher;
   end;
 
+  TMessagingOptions = class
+  private
+    class var
+      FMaxBurst           : integer;  // Maxium number of message to dequeue and process in one go.
+      FBlockedThreshold   : integer; // Start waiting for queues to unblock at this global queue depth
+      FUnblockedThreshold : integer;
+  protected
+    class procedure SetMaxBurst(const Value: integer); static;  
+    class procedure SetBlockedThreshold(const Value: integer); static;
+    class procedure SetUnblockThreashold(const Value: integer); static;
+    class constructor Create;
+  public
+    class property MaxBurst : integer read FMaxBurst write SetMaxBurst;  // Maxium number of message to dequeue and process in one go.
+    class property BlockedThreshold : integer read FBlockedThreshold write SetBlockedThreshold; // Start waiting for queues to unblock at this global queue depth
+    class property UnblockThreshold : integer read FUnblockedThreshold write SetUnblockThreashold; //Stop waiting at this queue depth  (must be at least one less than block threshold)
+  end;
+
 implementation
 
 uses
   System.Classes,
   VSoft.Messaging.Channel,
   VSoft.Messaging.Dispatchers;
 
+const
+  cDefaultMaxBurst           = 20;  // Maxium number of message to dequeue and process in one go.
+  cDefaultBlockedThreshold   = 100; // Start waiting for queues to unblock at this global queue depth
+  cDefaultUnblockedThreshold = 50;  // Stop waiting at this queue depth  (must be at least one less than block threshold)
+  
 
 
 { TChannelHelper }
@@ -194,4 +215,37 @@ class function TMessageDispatcherFactory.CreateUIDispatcher(const target : TObje
   result := TUIMessageDispatcher.Create(target);
 end;
 
+{ TMessagingOptions }
+
+class constructor TMessagingOptions.Create;
+begin
+  FMaxBurst := cDefaultMaxBurst;
+  FBlockedThreshold := cDefaultBlockedThreshold;
+  FUnblockedThreshold := cDefaultUnblockedThreshold;
+end;
+
+class procedure TMessagingOptions.SetBlockedThreshold(const Value: integer);
+begin
+  if value > 1  then
+    FBlockedThreshold := Value
+  else
+    FBlockedThreshold := 2;
+end;
+
+class procedure TMessagingOptions.SetMaxBurst(const Value: integer);
+begin
+  if value > 0 then
+    FMaxBurst := Value
+  else
+    FMaxBurst := 1; 
+end;
+
+class procedure TMessagingOptions.SetUnblockThreashold(const Value: integer);
+begin
+  if value < FBlockedThreshold -1 then
+    FUnblockedThreshold := Value
+  else
+    FUnblockedThreshold := FBlockedThreshold -1; 
+end;
+
 end.