Array in Thread übergeben

[Osse]

Hallo, ich hab noch mal ne Frage zur Übergabe von Variablen zwischen 2 Threads. Ich habe es mit Synchronize gemacht, und ihr habt mir geraten es mit CriticalSections zu machen. Habe in der Delphi Hilfe nachgelesen:

Zitat:

Mit TCriticalSection können Operationen abgesichert werden, die nicht durch den Start eines anderen Threads unterbrochen werden dürfen. Kritische Abschnitt arbeiten wie Gates, die jeweils nur einen einzelnen Thread durchlassen. Da die Ausführung aller anderen Threads blockiert wird, verringert sich die Leistung der Anwendung erheblich, wenn kritische Abschnitte zu häufig eingesetzt werden.

Kritische Abschnitte müssen einen globalen Gültigkeitsbereich besitzen, damit sie für alle Threads zur Verfügung stehen. In jedem Thread sollten Aktionen, die die Ausführung der durch TCriticalSection geschützten Operationen stören könnten, nur nach dem Aufruf der Methode Acquire oder Enter stattfinden. Threads, die für wichtige Operationen TCriticalSection nicht verwenden, können Fehler verursachen.

Mich stört daran, folgender Teilsatz: "verringert sich die Leistung der Anwendung erheblich, wenn kritische Abschnitte zu häufig eingesetzt werden."

Ich habe in meiner Anwendung 2 Threads, welche die ganze Zeit laufen. Der eine ließt Daten ein und der Andere schreibt diese anschließend in eine Datei, somit habe ich nie ein gleichzeitiges schreiben auf die Daten. Ich bekomme so ca. 130Telegramme pro ms, deshalb meine Frage ob ich CriticalSections benutzen soll. Ich will nicht, dass ein Thread den anderen sperrt, da dieser schön die Daten weiter einlesen soll (sonst könnte ich mir das ganze auch schenken mit 2Threads).

Zu Synchronize steht folgendes in der Hilfe:

Zitat:

Beschreibung

Synchronize löst den Aufruf einer bestimmten Methode aus, die vom VCL-Haupt-Thread ausgeführt werden soll. Durch dieses indirekte Verfahren werden Konflikte in Multithread-Anwendungen vermieden. Wenn Sie nicht sicher sind, ob ein Methodenaufruf Thread-sicher ist, rufen Sie die Methode vom VCL-Haupt-Thread aus auf, indem Sie sie an die Methode Synchronize übergeben.

Der Thread wird unterbrochen, während die angegebene Methode ausgeführt wird.

Hinweis

Unsichere Methoden können Sie auch durch kritische Sektionen oder mit Hilfe eines Synchronisierers, der mehrfaches Lesen, aber nur exklusives Schreiben zuläßt, schützen.

Hieran verstehe ich nicht, was sind denn "Unsichere Methoden"?? :wiejetzt:

[messie]

Woher weißt Du denn, daß auf Deine Daten niemals von beiden Threads zugegriffen wird.
Wie Du schon sagst, hast Du zwei Threads, damit Du sie unabhängig voneinander laufen lassen kannst. Also kannst Du auch nicht garantieren, das sie das nicht tun. Vielleicht erst nach ein paar Stunden oder Tagen Laufzeit....

Also mußt Du den Speicher, auf den beide Threads zugreifen, so verriegeln, daß nur ein Thread drankommt. Dies beschränkt sich ja nur auf die Datenübergabe. Ich würde die Daten zum Speichern in einen Puffer übergeben und dann den Thread in Ruhe schreiben lassen. Die Verriegelungszeit dafür dürfte im Microsekundenbereich liegen, das sollte bei 130 Hz Daten gut ausreichen.

Zur Performance: die Methode synchronize hält den gesamten Thread an, der muß dann auch wieder gestartet werden. Ich gehe davon aus, daß diese Methode wesentlich mehr Overhead-Operationen verursacht als der Schutz des Übergabe-Speicherbereichs mit waitforsingleobject.

Grüße, Messie

[Osse]

Danke,

Zitat:

daß auf Deine Daten niemals von beiden Threads zugegriffen wird

Weiß ich natürlich Nicht :pale:. Habe zwei eingangspuffer, die ich abwechselnd beschreibe. Da ich diese sehr groß gewählt habe gehe ich davon aus(weiß es aber nicht), dass nicht gleichzeitig auf diese zugegriffen wird.

Danke, werde es mit TCriticalSection machen. War nur so ne Frage, die mich mal stutzig gemacht hat. :cyclops:

[alzaimar]

Hallo, Osse: Immer noch am hacken...?

Eine "unsichere Methode" ist in meinen Augen eine Methode, die von mehreren Threads aufgerufen werden kann, und die nicht durch Synchronisationsmassnahmen geschützt ist.

Z.B. ist die Methode 'AddOne' (als V:=V+1) erstmal unsicher. wenn 2 Threads die gleichzeitig aufrufen, dann ist der wert nicht etwa um 2 erhöht, sondern vielleicht nur um eins. Weil sie eben nicht geschützt ist.

