Changeset 148 for MicroThreading/UMicroThreading.pas

Timestamp:

Jan 26, 2011, 2:16:19 PM (14 years ago)

Author:

george

Message:

• Modified: Not completed thread safe support.

File:

• MicroThreading/UMicroThreading.pas (modified) (20 diffs)

MicroThreading/UMicroThreading.pas

@@ -105 +105 @@
     FTerminated: Boolean;
     FTempPointer: Pointer;
+    FCurrentMicroThread: TMicroThread;
+    FScheduler: TMicroThreadScheduler;
     function Execute(Count: Integer): Integer;
   public
-    Scheduler: TMicroThreadScheduler;
-    CurrentMicroThread: TMicroThread;
     procedure Yield;
     constructor Create;
     destructor Destroy; override;
-  end;
+    property Scheduler: TMicroThreadScheduler read FScheduler;
+    property CurrentMicroThread: TMicroThread read FCurrentMicroThread;
+  end;
+
+  TMicroThreadSchedulerState = (ssStopped, ssRunning, ssTerminating);
 
   { TMicroThreadScheduler }
@@ -118 +122 @@
   TMicroThreadScheduler = class
   private
-    ThreadPool: TThreadPool;
-    RoundRobinIndex: Integer;
+    FActive: Boolean;
+    FThreadPool: TThreadPool;
+    FThreadPoolLock: TCriticalSection;
+    FThreadPoolSize: Integer;
+    FRoundRobinIndex: Integer;
     FLastId: Integer;
     FFrequency: Int64;
+    FTerminate: Boolean;
     FTerminated: Boolean;
+    FMicroThreads: TObjectList; // TList<TMicroThread>
+    FMainThreadManager: TMicroThreadManager;
+    FMicroThreadsLock: TCriticalSection;
+    FState: TMicroThreadSchedulerState;
     function GetMicroThreadCount: Integer;
     function GetThreadPoolSize: Integer;
+    procedure SetActive(const AValue: Boolean);
     procedure SetThreadPoolSize(const AValue: Integer);
     function GetNextMicroThread: TMicroThread;
+    procedure WaitFor;
+    procedure Start;
+    procedure Stop;
+    function ThreadPoolTerminated: Boolean;
   public
-    MainThreadManager: TMicroThreadManager;
-    MicroThreads: TObjectList; // TList<TMicroThread>
-    Lock: TCriticalSection;
     function GetNow: TDateTime;
     function Add(MicroThread: TMicroThread): Integer;
     function AddMethod(Method: TMicroThreadEvent): Integer;
+    function GetCPUCoreCount: Integer;
     constructor Create;
     destructor Destroy; override;
-    procedure Start;
-    procedure Stop;
     property MicroThreadCount: Integer read GetMicroThreadCount;
     property ThreadPoolSize: Integer read GetThreadPoolSize
       write SetThreadPoolSize;
+    property MicroThreads: TObjectList read FMicroThreads;
+    property MicroThreadsLock: TCriticalSection read FMicroThreadsLock;
+    property MainThreadManager: TMicroThreadManager read FMainThreadManager;
+    property Active: Boolean read FActive write SetActive;
   end;
 
@@ -167 +184 @@
   with MainScheduler do begin
     try
-      Lock.Acquire;
+      FMicroThreadsLock.Acquire;
       I := 0;
-      while (I < MicroThreads.Count) and
-        not ((CurrentStack >= TMicroThread(MicroThreads[I]).FStack) and
-        (CurrentStack <= (TMicroThread(MicroThreads[I]).FStack +
-        TMicroThread(MicroThreads[I]).FStackSize))) do Inc(I);
-      if I < MicroThreads.Count then begin
-        Result := TMicroThread(MicroThreads[I]).FId;
+      while (I < FMicroThreads.Count) and
+        not ((CurrentStack >= TMicroThread(FMicroThreads[I]).FStack) and
+        (CurrentStack <= (TMicroThread(FMicroThreads[I]).FStack +
+        TMicroThread(FMicroThreads[I]).FStackSize))) do Inc(I);
+      if I < FMicroThreads.Count then begin
+        Result := TMicroThread(FMicroThreads[I]).FId;
       end else Result := -1;
     finally
-      Lock.Release;
+      FMicroThreadsLock.Release;
     end;
   end;
@@ -199 +216 @@
   CurrentTime: TDateTime;
 begin
