Daten parallel mithilfe eines Ringbuffers wegspeichern - Critical Sect

[himitsu]

wenn dir 16 MB nicht zuviel Puffer sind, könntest du FHead und FTail auch einfach als Byte definieren und dann statt 0..10 auf 0..255 umsteigen, da würde die Überlaufprüfung (if FHead > 10 then) entfallen, da sie ja schon im Typ eingebaut wäre :angel:

hier würde also ein einfaches "Inc" ausreichen

Delphi-Quellcode:

procedure TRingBuffer.IncHead;
begin
  InterlockedIncrement(FHead);
end;

ansonsten würde ich besser noch den Inhalt vorm Setzen vergleichen,

Delphi-Quellcode:

procedure TRingBuffer.IncHead;
begin
  if InterlockedCompareExchange(FHead, 10, 0) <> 10 then
    InterlockedIncrement(FHead);
end;

(ich hoff ich hab das jetzt richtig gemacht, aber im Prinzip würd ich es irgendwie so machen :angel2: )

denn z.B. bei

Delphi-Quellcode:

procedure TRingBuffer.IncHead;
begin
  InterlockedIncrement(FHead);
  if FHead > 10 then
    //FHead := 0;
    InterlockedXor(FHead, 0);
end;

ist FHead hier kurzzeitig außerhalb des "zuläßigen" Bereichs und wenn gerade in der, wenn auch sehr kurzen Zeitspanne ein anderer Thread den wert ausließt, dann bekommt der ja einen "ungültigen" Wert geliefert

[Viktorii]

Nochmals vielen Dank an alle.

Habe mich vorerst für sirius' Lösung entschieden:

Delphi-Quellcode:

if FHead<10 then
  InterlockedIncrement(FHead)
else
  InterlockedExchangeAdd(FHead,-9);

Auch wenn es so jetzt sauberer ist, möchte ich die anderen Möglichkeiten auch noch verstehen.

Kann mir hierbei noch jemand auf die Sprünge helfen:

Zitat:

Zitat von sirius

Allerdings ist das nicht die prinzipielle Vorgehensweise. Diese würde zwar ein paar Performanceverluste bringen, dafür aber "sicherer" sein. Das heißt, man kann einen Synchronisationsfehler nicht so leicht übersehen, wie in deinem Code.
Du müsstest dazu den Zugriff auf DataArray komplett kapseln. Dann liest du mit GetData in eine lokale Variable und kopierst diese gesichert nach in deinen Buffer.

:?:

[sirius]

Eine vollständige Lösung sähe vielleicht folgendermaßen aus. Wie gesagt: das ist jetzt etwas schlechter von der Performance her und auch etwas größer. Deswegen würde ich bei deinem Problem auch etwas kleineres wählen. Beim Schreiben dieses Codes fielen mir aber noch zwei Probleme in deinem Code auf.

1. Threadsteuerung mit Suspend und Resume zu machen ist nicht so gut und wird von M$ nicht empfohlen. Besser sind Events.
2. Was machst du, wenn dein GetDataThread den SaveDatathread einholt und beide auf das gleiche Array schreiben? Bzw.: Wie verhinderst du dies?

Ich habe mal ein Beispiel versucht.
Also die Idee ist eine Klasse, welche die CS und den Zugriff auf den Inhalt verwaltet:

Delphi-Quellcode:

type
     //Erstmal packen wir das Array in eine Klasse, wenn auch noch sehr rudimentär
     TDataArray=class(TPersistent)
      public
        //ist mal nicht mit einem property gekapselt, weil ich eh nur den Pointer darauf benötige
        //evtl. wird das ja hier noch anders verwendet
        Values:array[1..65536] of word;
      protected
        //zum Kopieren der Klasse
        procedure AssignTo(Dest: TPersistent); override;
     end;

     //der Zweite Teil des Array wird eine List
     //hier eine TObjectList nur mit angpeassten Methoden, damit man nicht
     //ständig außerhalb mit dem As-Operator handeln muss
     //zudem wird noch auf Notify reagiert
     TDataList=class(TObjectList)
       private
         FonAdd:TNotifyevent;
       protected
         procedure Notify(Ptr: Pointer; Action: TListNotification); override;
         function GetItem(Index: Integer): TDataArray;
         procedure SetItem(Index: Integer; ADataArray: TDataArray);
       public
         property Items[Index: Integer]: TDataArray read GetItem write SetItem; default;
         property OnAdd:TNotifyEvent read FonAdd write FonAdd;
     end;


     //Kapselung der Liste mit einer CS
     TThreadDataList=class
       Constructor Create; reintroduce;
       Destructor Destroy; override;
     private
       FList:TDataList;
       FLock:TCriticalSection;
     public
        procedure Add(Item: TDataArray);
        function  LockList: TDataList;
        procedure UnlockList;
     end;