So, wie ich das sehe, benötigst Du sowas wie eine Queue. Vorne stopfst Du per Thread #1 etwas herein und mit Thread #2 holst Du 'hinten' etwas heraus. Dabei soll:
-Thread #2 schneller sein als Thread #1
-Thread #2 'einschlafen', wenn der Buffer leer ist.

Damit Du das Optimal hinbekommst, verwendest Du einen Ringbuffer (einfach ein grosses Array). Der hat 2 Indizes (Head und Tail).
Wenn ich was reinstopfe (in den 'Head'), erhöht sich der um 1, aber nur, wenn dadurch der Schwanz nicht überschrieben wird. Wenn doch, ist Forderung (1) verletzt und ich muss ausnahmsweise warten, was ein GAU ist, aber was solls.

Wenn ich was raushole (vom 'Tail'), geht das nur, wenn was drin ist (logisch). Dann wird der Tail um eins erhöht. Wenn Tail=Head, ist der Puffer leer.

So, um das jetzt threadtechnisch umzusetzen und die Threads fein warten/schlafen zu lassen, machen wir Folgendes:
Thread #2 wartet auf ein 'signal' vom Ringbuffer, das daten drin sind.
Das geht einem Event. Ein Event kannst du separat an- und wieder ausschalten.

Der Ringbuffer schaltet das Event 'an', wenn Daten eingefügt wurden.
Weiterhin schaltet er das Signal wieder aus, wenn alle Daten abgeholt wurden.

Wie lässt man nun Thread#2 elegant warten? Mit 'WaitForSingleObject' (Schau mal in der Hilfe).

Der Peseudocode für die Execute Methode des Thread#2 wäre also ungefähr so:

Delphi-Quellcode:

Procedure TWritingThread.Execute;
Begin
  While not Terminated Do Begin
    If WaitForSingleObject (fRingBufferSignal, INFINITE) = WAIT_OBJECT_0 then
      WriteDataFromRingBufferToFile;
End;

Solange also fRingBufferSignal an ist, werden daten aus dem Buffer geholt und gespeichert. Ansonsten wartet der Thread. Laut Windows Hilfe mit sehr geringem Overhead.

Wenn das soweit klappt, kannst Du dir noch überlegen, ob Du jede CAN-Message in den Buffer stopfst, oder gleich einen Block von (sagen wir) 1000.

Nachtrag: Mit deinen alternierenden Buffern geht es natürlich auch (vermutlich sogar noch schneller). Dann signalisiert Thread #1, wenn ein Buffer voll ist. Thread #2 setzt das Signal zurück, schreibt den Buffer und geht wieder ins Bett (mit WaitForSingleObject). Kann sein, das Thread #1 in der zwischenzeit den anderen Buffer gefüllt hat. Na dann wird's für #2 nichts mit dem Nickerchen und er darf gleich an die Arbeit.

Im Grunde genommen ist das das Gleiche wie ein Ringbuffer mit 2 Elementen und ziemlich grossen Elementen. Zur Sicherheit würde ich aber mehr als 2 alternierende Buffer nehmen.
Wenn nämlich Thread #2 mit Buffer A noch nicht fertig ist, aber Thread #1 in der Zwischenzeit Buffer B gefüllt hat, dann fängt #1 gleich an Buffer A zu überschreiben...

Wenn ich mir's recht überlege, solltest Du doch einen Ringbuffer mit großen Blocken nehmen. dann hast Du Luft.

[Osse]

Hey alzaimer,

habe im Moment recht wenig Zeit zum hacken.

Hab mich mal gefragt, wie ich die beste Übergabe zwischen denbeiden Threads mache. Wollte immer so 1000 telegramme schreiben, da die Datei zugriffszeit recht "lang" ist.

Dachte nur das mit 2 Arrays zu machen, und die Daten dann immer schön in einer CriticalSection in die Datei zu schreiben, um völlig sicher zu sein. Außerdem wird der Array doch gesperrt, wenn er von dem einen Thread in der CriticalSection benutzt wird, oder hab ich da mal wieder ein Denkfehler :gruebel: ??

Das würde ja bedeuten dass der 1. Thread die ganze Zeit keine Daten in meinen Ringspecher schreiben kann, während der 2. Thread die Daten in die Datei schreibt.

Aber das ist ja genau das, was ich möchte, der 1. Thread ließt die Daten und der 2. schreibt parallel diese in eine Datei.

Im moment hab ich das mit schlafen legen(Suspend) und wieder aufwecken (Resume) realisiert. Werde es aber mal mit dem Event versuchen.

Delphi-Quellcode:

Procedure TWritingThread.Execute; 
Begin 
  While not Terminated Do Begin 
    If WaitForSingleObject (fRingBufferSignal, INFINITE) = WAIT_OBJECT_0 then 
      WriteDataFromRingBufferToFile; 
End;

fRingBufferSignal ist ein Event, und welches ich mit fRingBufferSignal.Setevent aufrufe, oder??

