TAdvThread mit "soft" Suspend und "nonblocking" WaitFor - Bitte mal drüberschauen

[RSE]

Hallo,

ich habe mich in den letzten Tagen mit Threads beschäftigt. Davor hatte ich nicht viel Ahnung davon. Wer sich damit beschäftigen möchte, dem empfehle ich http://www.eonclash.com/Tutorials/Mu...ey1.1/ToC.html (Englisch, sehr gut erklärt, für Einsteiger bis Profi Interessantes dabei!).

Aus meinen Erkenntnissen der letzten Tage habe ich folgende Klasse TAdvThread erstellt, die ein paar bekannte Probleme abfängt (im Startkommentar erklärt). Da der Quellcode an sich Inhalt des Threads ist und nicht nur ein Codeschnipsel zum Problem, habe ich mir unten erlaubt den Code direkt einzufügen, obwohl er etwas länger ist.

Sinn dieses Postings soll sein, dass sich viele (vor allem erfahrenere) Leute diesen Code ansehen, um eventuelle Fehler zu finden. Als Gegenleistung darf der Quelltext von jedem verwendet werden. Es darf auch gern über Sinn und Unsinn der einzelnen umgesetzten Punkte diskutiert werden. Ich freue mich über viele Antworten!

Anmerkung: Ich arbeite noch mit Delphi 5 - ein Umstieg auf XE ist geplant, aber in unserem Fall nicht auf die Schnelle umsetzbar. Kommentare zu Änderungen in späteren Versionen (> Delphi 5) werde ich also gern zur Kenntnis nehmen, im Kern ist mir aber erst einmal der Einsatz in Delphi 5 wichtig.

Edit: Ich habe noch ein winziges Testprojekt zur Demonstration angehangen.

Delphi-Quellcode:

unit ETC_AdvancedThread;