-  CurrentTime := Scheduler.GetNow;
-  if Assigned(CurrentMicroThread) then begin
-    CurrentMicroThread.FExecutionEndTime := CurrentTime;
-    CurrentMicroThread.FExecutionTime := CurrentMicroThread.FExecutionTime +
-      (CurrentMicroThread.FExecutionEndTime - CurrentMicroThread.FExecutionStartTime);
-    if CurrentMicroThread.FState = tsRunning then
-      CurrentMicroThread.FState := tsWaiting;
-    StaticMicroThread := CurrentMicroThread;
+  CurrentTime := FScheduler.GetNow;
+  if Assigned(FCurrentMicroThread) then begin
+    FCurrentMicroThread.FExecutionEndTime := CurrentTime;
+    FCurrentMicroThread.FExecutionTime := FCurrentMicroThread.FExecutionTime +
+      (FCurrentMicroThread.FExecutionEndTime - FCurrentMicroThread.FExecutionStartTime);
+    if FCurrentMicroThread.FState = tsRunning then
+      FCurrentMicroThread.FState := tsWaiting;
+    StaticMicroThread := FCurrentMicroThread;
     asm
       // Store microthread stack
@@ -215 +232 @@
       mov [eax].TMicroThread.FBasePointer, edx
     end;
-    StaticManager := CurrentMicroThread.FManager;
+    StaticManager := FCurrentMicroThread.FManager;
     asm
-      // Restore scheduler stack
+      // Restore FScheduler stack
       mov eax, StaticManager  // Self is invalid before BP restore
       mov edx, [eax].TMicroThreadManager.FStackPointer
@@ -224 +241 @@
       mov ebp, edx
     end;
-    CurrentMicroThread.FManager := nil;
-    CurrentMicroThread := nil;
-  end;
-
-  CurrentMicroThread := Scheduler.GetNextMicroThread;
-
-  if Assigned(CurrentMicroThread) and (FExecutedCount < FExecuteCount) then begin
-    CurrentMicroThread.FManager := Self;
+    FCurrentMicroThread.FManager := nil;
+    FCurrentMicroThread := nil;
+  end;
+
+  FCurrentMicroThread := FScheduler.GetNextMicroThread;
+
+  if Assigned(FCurrentMicroThread) and (FExecutedCount < FExecuteCount) then begin
+    FCurrentMicroThread.FManager := Self;
     Inc(FExecutedCount);
     asm
-      // Store scheduler stack
+      // Store FScheduler stack
       mov eax, Self
       mov edx, esp
@@ -241 +258 @@
       mov [eax].TMicroThreadManager.FBasePointer, edx
     end;
-    if not CurrentMicroThread.FExecuted then begin
-      CurrentMicroThread.FExecuted := True;
-      CurrentMicroThread.FState := tsRunning;
-      CurrentMicroThread.FExecutionStartTime := CurrentTime;
-      StaticMicroThread := CurrentMicroThread;
+    if not FCurrentMicroThread.FExecuted then begin
+      FCurrentMicroThread.FExecuted := True;
+      FCurrentMicroThread.FState := tsRunning;
+      FCurrentMicroThread.FExecutionStartTime := CurrentTime;
+      StaticMicroThread := FCurrentMicroThread;
       asm
         // Restore microthread stack
@@ -260 +277 @@
       end;
       //FSelected.Method(FSelected);
-      StaticManager := CurrentMicroThread.FManager;
+      StaticManager := FCurrentMicroThread.FManager;
       asm
-        // Restore scheduler stack
+        // Restore FScheduler stack
         mov eax, StaticManager // Self is invalid before BP restore
         mov edx, [eax].TMicroThreadManager.FStackPointer
@@ -269 +286 @@
         mov ebp, edx
       end;
-      CurrentMicroThread.FManager := nil;
-      CurrentMicroThread.FExecutionEndTime := CurrentTime;
-      CurrentMicroThread.FExecutionTime := CurrentMicroThread.FExecutionTime +
-       (CurrentMicroThread.FExecutionEndTime - CurrentMicroThread.FExecutionStartTime);
-      CurrentMicroThread.FFinished := True;
-      if CurrentMicroThread.FFreeOnTerminate then begin
+      FCurrentMicroThread.FManager := nil;
+      FCurrentMicroThread.FExecutionEndTime := CurrentTime;
+      FCurrentMicroThread.FExecutionTime := FCurrentMicroThread.FExecutionTime +
+       (FCurrentMicroThread.FExecutionEndTime - FCurrentMicroThread.FExecutionStartTime);
+      FCurrentMicroThread.FFinished := True;
+      if FCurrentMicroThread.FFreeOnTerminate then begin
         // Microthread is finished, remove it from queue