Die "eigentliche" Klasse ist TThreadDataList. Darin enthalten ist eine Liste(TDataList). Und diese Liste beinhaltet das Array (TDataArray)
TDataArray hat nur das Array für einen Lese-Vorgang und noch die AssignTo Methode um die Klasse zu kopieren (geerbt von TPersistent).
TDataList verdeckt eigentlich nur die Methoden von TObjectList um aus den Parametern TObject TDataArray zu machen. Ist dann einfacher in der Anwendung. Außerdem wird Notify überschrieben, so dass darin reagiert werden kann, wann die Liste >10 Einträge hat, damit der älteste gelöscht wird. Außerdem kann ein Ereigniss ausgelöst werden (FonAdd)
TThreadDataList hat nur drei Methoden. Add um einen Eintrag der obigen Liste hinzuzufügen (mit CS gekapselt) und das Paar LockLsit und Unlocklist um die Liste nur bei gleichzeitiger Synchronisation zu bekommen (siehe dazu TThreadList)
Der Code hierzu ist:

Delphi-Quellcode:

function TDataList.GetItem(Index: Integer): TDataArray;
begin
  result:=inherited GetItem(Index) as TDataArray;
end;

procedure TDataList.Notify(Ptr: Pointer; Action: TListNotification);
begin
  inherited;
  if Action=lnAdded then
  begin
    while Count>10 Do Delete(0);
    if assigned(FonAdd) then FonAdd(self); //siehe TSaveDataThread.ListAdd
  end;
end;

procedure TDataList.SetItem(Index: Integer; ADataArray: TDataArray);
begin
  inherited SetItem(Index,ADataArray);
end;

{ TThreadDataList }

procedure TThreadDataList.Add(Item: TDataArray);
begin
  LockList;
  try
    FList.Add(Item);
  finally
    UnlockList;
  end;
end;

constructor TThreadDataList.Create;
begin
  FLock:=TCriticalSection.Create;
  FList:=TDataList.Create(True);
end;

destructor TThreadDataList.Destroy;
var List:TList;
begin
  LockList;
  try
    FList.Free;
    inherited;
  finally
    FLock.Free;
  end;
end;

function TThreadDataList.LockList: TDataList;
begin
  FLock.Enter;
  result:=FList;
end;


procedure TThreadDataList.UnlockList;
begin
  FLock.Leave;
end;


{ TDataArray }

procedure TDataArray.AssignTo(Dest: TPersistent);
begin
  inherited;
  (Dest as TDataArray).FValues:=FValues;
end;

Die Threads habe ich mal kurz so implementiert:

Delphi-Quellcode:

type
     //BeispielThreads:
     TDataThread=class(TThread)
        Constructor Create(CreateSuspended:Boolean; ThreadDataList:TThreadDataList); reintroduce;
      protected
        FThreadDataList:TThreadDataList;
     end;

     TGetDataThread=class(TDataThread)
      protected
        procedure Execute; override;
     end;
     TSaveDataThread=class(TDataThread)
      protected
        procedure Execute; override;
        procedure ListAdd(Sender:TObject);
      private
        FEvent:TEvent;
      public
        procedure Terminate; reintroduce;
     end;


//...

{ TDataThread }

constructor TDataThread.Create(CreateSuspended: Boolean;
  ThreadDataList: TThreadDataList);
begin
  inherited Create(CreateSuspended);
  FThreadDataList:=ThreadDataList;
end;

{ TGetDataThread }

procedure TGetDataThread.Execute;
var DataArray:TDataArray;
    List:TDataList;
begin
  while not terminated do
  begin
    DataArray:=TDataArray.Create;
    Get_Data(@DataArray.Values);
    List:=FThreadDataList.LockList;
    try
      List.Add(DataArray);
    finally
      FThreadDataList.UnlockList;
    end;
  end;
end;

{ TSaveDataThread }

procedure TSaveDataThread.Execute;
var List:TDataList;
    DataArray:TDataArray;
    isEntry:Boolean;