{
 -----------------------  Was bedeutet Threadsicherheit?  ----------------------

 Lesezugriffe aus unterschiedlichen Threadkontexten auf einen Speicherbereich
 sind unschädlich. Soll der Speicherbereich aber geschrieben werden, muss
 sichergestellt werden können, dass kein anderer Threadkontext gleichzeitig
 daraus liest. Daher ist (außer bei Konstanten) auch für Lesezugriffe eine
 Synchronisation notwendig.

 Um threadsicher zu arbeiten, müssen also alle Initialisierungen der
 Threadklasse erfolgen, bevor der Thread startet. Während der Thread läuft, muss
 klar sein, welche Variablen dem Threadkontext angehören. Innerhalb des
 Threadkontexts darf dann nur auf diese Variablen ohne spezielle
 Synchronisierung zugegriffen werden, außerhalb des Threadkontexts nur auf alle
 anderen.

 Eine Ausnahme bilden TThread.FTerminated und TThread.FFinished. Beide Variablen
 werden bei der Erstellung des TThread-Objekts mit Null=False initialisiert. Sie
 sind beide als private deklariert, auf sie kann also nur über TThread.Terminate
 und TThread.Terminated sowie innerhalb TThread zugegriffen werden. FTerminated
 wird nur von Terminate geschrieben (auf True gesetzt). Der Zustand von
 FTerminated wechselt also während die Instanz von TThread existiert genau 1x
 von False nach True. Da Terminated regelmäßig aufgerufen wird, macht es nichts,
 wenn der Schreibvorgang verpasst wird, beim nächsten Test wird der neue Wert
 gelesen.
 Der Nachteil ist, dass es auf Multicore-Prozessoren mit getrennten Caches etwas
 länger dauern könnte, bis der Wert beim Thread (der u.U. auf dem anderen Core
 läuft) ankommt. Vergleichbar verhält es sich mit FFinished: Es wird lediglich
 1x nach Beendigung der Methode Execute auf True gesetzt. Gelesen wird es nur im
 Destruktor. Die beiden Methoden Terminate und WaitFor verursachen keine
 kritischen Aktionen, falls FFinished zwischenzeitlich gesetzt worden sein
 sollte.


 ----------------------------  Threads in Delphi 5  ----------------------------

 Beim Aufruf von TThread.Suspend wird der Thread sofort unterbrochen. Dabei kann
 er sich unter Umständen auch gerade in einem kritischen Abschnitt befinden -
 z.B. bei der Reservierung von Speicher während einer Stringzuweisung. Muss nun
 zwischen Suspend und Resume Speicher reserviert werden, so kommt es zu einem
 Deadlock, da der kritische Abschnitt im suspendierten Thread belegt ist und
 erst freigegeben werden kann, wenn er fortgesetzt wird (Resume).

 TThread.WaitFor blockiert den Hauptthread, bis der Thread beendet wurde.
 Derweil wird auf keinerlei Windows-Messages reagiert - für den User scheint die
 Anwendung abgestürzt zu sein.


 --------------  Was bietet TAdvThread + Benutzung von TAdvThread  -------------

 Jeglicher Zugriff auf TAdvThread.Terminate und TAdvThread.Terminated erfolgt in
 einem kritischen Abschnitt. Das bewirkt, dass der Wert nicht im Cache bleibt,
 sondern sofort in den Hauptspeicher geschrieben wird und dadurch auch dann
 sofort im Thread ankommt, wenn dieser auf einem anderen Core mit eigenem Cache
 läuft.

 TAdvThread.WaitForAdv kehrt genauso wie TThread.WaitFor erst zurück, wenn der
 Thread beendet wurde, reagiert allerdings währenddessen auf Windows-Messages.

 Neues System zur Unterbrechung der Threadausführung (Suspend):
   -TAdvThread.SuspendAdv signalisiert dem Thread, dass er anhalten soll und
    kehrt sofort zurück (asynchron). Bei der nächsten Abfrage von Terminated
    bleibt der Thread stehen (Terminated kehrt nicht zurück - der Thread ist
    angehalten).
   -TAdvThread.WaitForSuspensionAdv prüft, ob Suspended gesetzt ist und kehrt
    sofort zurück, wenn das nicht der Fall ist. Ist Suspended gesetzt (= soll
    der Thread angehalten werden), dann wartet WaitForSuspensionAdv darauf, dass
    der Thread das nächste mal Terminated prüft (und somit anhält). Während
    gewartet wird, wird weiterhin auf Messages reagiert.
   -TAdvThread.Suspend ruft TAdvThread.SuspendAdv gefolgt von
    TAdvThread.WaitForSuspensionAdv auf und kann somit genauso wie
    TThread.Suspend benutzt werden.
   -TAdvThread.Resume hebt TAdvThread.Suspended auf und kehrt sofort zurück.
    Terminated kehrt zurück, sobald der Thread das nächste mal Rechenzeit
    erhält - der Thread läuft wieder.
   -Der Zugriff auf den Suspend-Counter ist in einem eigenen kritischen
    Abschnitt implementiert, um auch hier den Wert ggf. schnell in einen anderen
    Core mit eigenem Cache zu transportieren.
 Damit alles reibungslos funktioniert, darf Terminated niemals innerhalb
 kritischer Abschnitte überprüft werden. Dennoch muss es oft genug überprüft
 werden, da sonst die Performance der Suspend-Aufrufe leidet.
}

interface

uses
  Classes, SyncObjs;

type
  TSuspensionResult = (
    srNotSuspended, // Thread soll gar nicht angehalten werden (Suspended = False)
    srSuspended, // Thread ist jetzt angehalten
    srTerminated, // Threadausführung ist beendet
    srAbandoned, // Thread ist freigegeben worden
    srError // Ein Fehler ist aufgetreten -> GetLastError überprüfen
  );

  TAdvThread = class(TThread)
  private
    FSuspendCount: Cardinal;
    FSuspendCS: TCriticalSection;
    FSuspendEvent: TSimpleEvent; // gesetzt = Thread kann laufen (nur neue Version!)
    FSuspendedEvent: TSimpleEvent; // gesetzt = Thread ist tatsächlich angehalten
    FTerminateCS: TCriticalSection;
    FTerminated: Boolean;
    // Getter und Setter
    function GetSuspended: Boolean;
    procedure SetSuspended(const Value: Boolean);
    // sonstige Methoden
    function CheckTerminated: Boolean;
  protected
    property Terminated: Boolean read CheckTerminated;
  public
    // Konstruktoren und Destruktoren
    constructor Create(CreateSuspended: Boolean);
    destructor Destroy; override;
    // sonstige Methoden
    procedure Resume; reintroduce;
    procedure Suspend; reintroduce;
    procedure SuspendAdv; // asynchron -> kehrt zurück, bevor der Thread anhält
    procedure Terminate; reintroduce;
    function WaitForAdv: LongWord;
    function WaitForSuspensionAdv: TSuspensionResult; // wartet, bis Thread anhält
    // Eigenschaften
    property Suspended: Boolean read GetSuspended write SetSuspended;
  end;

