Kurze Frage zur Thread Sicherheit

[luke2]

Hi, ist es sicher, wenn ein Thread lesend und schreibend auf ein Index eines Array zugreift und ein anderer Thread lesend und schreibend auf das gleiche Array mit einem anderen Index zugreift? Oder müssen alle Zugriffe synchronisiert werden?

[Zacherl]

Sollte theoretisch sicher sein, solange wirklich immer nur ein Thread auf einen Index gleichzeitig zugreift.

[luke2]

OK, danke für die Antwort.

[himitsu]

Und sich das Array selber nicht verändert, also kein SetLength und Co.

Was ist das denn für ein Array?

[geskill]

Hallo luke2,
du kannst dafür am besten einen MREW benutzen. Damit können mehrere Threads gleichzeitig lesend drauf zugreifen, aber immer nur einer schreibend. Natürlich müssen beide Threads die gleiche Instanz von dem TSafeArray bekommen ;)

Bei so kritischen Sachen würde ich nie wie du vorhast die Array Instanz einem Thread direkt mitteilen.

Dies ist eine vereinfachtere Variante die ich gerade im Einsatz habe:
Das ist an sich eigentlich schon ziemlich performant, aber du könntest jetzt ein lokales Array/Liste etc. anlegen, die alle Lesezugriffe auf die versch. Indexe speichert. Und beim Schreiben wird dann geprüft ob dieser Index in der Liste ist, wenn ja muss dort beim zugreifen eben gewartet werden. Aber ich würde das mit dem gleichzeitigen auf ein Index auch sein lassen und immer als schreibend behandeln.

Delphi-Quellcode:

uses
  // Delphi
  Math,
  // OmniThreadLibrary
  OtlSync;

type
  TSafeArray = class
  private
    FArray: array of Integer;
    FLock: TOmniMREW;
  public
    constructor Create;

    procedure SetResult(AIndex: Integer; const AValue: Integer);
    function GetResult(AIndex: Integer): Integer;
  end;

constructor TSafeArray.Create;
begin
  SetLength(FArray, 0);
end;

procedure TSafeArray.SetResult(AIndex: Integer; const AValue: Integer);
begin
  FLock.EnterWriteLock;
  try
    SetLength(FArray, Max(AIndex, length(FArray)) + 1); // Lesezugriff: length(); Schreibzugriff!: SetLength()
    FArray[AIndex] := AValue; // Schreibzugriff: FArray[Index]
  finally
    FLock.ExitWriteLock;
  end;
end;

function TSafeArray.GetResult(AIndex: Integer): Integer;
var
  Index: Integer;
begin
  Result := -1;

  Index := AIndex;

  FLock.EnterReadLock; // starte das lesen
  try
    if Index < length(FArray) then // Lesezugriff: length()
      Result := FArray[Index]; // Lesezugriff: FArray[Index]
  finally
    FLock.ExitReadLock; // beende das lesen
  end;
end;

Grüße

[luke2]

Danke, das sieht gut aus, geskill, aber ich möchte eigentlich keine externen Komponenten benutzen.

Es sieht ungefähr so aus:

Delphi-Quellcode:

type
  T = record
    Str: string;
    Use: Boolean;
  end;

private
  A: Array[0..5] of T;

procedure TThread.Add(const S: string);
begin
  with A[GetFreeField] do
  begin
    Str := S;
    Use := True;
  end;
end;

procedure TThread.Execute;
begin
  while not Terminated do
  begin
    for I := Low(A) to High(A) do
      if Use then
      begin
        //hier werden nur Felder von A[I] gelesen und geändert
      end;
    Sleep(1);
  end;
end;

Add wird aus dem Hauptthread aufgerufen. Use ist natürlich anfangs immer False und wird vom Thread auch irgendwann wieder auf False gesetzt. Das sollte doch sicher sein?

[geskill]

Also wenn du Add() aufrufst bevor der Thread startet, dann geht das in Ordnung.

Nur so musst du das immer wissen, wann du es aufrufen darfst. Bei kleinen Programmen ist das noch in Ordnung, aber man sollte sich besser direkt angewöhnen das sauber umzusetzen. In dem Beispiel bei mir kann man dies ohne zu überlegen einfach aufrufen.

Das Auslesen passiert dann auch erst wieder wenn der Thread definitiv beendet wurde (z.B. via Join)?

