Daten parallel mithilfe eines Ringbuffers wegspeichern - Critical Sect

[Viktorii]

Moin zusammen.

Ich möchte eine Funktion (Get_Data) aufrufen, welche eine gewisse Zeit aktiv bleibt (z.B 500 ms) und welche gewisse Daten liefert. Unmittelbar nach derm wiederkehren soll die Funktion wieder aufgerufen werden um die nächste Daten zu liefern.

Während dessen (also z.B. in den nächsten 500 ms) sollen die Daten die die Funktion beim letzen mal geliefert hat weggespeichert werden.

Um diese zu realisieren habe ich mir einen Ringbuffer geschaffen. Desweiteren habe ich den Get_Data-Funktionsaufruf sowie das Wegspeichern der Daten in einen separaten Thread ausgelagert.

Die ganze Geschichte funktioniert soweit einwandfrei. Ich habe jedoch zwei Fragen:

1. Sind die Critical Sections im 'Ring Buffer' sinnvoll bzw. nötig?
2. Ist das Konzept so in Ordnung, oder kann man etwas besser, effizienter der eleganter lösen? Da gehe ich mal starkt von aus, da dies das erste mal ist, das ich sowas geschrieben habe. Für jegliches konstruktives Feedback bin ich sehr dankbar.


Jetzt zur Implementierung:


Erstellung der einzelnen Objekte:

Delphi-Quellcode:

procedure TMainForm.FormCreate(Sender: TObject);
begin
  ...
  ...
  FGetDataThread := TGetDataThread.Create(TRUE);    
  FSaveDataThread := TSaveDataThread.Create(TRUE);  

  FRingBuffer := TRingBuffer.Create;                

  FGetDataThread.RingBuffer := FRingBuffer;      
  FSaveDataThread.RingBuffer := FRingBuffer;     

  WM_GET_DATA_COMPLETE    := RegisterWindowMessage('{4F775E08-98FB-41b0-9852-10C7C13DD81C}');
end;

In der überschriebenen Fensterprocedure wir der Thread zum wegspeichern der Daten angestoßen, sobald Daten von den Get_Data-Thread verfügbar sind:

Delphi-Quellcode:

procedure TMainForm.WndProc(var Message: TMessage);
begin
  if (Message.Msg = WM_GET_DATA_COMPLETE) then
  begin
    // Wenn Head(-pointer) <> Tail(-pointer) -> 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;

Im GetData-Thread wird die Funktion Get_Data ausgeführt, der "Head-Pointer" wird inkrementiert und die Fensterprocedure wird benachrichtigt, dass nun Data vorliegen. Unmittelbar danach wird die Get_Data-Funktion erneut aufgerufen:

Delphi-Quellcode:

{ TGetDataThread }
procedure TGetDataThread.Execute;
begin
  while not terminated do
  begin
    ret := Get_Data(@FRingBuffer.pDataArray[FRingBuffer.Head][1]);

    FRingBuffer.IncHead;
    SendMessage(MainForm.Handle, MainForm.WM_GET_DATA_COMPLETE, 0, 0);
  end;
end;

Im SaveData-Thread werden die Daten solange weggespeichert, bis der "Tail-Pointer" den "Head-Pointer" erreicht hat. Ist dies der Fall liegen (noch) keine weiteren Daten vor:

Delphi-Quellcode:

{ TSaveDataThread }
procedure TSaveDataThread.Execute;
begin
  repeat
    repeat
      FFileStream.Write(FRingBuffer.pDataArray[FRingBuffer.Tail][1], NUMBER_OF_DATA_BYTES);
      FRingBuffer.IncTail;
    until (FRingBuffer.Tail = FRingBuffer.Head);

    suspend;
  until (Terminated);

  FreeAndNil(FFileStream);
end;

Hier nun der Ringbuffer mit den Critical Sections:

Delphi-Quellcode:

type
  TBuffer     = array[0..10, 1..65536] of word;
  TpBuffer    = ^TBuffer;