implementation

uses
  Windows, SysUtils, Forms;

{ TAdvThread }

constructor TAdvThread.Create(CreateSuspended: Boolean);
begin
  // FTerminateCS muss vor Anlauf des Threads mit einer Instanz von
  // TCriticalSection gefüllt werden. Das muss also vor dem inherited-Aufruf
  // erfolgen, da der Thread dort anläuft, wenn CreateSuspended = False
  // Felder, die in dieser Klasse neu definiert worden sind, können vor dem
  // Aufruf des inherited Konstruktors benutzt werden (der Speicher ist bereits
  // vor dem Aufruf dieses Konstruktors reserviert und mit Null initialisiert).
  FSuspendCS := TCriticalSection.Create;
  FSuspendEvent := TSimpleEvent.Create;
  FSuspendEvent.SetEvent; // Event gesetzt = Thread ist nicht angehalten
  FSuspendedEvent := TSimpleEvent.Create; // nicht gesetzt = Thread läuft
  FTerminateCS := TCriticalSection.Create;
  inherited;
end;

destructor TAdvThread.Destroy;
begin
  FTerminateCS.Free;
  FSuspendedEvent.Free;
  FSuspendEvent.Free;
  FSuspendCS.Free;
  inherited;
  // wären die Freigaben auch hier (hinter inherited) möglich/sicher?
end;

// private

function TAdvThread.GetSuspended: Boolean;
begin
  Result := inherited Suspended;
  if not Result then
  begin
    FSuspendCS.Enter;
    try
      Result := FSuspendCount > 0;
    finally
      FSuspendCS.Leave;
    end;
  end;
end;

procedure TAdvThread.SetSuspended(const Value: Boolean);
begin
  if Value <> Suspended then
    if Value then
      Suspend
    else
      Resume;
end;

function TAdvThread.CheckTerminated: Boolean;
var
  Suspended: Boolean;
begin
  Result := False;
  FSuspendCS.Enter;
  try
    Suspended := FSuspendCount > 0;
  finally
    FSuspendCS.Leave;
  end;
  // Der Aufruf von FSuspendEvent.WaitFor darf in keinem kritischen Abschnitt
  // sein, sonst könnte ein Deadlock auftreten
  if Suspended then
  begin
    // bekanntgeben, dass der Thread jetzt angehalten ist
    FSuspendedEvent.SetEvent;
    try
      case FSuspendEvent.WaitFor(INFINITE) of
        wrAbandoned: // FSuspendEvent wurde freigegeben -> terminieren
          Result := True; // Terminated soll auf True gesetzt werden
        wrError: // während des Wartens ist ein Fehler aufgetreten
          raise Exception.Create(
            'Error in Object of Class ' + Self.ClassName + ':'#13 +
            'In ancestor-class TSafeThread in procedure CheckTerminated an ' +
            'Errorcode ' + IntToStr(FSuspendEvent.LastError) + ' was ' +
            'returned by call to TSimpleEvent.WaitFor');
        wrSignaled: // Resume wurde aufgerufen
          ; // nichts zu tun
        wrTimeout: // das Timeout INFINITE wurde erreicht ;-)
          ; // nichts zu tun
      end;
    finally
      // bekanntgeben, dass der Thread wieder läuft
      FSuspendedEvent.ResetEvent;
    end;
  end;
  // Result von Terminated erst zum spätestmöglichen Zeitpunkt feststellen
  FTerminateCS.Enter;
  try
    // falls FSuspendEvent.WaitFor wrAbandoned zurückgegeben hat (genau dann ist
    // Result = True), muss der Thread terminiert werden
    FTerminated := FTerminated or Result;
    Result := FTerminated;
  finally
    FTerminateCS.Leave;
  end;