Da es ein statisches Array ist sind die Schreibtugriffe ja wirklich nur aus die einzelnen Elemente begrenzt.

Und mit 2 Threads meinst du Hauptthread und TThread - oder? ;)

[luke2]

Zitat:

Zitat von geskill (Beitrag 1185738)

Also wenn du Add() aufrufst bevor der Thread startet, dann geht das in Ordnung.

Add wird eigentlich nur aufgerufen, wenn der Thread bereits läuft. Was kann da denn passieren? Bisher lief die Anwendung eigentlich immer stabil.

Zitat:

Zitat von geskill (Beitrag 1185738)

Und mit 2 Threads meinst du Hauptthread und TThread - oder? ;)

Jep

Kann man dein Beispiel auch mit den Windows CriticalSections (TRTLCriticalSection) umsetzen, so dass keine externe Unit/Komponente benutzt werden muss?

[geskill]

Die Schleife ist ja kurz und bei den heutigen leistungsstarken PC gehört schon ein bisschen Glück dazu das ein Fehler Auftritt. Weil beim Add() schreibst du ja auf einen Index I und bei jedem lesen greifst du immer auf Used von I lesend zu. Das ist nicht in Ordnung.

Mit einer TCriticalSection kannst du das nun einschränken. Der Vorteil bei den OTL Sections ist das man sie nicht initialisieren und freigeben muss. Zudem ist der MREW auch "sicherer" als der von Delphi.

[luke2]

OK, vielen Dank. Eine letzte Frage: Ist es denn so sicher (auch wenn es keine optimale Lösung ist)?

Delphi-Quellcode:

type
  T = record
    Str: string;
    Use: Boolean;
  end;

private
  A: Array[0..5] of T;
  CS: TRTLCriticalSection;

procedure TThread.Add(const S: string);
begin
  EnterCriticalSection(CS);
  try
    with A[GetFreeField] do
    begin
      Str := S;
      Use := True;
    end;
  finally
    LeaveCriticalSection(CS);
  end;
end;

procedure TThread.Execute;
begin
  while not Terminated do
  begin
    for I := Low(A) to High(A) do
    begin
      EnterCriticalSection(CS);
      if Use then
      begin
        LeaveCriticalSection(CS);
        //hier werden nur Felder von A[I] gelesen und geändert
      end
      else
        LeaveCriticalSection(CS);
    end;
    Sleep(1);
  end;
end;

[Klaus01]

Delphi-Quellcode:

procedure TThread.Execute;
begin
  while not Terminated do
  begin
    for I := Low(A) to High(A) do
    begin
      EnterCriticalSection(CS);
      if Use then
      begin
        //hier werden nur Felder von A[I] gelesen und geändert
      end;
      LeaveCriticalSection(CS)
    end;
    Sleep(1);
  end;
end;

Delphi-Quellcode:

procedure TThread.Execute;
begin
  while not Terminated do
  begin
    EnterCriticalSection(CS);
    try
      for I := Low(A) to High(A) do
      begin
        if Use then
          begin
            //hier werden nur Felder von A[I] gelesen und geändert
          end;
      end;
    finally
      LeaveCriticalSection(CS)
    end;
    Sleep(1);
  end;
end;

oder TThreadList verwenden.

Grüße
KLaus

[luke2]

So kann ich es nicht machen, da das den Thread nutzlos machen würde. Es werden in der Schleife einige Berechnungen durchgeführt. Dies würde dazu führen, dass der Hauptthread beim Einfügen einfriert.

[Aphton]

Der Sinn der Sache ist ja, dass beim Einfügen bzw. Ändern einer Liste kein anderer Thread Zugriff darauf haben sollte.

[Medium]

Da sich das Array selbst ja nicht ändert, könnte man hier für jedes Array-Element eine eigene CS (als Record-Member) nehmen. Das Array an sich so ist safe, so lange seine Länge nie verändert wird. Nur kann es dir passieren, dass ein Add() Aufruf nur halb fertig ist, und du ggf. "unfertige" Daten in dem Record hast. Und gaaaaanz selten kann da ggf., grad bei der String-Zuweisung, ein Zugriffskonflikt passieren. Ganz selten, weil du über das Setzen von "use" ganz am Schluss schon ein wenig Sicherheit hast, aber das ist trotzdem ein bischen Spiel mit dem Feuer. Sowas kann wochenlang sauber laufen, und dann wundert man sich (evtl. nach Änderungen, bei denen man nicht an diese besonderen Umstände dachte) woher auf einmal diese grausam schlecht zu debuggenden sporadischen AVs her kommen.