[messie]

suspend und resume sind sehr riskant! Du weißt nicht, an welcher Stelle der Thread schlafen gelegt wird und die vom Thread gerade in Anspruch genommenen Speicherbereiche werden nicht freigegeben. Kann also sein, daß Du dann mit verschiedenen Zugriffsarten nicht an den Speicher rankommst.
Dann lieber die Eventsteuerung

Grüße, Messie

[alzaimar]

@messie: Das mit dem 'schlafen' legen war nicht so gemeint (Suspend/Resume), sondern im übertragenen Sinne per WaitForSingleObject. Suspend/Resume benutzte ich nur beim Create (Suspended)...Initialisierung...Resume.
Ansonsten benutze ich WaitForSingleObject.

@Ossi: Ich meine, das die beiden Threads voll parallel laufen werden. T1 schreibt in B1 während T2 von B2 liest und umgekehrt.
Hier ist mal so ein Buffer. Put Put schiebst du was rein (von Thread #1) und kannst gleichzeitig von Thread #2 pber 'Get' was rauslesen.

'Get' wartet, bis was im Puffer ist.
'Put' wartet, bis der Puffer nicht mehr voll ist.

Ich verwende 2 Events, eins, um zu signalisieren, das der Buffer voll ist, und eins für 'ist leer'.
Die Modifikation des Ringbuffers ist durch eine CriticalSection geschützt.

Ich würde mit einer Puffergröße von >=3 arbeiten. Als 'Items' nimmst Du Deine 1000er Blöcke.

Viel Spass

Delphi-Quellcode:

unit csBuffer;

interface
uses SysUtils, Classes, windows, SyncObjs;
Type
(* Implementierung eines einfachen Ringbuffers mit einem Event.
 * "Put" wartet, bis der Buffer nicht mehr voll ist und schreibt dann ein
 * Element in den Buffer.
 * "Get" wartet, bis etwas im Buffer ist und liefert das älteste Element
 *)
  TRingBuffer = Class (TObject)
  Private
    FItems : Array Of Pointer;
    FTotal, FSize,
    FHead, FTail : Integer;
    FCS : TCriticalSection;
    FIsFullEvent, FIsEmptyEvent : TEvent;
  Public
    Constructor Create (aTotalSize : Integer);
    Destructor Destroy;
    Procedure Put (aItem : Pointer);
    Procedure Get (Var aItem : Pointer);
    End;

implementation

{ TRingBuffer }

constructor TRingBuffer.Create(aTotalSize: Integer);
begin
  Inherited Create;
  FSize := aTotalSize;
  SetLength (fItems, aTotalSize);
  FHead := 0;
  FTotal := 0;
  FTail := 0;
  fCS := TCriticalSection.Create;
  FIsFullEvent := TEvent.Create(nil,True,True,'');
  FIsEmptyEvent := TEvent.Create(nil,True,False,'');
end;

destructor TRingBuffer.Destroy;
begin
  SetLength (fItems,0);
  fCS.Free;
  FIsEmptyEvent.Free;
  FIsFullEvent.Free;
  Inherited;
end;

procedure TRingBuffer.Put(aItem: Pointer);
Var
  lIsEmpty : Boolean;
  NewHead : Integer;
begin
// Warten, bis der Buffer nicht mehr voll ist
  if FIsFullEvent.WaitFor (INFINITE) = wrSignaled Then Begin
    fCS.Enter;
    Try
      FItems [FHead] := aItem;                        // Element vorne anhängen
      FHead := (FHead + 1) mod FSize;           // Vorne um eins nach vorne ;-)
      If FTotal = 0 Then                      // Wenn der Buffer leer war, dann
        FIsEmptyEvent.SetEvent;                  // isser jetzt nicht mehr leer
      Inc (FTotal);
    Finally
      fCS.Leave;
      End
    End
  Else Raise Exception.Create ('Systemfehler');
end;

procedure TRingBuffer.Get(var aItem: Pointer);
begin
// Warten, bis der Buffer nicht leer ist
  if FIsEmptyEvent.WaitFor(INFINITE) = wrSignaled Then Begin
    fCS.Enter;
    Try
      aItem := FItems [FTail];                       // Letztes Element abholen
      FTail := (FTail + 1) mod FSize;
      if FTotal = 1 Then          // Wenn es leer wird, dann Event zurücksetzen
        FIsEmptyEvent.ResetEvent
      Else If FTotal = FSize  Then       // Wenn es voll war, dann Event setzen
        FIsFullEvent.SetEvent;
      Dec (FTotal);
    Finally
      fCS.Leave;
      End
    End
  Else Raise Exception.Create ('Systemfehler');
end;

end.

Und so arbeiten die threads:

Delphi-Quellcode:

While not Terminated do Begin
  fBuffer.Get (aData);
  SaveToFile (aData);
  End;

und der Andere:

Delphi-Quellcode:

While not Terminated do Begin
  GetCANData (aData)
  fBuffer.Put (aData);
  End;