begin
  FEvent:=TEvent.Create(nil,false,true,'');
  List:=FThreadDataList.LockList;
  try
    List.OnAdd:=ListAdd;
  finally
    FThreadDataList.UnlockList;
  end;

  while not terminated do
  begin
    FEvent.WaitFor(infinite);

    repeat
      List:=FThreadDataList.LockList;
      try
        isEntry:=List.Count>0;
        if isEntry then
        begin
          DataArray.Assign(List.Items[0]);
          List.Delete(0);
        end;
      finally
        FThreadDataList.UnlockList;
      end;

      if isEntry then
      begin
        FFileStream.Write(DataArray.Values[1], NUMBER_OF_DATA_BYTES);
      end;

    until not isEntry;

  end;

end;

procedure TSaveDataThread.ListAdd(Sender: TObject);
begin
  FEvent.SetEvent;
end;

procedure TSaveDataThread.Terminate;
begin
  inherited;
  FEvent.SetEvent;
end;

die MEssage brauche ich gar nicht mehr, ich benutze einfach das Event. In Waitfor wartet der Thread, bis die Liste über ListAdd das Event setzt und der Durchlauf einmal beginnt, bis er wieder bei Waitfor ist.

[Viktorii]

Hallo sirius,

vielen Dank, dass Du dir die Mühe gemacht hast mir so ausführlich zu antworten. Muss mir das nachher erstmal in aller Ruhe zu Gemüte führen. Vermutlich wird da noch die ein oder andere Frage bei mir noch aufkommen :-D

Zitat:

Zitat von sirius

Threadsteuerung mit Suspend und Resume zu machen ist nicht so gut und wird von M$ nicht empfohlen. Besser sind Events.

Okay, dann müsste ich mal schauen dass ich das umstricke... Welchen Vorteil hat denn die Lösung mit Events und WaitForSingleObject bzw. Waitfor gegenüber den suspenden?

Zitat:

Zitat von sirius

Was machst du, wenn dein GetDataThread den SaveDatathread einholt und beide auf das gleiche Array schreiben? Bzw.: Wie verhinderst du dies?

Tja, das ist echt ein Problem. Wenn der GetDataThread schneller ist als der SaveDatathread wir er ihn auf kurz über lang ja auf jeden einholen, egal wie groß der Ringbuffer ist. Das prüfe ich ja in der WndProc und im Fehlerfall gebe ich eine Messagebox aus. Zusätzlich muss ich da natürlich noch ErrorHandling betreiben, was hier im Code aber nicht vorhanden ist:

Delphi-Quellcode:

procedure TMainForm.WndProc(var Message: TMessage);
begin
  if (Message.Msg = WM_GET_DATA_COMPLETE) then
  begin
    // Wenn Head(-index) <> Tail(-index) -> alles OK
    if FRingBuffer.Head <> FRingBuffer.Tail then
    begin
      // SaveDataThread aufwecken um gerade bekommenen Daten auf HD zu speichern
      FSaveDataThread.Resume;
    end
    else
    begin
      // Fehler. Head hat den Tail eingeholt :-(
      MessageBox(0, pWideChar('Overrun!'), '', MB_ICONSTOP or MB_OK);
    end;
  end;
  inherited;
end;

[jaenicke]

Zitat:

Zitat von Viktorii

Okay, dann müsste ich mal schauen dass ich das umstricke... Welchen Vorteil hat denn die Lösung mit Events und WaitForSingleObject bzw. Waitfor gegenüber den suspenden?

Wenn du den Thread von außen schlafen legst, kennst du dessen Zustand nicht unbedingt. Wenn du dem Thread nur sagst, dass er das tun soll, dann kann der ggf. noch fertig rechnen usw., oder z.B. eine offene Datei vorher schließen.

Und zum Einholen: Genau das würdest du mit Pipes vermeiden. Denn da kommen eben an einer Seite Daten rein und können am anderen abgeholt werden, mehr nicht.

[sirius]

Zu suspend:

Zitat:

Zitat von msdn

This function is primarily designed for use by debuggers. It is not intended to be used for thread synchronization. Calling SuspendThread on a thread that owns a synchronization object, such as a mutex or critical section, can lead to a deadlock if the calling thread tries to obtain a synchronization object owned by a suspended thread. To avoid this situation, a thread within an application that is not a debugger should signal the other thread to suspend itself. The target thread must be designed to watch for this signal and respond appropriately.

Du machst zwar das supend innerhalb deines Threads. Insofern könnte das sogar ok sein, aber ich würde trotzdem die Finger davon lassen.