end;

// protected

// public

procedure TAdvThread.Resume;
begin
  // Wenn der Thread suspendiert erstellt wurde (Create(True)), oder sonstwie
  // auf herkömmliche Art und Weise suspendiert wurde, dann muss er auch auf
  // herkömmliche Art und Weise wieder "angeschubst" werden!
  if inherited Suspended then
    inherited Resume
  else
  begin
    FSuspendCS.Enter;
    try
      if FSuspendCount > 0 then
        Dec(FSuspendCount);
      if FSuspendCount = 0 then
        FSuspendEvent.SetEvent; // Event gesetzt = Thread ist nicht angehalten
    finally
      FSuspendCS.Leave;
    end;
  end;
end;

procedure TAdvThread.Suspend;
begin
  SuspendAdv;
  WaitForSuspensionAdv;
end;

procedure TAdvThread.SuspendAdv;
begin
  FSuspendCS.Enter;
  try
    if FSuspendCount = 0 then
      FSuspendEvent.ResetEvent; // Event rückgesetzt = Thread ist angehalten
    Inc(FSuspendCount);
  finally
    FSuspendCS.Leave;
  end;
end;

procedure TAdvThread.Terminate;
begin
  FTerminateCS.Enter;
  try
    FTerminated := True;
  finally
    FTerminateCS.Leave;
  end;
end;

function TAdvThread.WaitForAdv: LongWord;
var
  H: THandle;
begin
  H := Handle;
  // MsgWaitForMultipleObjects kehrt nur bei NEUEN Messages zurück
  // -> vorhandene Messages vorher bearbeiten
  Application.ProcessMessages;
  while MsgWaitForMultipleObjects(1, H, False, INFINITE,
                                  QS_ALLINPUT) = WAIT_OBJECT_0 + 1 do
    Application.ProcessMessages;
  GetExitCodeThread(H, Result);
end;

function TAdvThread.WaitForSuspensionAdv: TSuspensionResult;
const
  AnzHandles = 2;
var
  Handles: array[0..AnzHandles-1] of THandle;
begin
  Result := srNotSuspended;
  if not Suspended then
    Exit;
  // wenn der Thread auf althergebrachte Art und Weise angehalten wurde, dann
  // muss nicht gewartet werden.
  if inherited Suspended then
  begin
    Result := srSuspended;
    Exit;
  end;
  Handles[0] := Handle;
  Handles[1] := FSuspendedEvent.Handle;
  // MsgWaitForMultipleObjects kehrt nur bei NEUEN Messages zurück
  // -> vorhandene Messages vorher bearbeiten
  Application.ProcessMessages;
  repeat
    case MsgWaitForMultipleObjects(AnzHandles, Handles, False, INFINITE,
                                   QS_ALLINPUT) of
      WAIT_OBJECT_0 + AnzHandles: // neue Messages vorhanden
        Application.ProcessMessages;
      WAIT_OBJECT_0: // Threadausführung ist beendet
        Result := srTerminated;
      WAIT_OBJECT_0 + 1: // Thread ist angehalten
        Result := srSuspended;
      WAIT_ABANDONED_0 .. WAIT_ABANDONED_0 + (AnzHandles-1):
        Result := srAbandoned; // self ist freigegeben worden :-p
      $FFFFFFFF: // Error
        Result := srError;
    end;
  until Result <> srNotSuspended;
end;

end.

[DelphiSecurity]

Omg! Da schau ich nicht mehr durch!

DS

[RSE]

Immer gerne fragen, wo´s klemmt! Es sieht zwar viel aus, aber ich denke es ist gut strukturiert und kommentiert. Auch dazu nehme ich gerne Hinweise an.

[SirThornberry]

Du hast bei der Deklaration des Constructors override vergessen?!
[Edit]
Doch nicht, die haben doch tatsächlich den Constructor nicht virtell gemacht.
[/Edit]

[RSE]

Also zumindest in Delphi 5 ist der Konstruktor von TThread nicht virtuell... daher kann ich ihn auch nicht mit override überschreiben sondern lediglich verdecken.