-        with Scheduler do
+        with FScheduler do
         try
-          Lock.Acquire;
-          MicroThreads.Delete(MicroThreads.IndexOf(CurrentMicroThread));
+          FMicroThreadsLock.Acquire;
+          FMicroThreads.Delete(FMicroThreads.IndexOf(FCurrentMicroThread));
         finally
-          Lock.Release;
+          FMicroThreadsLock.Release;
         end;
       end;
-      CurrentMicroThread := nil;
+      FCurrentMicroThread := nil;
     end else
-    if CurrentMicroThread.State = tsWaiting then begin
+    if FCurrentMicroThread.State = tsWaiting then begin
       // Execute selected thread
-      CurrentMicroThread.FState := tsRunning;
-      CurrentMicroThread.FExecutionStartTime := CurrentTime;
-      FTempPointer := CurrentMicroThread.FStackPointer;
+      FCurrentMicroThread.FState := tsRunning;
+      FCurrentMicroThread.FExecutionStartTime := CurrentTime;
+      FTempPointer := FCurrentMicroThread.FStackPointer;
       asm
         // Restore microthread stack
@@ -297 +314 @@
         mov esp, edx
       end;
-      FTempPointer := CurrentMicroThread.FBasePointer;
+      FTempPointer := FCurrentMicroThread.FBasePointer;
       asm
         mov eax, Self
@@ -305 +322 @@
     end;
   end else begin
-    CurrentMicroThread := nil;
+    FCurrentMicroThread := nil;
   end;
 end;
@@ -311 +328 @@
 constructor TMicroThreadManager.Create;
 begin
-  CurrentMicroThread := nil;
+  FCurrentMicroThread := nil;
 end;
 
@@ -469 +486 @@
   MicroThread.FScheduler := Self;
   MicroThread.FId := FLastId;
-  Result := MicroThreads.Add(MicroThread);
+  Result := FMicroThreads.Add(MicroThread);
 end;
 
@@ -482 +499 @@
 end;
 
+function TMicroThreadScheduler.GetCPUCoreCount: Integer;
+var
+  SystemInfo: _SYSTEM_INFO;
+begin
+  GetSystemInfo(SystemInfo);
+  Result := SystemInfo.dwNumberOfProcessors;
+end;
+
 constructor TMicroThreadScheduler.Create;
 begin
-  Lock := TCriticalSection.Create;
-  MicroThreads := TObjectList.Create;
-  ThreadPool := TThreadPool.Create;
+  FTerminated := True;
+  FMicroThreadsLock := TCriticalSection.Create;
+  FMicroThreads := TObjectList.Create;
+  FThreadPool := TThreadPool.Create;
+  FThreadPoolLock := TCriticalSection.Create;
   {$IFDEF Windows}
   QueryPerformanceFrequency(FFrequency);
   {$ENDIF}
-  RoundRobinIndex := -1;
-  MainThreadManager := TMicroThreadManager.Create;
-  MainThreadManager.Scheduler := Self;
+  FRoundRobinIndex := -1;
+  FMainThreadManager := TMicroThreadManager.Create;
+  FMainThreadManager.FScheduler := Self;
 end;
 
 destructor TMicroThreadScheduler.Destroy;
 begin
-  MainThreadManager.Free;
-  FTerminated := True;
-  ThreadPool.Free;
-  MicroThreads.Free;
-  Lock.Free;
+  Active := False;
+  FMainThreadManager.Free;
+  FThreadPool.Free;
+  FMicroThreads.Free;
+  FMicroThreadsLock.Free;
   inherited Destroy;
 end;
@@ -511 +538 @@
 begin
   FTerminated := False;
-  for I := 0 to ThreadPool.Count - 1 do
-    TMicroThreadSchedulerPoolThread(ThreadPool[I]).Start;
+  FTerminate := False;
+  for I := 0 to FThreadPool.Count - 1 do
+    TMicroThreadSchedulerPoolThread(FThreadPool[I]).Start;
   repeat
-    Executed := MainThreadManager.Execute(10);
+    Executed := FMainThreadManager.Execute(10);
     Application.ProcessMessages;
     if Executed = 0 then Sleep(1);
-  until FTerminated;
+  until FTerminate;
+  FTerminated := True;
 end;
 
@@ -524 +553 @@
   I: Integer;
 begin