Bei Add() mit Threads wurde ich übrigens hellhörig: Ich hatte mal den Fauxpas begangen, eine einfache TObjectList mit Threads einzusetzen. Dort kann ein Add() ein SetLength() des internen Arrays auslösen, was meinem Thread 1-2x pro Woche (bei einer Anwendung, die 24/7 läuft) seine Daten unterm Poppo weggezogen hat. (Vermutlich öfter, aber der Speicher war wohl meist noch mit gültigem Kram belegt.) Eigentlich so offensichtlich, bei mir aber nach diversen Änderungen über Monate verteilt eingeschlichen, ohne dass ich es bewusst gemerkt habe. Gebranntes Kind scheut das Feuer: Seit dem bin ich mit CriticalSections per du =)

[Bummi]

Man könnte dem Thread auch über eine "gelockte" Property mitteilen, wie weit er Save arbeiten kann (ich nehme an er arbeitet in einer Schleife)

[luke2]

Medium / Bummi
Danke für die Antworten. Habt ihr evtl. ein ganz kurzes Beispiel wie man es (nur mit TRTLCriticalSection) machen könnte, damit es ganz sicher ist?

[Bummi]

Den Setter/Getter der Property mit einer CriticalSection wrappen, im Thread auch nur auf die Property nicht auf die dahinterliegende Variable zugreifen.

[Sir Rufo]

Wenn du schon XE2 einsetzt, warum willst du unbedingt mit Arrays arbeiten?
IMHO wäre eine TQueue hier genau richtig. Die Verarbeitung dauert relativ lange (hast du gesagt), dann würde ich auf jeden Fall nicht erst alles abarbeiten und dann zu schauen, ob der Thread evtl. abgebrochen wird, sondern immer nur ein Paket bearbeiten.

Delphi-Quellcode:

TMyThread = class( TThread )
strict private
  FCS : TCriticalSection;
  FQueue : TQueue<TMyRecord>;
protected Execute; override;
public
  procedure Add( ARecord : TMyRecord );
end;

TMyThread.Add( ARecord : TMyRecord );
begin
  FCS.Enter;
  try
    FQueue.Enqueue( ARecord );
  finally
    FCS.Leave;
  end;

TMyThread.Execute;
var
  LRecord : TMyRecord;
  LHasData : Boolean;
begin
  while not Terminated do
    begin
      // Raus aus der Queue
      FCS.Enter;
      try
        LHasData := ( FQueue.Count > 0 );
        if LHasData then
          LRecord := FQueue.Dequeue;
      finally
        FCS.Leave;
      end; 

      // Hier bietet sich auch wunderbar ein Event an, da der Thread dann solange schläft
      // bis wieder Daten vorhanden sind, oder der Thread terminiert wurde

      // Wenn keine Daten, dann schlafen und wieder von Vorne
      if not LHasData then
        begin
          Sleep( 1 );
          Continue;
        end;

      // Bearbeiten
      if LRecord.Use then
        ...

      // Rein in die Queue (wenn es denn sein soll)
      if LRecord.Use then
        begin
          FCS.Enter;
          try
            FQueue.Enqueue( LRecord );
          finally
            FCS.Leave;
          end;
        end;
    end;
end;

[Furtbichler]

Also ich mach das mit einer Semaphore und einer CS.

Delphi-Quellcode:

Procedure TMyThread.Add(aSomething : TData);
Begin
  FCS.Enter;
  Try
    Queue.Enqueue(aSomething);
    ReleaseSemaphore(FSemaphore,1,nil);
  Finally
    FCS.Leave
  End
End;

Function TMyThread.GetNextData : TData;
Begin
 FCS.Enter;
 Try
   If Queue.Count>0 then
     Result := Nil
   else
     Result := Queue.Dequeue;
 Finally
   FCS.Leave;
 End;
End;
     
Procedure TMyThread.Execute;
Begin
  While not Terminated do
    If WaitForSingleObject(FSemaphore,INFINITE)=WAIT_OBJECT_0 then 
      DoWhatever(GetNextData); 
end;

Ich sträube mich, einen Thread mit Sleep(1) im Kreis rennen zu lassen.