Delphi-PRAXiS - Delphi TAdvThread mit "soft" Suspend und "nonblocking" WaitFor

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.