-  for I := 0 to ThreadPool.Count - 1 do
-    TMicroThreadSchedulerPoolThread(ThreadPool[I]).Terminate;
-  FTerminated := True;
+  try
+    FThreadPoolLock.Acquire;
+    for I := 0 to FThreadPool.Count - 1 do begin
+      TMicroThreadSchedulerPoolThread(FThreadPool[I]).Terminate;
+    end;
+  finally
+    FThreadPoolLock.Release;
+  end;
+  FTerminate := True;
+
+  // Wait for all thread managers to finish
+  repeat
+    Application.ProcessMessages;
+    Sleep(1);
+  until FTerminated and (ThreadPoolSize = 0);
+end;
+
+function TMicroThreadScheduler.ThreadPoolTerminated: Boolean;
+var
+  I: Integer;
+begin
+  try
+    FThreadPoolLock.Acquire;
+    I := 0;
+    while (I < FThreadPool.Count) and
+      (TMicroThreadSchedulerPoolThread(FThreadPool[I]).Terminated do
+  finally
+    FThreadPoolLock.Release;
+  end;
 end;
 
@@ -537 +592 @@
   Result := nil;
   try
-    Lock.Acquire;
+    FMicroThreadsLock.Acquire;
     I := 0;
-    Inc(RoundRobinIndex);
-    if RoundRobinIndex >= MicroThreads.Count then
-      RoundRobinIndex := 0;
-    while (I < MicroThreads.Count) and
-     (TMicroThread(MicroThreads[RoundRobinIndex]).State <> tsWaiting) do begin
+    Inc(FRoundRobinIndex);
+    if FRoundRobinIndex >= FMicroThreads.Count then
+      FRoundRobinIndex := 0;
+    while (I < FMicroThreads.Count) and
+     (TMicroThread(FMicroThreads[FRoundRobinIndex]).State <> tsWaiting) do begin
       // WakeUp sleeping threads
-      if (TMicroThread(MicroThreads[RoundRobinIndex]).FState = tsSleeping) and
-        (TMicroThread(MicroThreads[RoundRobinIndex]).FWakeupTime < CurrentTime) then
-          TMicroThread(MicroThreads[RoundRobinIndex]).FState := tsWaiting else
+      if (TMicroThread(FMicroThreads[FRoundRobinIndex]).FState = tsSleeping) and
+        (TMicroThread(FMicroThreads[FRoundRobinIndex]).FWakeupTime < CurrentTime) then
+          TMicroThread(FMicroThreads[FRoundRobinIndex]).FState := tsWaiting else
       begin
         // Go to next thread
         Inc(I);
-        Inc(RoundRobinIndex);
-        if RoundRobinIndex >= MicroThreads.Count then
-          RoundRobinIndex := 0;
+        Inc(FRoundRobinIndex);
+        if FRoundRobinIndex >= FMicroThreads.Count then
+          FRoundRobinIndex := 0;
       end;
     end;
-    if I < MicroThreads.Count then begin
-      Result := TMicroThread(MicroThreads[RoundRobinIndex]);
+    if I < FMicroThreads.Count then begin
+      Result := TMicroThread(FMicroThreads[FRoundRobinIndex]);
     end;
   finally
-    Lock.Release;
+    FMicroThreadsLock.Release;
   end;
 end;
@@ -567 +622 @@
 begin
   try
-    Lock.Acquire;
-    Result := MicroThreads.Count;
+    FMicroThreadsLock.Acquire;
+    Result := FMicroThreads.Count;
   finally
-    Lock.Release;
+    FMicroThreadsLock.Release;
   end;
 end;
@@ -576 +631 @@
 function TMicroThreadScheduler.GetThreadPoolSize: Integer;
 begin
-  Result := ThreadPool.Count;
+  Result := FThreadPoolSize;
+end;
+
+procedure TMicroThreadScheduler.SetActive(const AValue: Boolean);
+begin
+  if FActive = AValue then Exit;
+  FActive := AValue;
+  if AValue then Start
+    else Stop;
 end;
 
@@ -584 +647 @@
   NewThread: TMicroThreadSchedulerPoolThread;
 begin
-  if AValue > ThreadPool.Count then begin
-    ThreadPool.Capacity := AValue;
-    while ThreadPool.Count < AValue do begin
-      NewThread := TMicroThreadSchedulerPoolThread.Create(True);
-      NewThread.Manager.Scheduler := Self;
-      ThreadPool.Add(NewThread);
-    end;
-  end else
-  ThreadPool.Count := AValue;
+  FThreadPoolSize := AValue;
+  if FState = ssRunning then
+    SetThreadPoolCount
 end;