type
  TRingBuffer = Class (TObject)
  private
    FDataArray  : TBuffer;
    FpDataArray : TpBuffer;
    FHead       : Integer;
    FTail       : Integer;
    FCSHead     : TCriticalSection;
    FCSTail     : TCriticalSection;
    function GetHead: Integer;
    function GetTail: Integer;
  public
    Constructor Create;
    Destructor Destroy; override;
    procedure IncHead;
    procedure IncTail;
    property Head: Integer read GetHead;
    property Tail: Integer read GetTail;
    property pDataArray: TpRolBufDataOfAllCCDs read FpDataArray;
  end;


implementation

uses
  SysUtils;




{ TRingBuffer }
constructor TRingBuffer.Create;
begin
  FpDataArray := @FDataArray;
  FHead := 0;
  FTail := 0;
  FCSHead := TCriticalSection.Create;
  FCSTail := TCriticalSection.Create;
end;

//------------------------------------------------------------------------------

destructor TRingBuffer.Destroy;
begin
  FreeAndNil(FCSHead);
  FreeAndNil(FCSTail)
end;

//------------------------------------------------------------------------------

function TRingBuffer.GetHead: Integer;
begin
  FCSHead.Enter;
  result := FHead;
  FCSHead.Leave;
end;

//------------------------------------------------------------------------------

function TRingBuffer.GetTail: Integer;
begin
  FCSTail.Enter;
  result := FTail;
  FCSTail.Leave;
end;

//------------------------------------------------------------------------------

procedure TRingBuffer.IncHead;
begin
  FCSHead.Enter;
  inc(FHead);
  if FHead > 10 then
    FHead := 0;
  FCSHead.Leave;
end;

//------------------------------------------------------------------------------

procedure TRingBuffer.IncTail;
begin
  FCSTail.Enter;
  inc(FTail);
  if FTail > 10 then
    FTail := 0;
  FCSTail.Leave;
end;

Vielen Dank im Vorrauch für jegliche Tipps, Hinweise und Verbesserungsvorschläge.

[jaenicke]

Also erst entfernst du im Delphi-Treff ein paar Minuten nach deinem Post den Text und fragst dann direkt hier nach. Willst du hier auch den Beitrag gleich wieder leeren?

Jetzt habe ich wirklich gar keine Lust mehr etwas dazu zu schreiben nachdem ich den Post drüben direkt wieder verwerfen konnte.

[Viktorii]

Zitat von jaenicke:

Also erst entfernst du im Delphi-Treff ein paar Minuten nach deinem Post den Text und fragst dann direkt hier nach. Willst du hier auch den Beitrag gleich wieder leeren?

Wollte hier posten, habe aber aus versehen dort gepostet. Der Beitrag stand dort auch nur wenige Minuten und wurde mittlerweile entfernt. Bin leider nur ein Mensch nur mache manchmal etwas falsch. So wie Du mich hier anblökst scheint Dir sowas ja nie zu passieren. Da beneide ich Dich Drum

Zitat von jaenicke:

Jetzt habe ich wirklich gar keine Lust mehr etwas dazu zu schreiben nachdem ich den Post drüben direkt wieder verwerfen konnte.

Dann lass es halt

[jaenicke]

Zitat von Viktorii:

Wollte hier posten, habe aber aus versehen dort gepostet.

Trotzdem hättest du ihn dort ja nicht gleich leeren müssen, sowas ist kein guter Stil. Was solls, ist ja nicht mehr zu ändern.

Zitat von Viktorii:

Bin leider nur ein Mensch nur mache manchmal etwas falsch. So wie Du mich hier anblökst scheint Dir sowas ja nie zu passieren. Da beneide ich Dich Drum

So meinte ich das nicht, jeder macht Fehler, aber deshalb gleich den Thread dort zu leeren statt auf Antworten zu warten... Egal.
Was ich eben meinte war, dass ich gerade einiges zu deinem Quelltext geschrieben hatte und als ich es abschicken wollte war dein Beitrag schon leer...


Zum Quelltext habe ich keine Lust das noch einmal zu schreiben.
Aber dann wollte ich noch auf Pipes hinweisen. Diese eignen sich sehr gut zur Datenweitergabe in eine Richtung, in deinem Fall vom Thread, der die Berechnungen durchführt zu dem, der die Speicherung macht. Ich weiß nicht wie gut das bei dir passt, könnte aber evtl. einen fließenden Datenstrom ermöglichen statt wie bisher alle x Millisekunden.

Denn bei einer Pipe ist das Schöne, dass der Thread jeweils einfach darauf warten kann, dass neue Daten ankommen und dann diese direkt verarbeiten kann.

[sirius]

1. Du brauchst hier kein RegisterWindowMessage, da du anscheinend ausschließlich in deinem eigenen Process bist und auch sonst braucht man es recht selten. Leg einfach eine Zahl bei WM_User+X fest (X als kleine natürliche Zahl).
2. So, wie es jetzt geschrieben ist, brauchst du keine CriticalSections. Die dauern viel zu lange. Hier kommt man mit den Interlockedxxxx- Funktionen weiter. Wenn du diese bei schreibenden Vorgängen benutzt, brauchst du bei lesenden Vorgängen keine Absicherung.
3. Das prinzipielle Vorgehen scheint so zu funktionieren (ich entdecke keinen Fehler beim Drüberschauen) und da es sehr speziell an dein Problem angepasst ist, dürfte es auch sehr effizient sein.
   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.

[Viktorii]

Vielen Dank für Eure Antworten

Über Pipes bin ich gerade am lesen... Interlockedxxxx- Funktionen werde ich mir danach anschauen...

Zitat von sirius:

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.

Könntest Du das evtl. etwas genauer ausführen? Das Prinzip ist mir nicht ganz klar geworden

[jaenicke]

Zitat von sirius:

Leg einfach eine Zahl bei WM_User+X fest (X als kleine natürliche Zahl).

Um genau zu sein: WM_APP + X, denn die Werte zwischen WM_USER und WM_APP sollten nur innerhalb einer Fensterklasse verwendet werden, da sie auch in Fensterklassen vom System verwendet werden. Da hier die Nachrichten in der WndProc der MainForm abgefangen werden, könnten diese glaube ich mit solchen Nachrichten mit dem gleichen Wert kollidieren.

[Viktorii]

Zitat von sirius:

So, wie es jetzt geschrieben ist, brauchst du keine CriticalSections. Die dauern viel zu lange. Hier kommt man mit den Interlockedxxxx- Funktionen weiter. Wenn du diese bei schreibenden Vorgängen benutzt, brauchst du bei lesenden Vorgängen keine Absicherung.

Wenn ich das jetzt richtig verstanden habe müsste es dann so aussehen:

Delphi-Quellcode:

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

Aber bin ich damit wirklich auf der sicheren Seite?

Folgendes Szenario: InterlockedIncrement wird ausgeführt. In diesem Moment kann die Variable nicht von anderen Threads gelesen werden. Angenommen der Wert von FHead sei vor dem inkrementieren 10 und danach folglich 11. Diesen 'Überlauf' fange ich zwar mit der folgenden if-Abfrage ab, aber was passiert wenn ein anderer Thread den Wert ausliest bevor diese geschehen ist? Dann hat er den Wert 11, sollte aber 0 haben

[jaenicke]

Du solltest dir einmal anschauen was es da noch alles für Interlocked-Funktionen gibt.
Unter anderem InterlockedExchange, InterlockedAnd, InterlockedXor, ...
Siehe Dokumentation:
http://msdn.microsoft.com/en-us/library/ms683614.aspx

// EDIT:
Allerdings könnte ein anderer Thread trotzdem den Wert zwischen den beiden Operationen auslesen.

[sirius]

Zitat von Viktorii:

Folgendes Szenario: InterlockedIncrement wird ausgeführt. In diesem Moment kann die Variable nicht von anderen Threads gelesen werden. Angenommen der Wert von FHead sei vor dem inkrementieren 10 und danach folglich 11. Diesen 'Überlauf' fange ich zwar mit der folgenden if-Abfrage ab, aber was passiert wenn ein anderer Thread den Wert ausliest bevor diese geschehen ist? Dann hat er den Wert 11, sollte aber 0 haben

Zum Nullsetzen solltest du natürlich auch eine Interlockedxxx verwenden. Und das Problem mit der 10 und der elf kannst du ja vorher abfragen.

Delphi-Quellcode:

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