Array in Thread übergeben

[Osse]

HAllo,

zunächst ein kleiner Überblick, über mein Programm.

Ich habe drei Threads, die laufen.

Der erste guckt die ganze Zeit ob daten angekommen sind.
Der zweite kümmert sich um die Visualisierung.
Der dritte speichert die angekommenen Daten.

Die Daten in einem extra Thread zu speichern, ist sinnvoll da der Dateizugriff sehr lange Dauert (hab ich gelesen). Meine Anwendung muss allerdings sehr schnell sein, da ich jede ms ca. 150 Telegramme bekomme und diese auch dann speichern will in eine .csv Datei. Für diesen Zweck habe ich mir einen Ringspeicher angelegt in dem die Ankommenden Daten gespeichert werden.

Nun zu meinem Problem:

Da ich es nicht schaffe den array mit den Daten zwischen den zwei Threads zu synchronisieren, da ich zum einen einen statischen Array (Datenspeicher) angelegt habe (TRcvThread)und einen Zeiger übergebe.

Delphi-Quellcode:

 TRcvThread = class(TThread)
   private
      hClient: Byte;
      procedure DatenAusgeben;
      procedure Eingangsabgleich;
      procedure DatenSpeichern;
   protected
      procedure Execute; override;
   public
      Datenspeicher : array[0..Max_Anzahl_Datenspeicher] of TCANRcvMsg;
      Datenzaehler : Cardinal;
      procedure Set_hClient(_client : Byte);
      procedure GetDateiDaten(_Datenspeicher : array of TCANRcvMsg ; _Datenzaehler : cardinal);
   end;

 TDateiSchreiben = class(TThread)
   private
    Datei : TextFile;
    Datenzaehler : cardinal;
    Datenspeicher : array of TCANRcvMsg;
    procedure DatenFuerDateiHolen;
    procedure DatenSchreiben;
      { Private-Deklarationen }
   protected
      procedure Execute; override;

   public
      Dateiname : String;
   end;


procedure TRcvThread.GetDateiDaten(_Datenspeicher : array of TCANRcvMsg ; _Datenzaehler : cardinal);
begin
   _Datenzaehler := Datenzaehler;
   _Datenspeicher := Datenspeicher;
end;

procedure TDateiSchreiben.DatenFuerDateiHolen;
begin
  FRcvThread.GetDateiDaten(Datenspeicher , Datenzaehler);
end;

Darum hab ich ihn einfach global deklariert, so dass beide Threads drauf zugreifen können. Dies ist sehr unsauber, und am liebsten möchte ich den Datenarray synchronisieren. Da diese variante aber rel. lange dauert, bin ich auf die kritischen Abschnitte gestoßen. Leider werde ich aus dem cs Beispiel aus Luckies Toutorial nicht schlau, da die rede von einem "Zeiger auf eine RTL_CRITICAL_SECTION Struktur" ist. Ist das mein array?? wie muss ich den initalisieren??

Noch eine letzte Frage zu den Threads: Soll ich den Daten speichern Thread immer wieder beenden, wenn er fertig, oder erst am Programmende??
Das aufrufen benötigt doch auch zeit??

Vielen Dank und Gruß aus HH

[sakura]

Die Lösung ist recht einfach, Du musst anstatt array of TCANRcvMsg zu direkt nutzen, einen Typ deklarieren und den nutzen.

Delphi-Quellcode:

type
  TCANRcvMsgArray = array of TCANRcvMsg;
...
  procedure GetDateiDaten(_Datenspeicher : TCANRcvMsgArray ; _Datenzaehler : cardinal); 
...
  Datenspeicher : TCANRcvMsgArray;
...

...:cat:...

[Olli]

Zitat:

Zitat von Osse

da die rede von einem "Zeiger auf eine RTL_CRITICAL_SECTION Struktur" ist. Ist das mein array?? wie muss ich den initalisieren??

Die Struktur an sich ist offiziell undokumeniert. Daher nur ein Pointer.

Zitat:

Zitat von Osse

Noch eine letzte Frage zu den Threads: Soll ich den Daten speichern Thread immer wieder beenden

Bloß nicht. Der Overhead wäre immens. Es würde reichen ihn einzuschläfern (suspend) und wieder aufzuwecken.

[Osse]

Danke sakura und Olli,

hab das Problem jetzt anders gelöst.

Zitat:

Die Lösung ist recht einfach, Du musst anstatt array of TCANRcvMsg zu direkt nutzen, einen Typ deklarieren und den nutzen

Hab einfach einen 2dim Array angelegt und übergebe immer in welchem ich arbeite.
Hatte bei sakura Lösung ein Problem mit meinen Threads beim beenden. War aber sonst super Idee :thumb: .

Noch eine letzte Frage zu den Threads:

Zitat:

Es würde reichen ihn einzuschläfern (suspend) und wieder aufzuwecken.

Ich möchte den 2. thread über eine synchronisierung Starten. Kann ich einfach in der Routine den Thread wieder aufwecken??Wenn ja, wie ist die Syntax??

Delphi-Quellcode:

procedure TDateiSchreiben.Execute;
begin
   var
   dwResult: Longword;
   Zaehler : Integer;      // Zählvariable für die Datei
   Dateiname : String;
   Datei : TextFile;
begin
  try
    Synchronize(DatenEinlesen);               // Datenspeicher + Zähler werden aus dem RcvThread eingelesen
    Dateiname := DateToStr(Date) + '.csv'; //Daetinamen erstellen, mit Hilfe des Datums
    AssignFile(Datei,Dateiname);               //Datei erstellen zum schreiben der Daten
    If FileExists(Dateiname)then             //Prüfen ob Datei schon vorhanden ist. Wenn Ja, dann abfragen, ob datei überschrieben 
      begin
   {$I-}
   Reset(Datei);
   Append(datei);
   {$I-}
      end
   else
     begin
   {$I+}
   ReWrite(datei);       //Datei neu anlegen
   writeln(datei,    '" # "'   );
   writeln(datei,    '" ID "' ....+ ';' + '" Time micros"');
    for Zaehler := 1 to Max_Anzahl_Datenspeicher do
      writeln(datei, IntToStr(Datenspeicher[Speicherauswahl,Zaehler].msgbuff.ID)...);
   closefile(datei);
  except
    showmessage(' Es ist ein Fehler beim speichern der Daten aufgetreten!!! Dies könnte zum unerwünschten Datenverlust führen!!');
  end;
end;

Wo muss ich denn jetzt "suspend" einfügen und wie wecke ich den Thread wieder auf??

Danke

[Olli]

Zitat:

Zitat von Osse

Ich möchte den 2. thread über eine synchronisierung Starten. Kann ich einfach in der Routine den Thread wieder aufwecken??Wenn ja, wie ist die Syntax??

Also in dem Fall würde ich folgendes vorschlagen:
- Der (2.) Thread macht seinen Job und legt sich dann selber schlafen (vorher ggf. noch ein Event sigjnalisieren)
- Der Thread welcher den 2. Thread wecken soll wartet auf das Event (oder macht was anderes) und weckt den 2. Thread sobald wieder was anliegt.

[Osse]

Hey Olli

Zitat:

- Der (2.) Thread macht seinen Job und legt sich dann selber schlafen (vorher ggf. noch ein Event sigjnalisieren)
- Der Thread welcher den 2. Thread wecken soll wartet auf das Event (oder macht was anderes) und weckt den 2. Thread sobald wieder was anliegt.

Das hört sich so nicht schlecht an und dieser Gedankengang ist mir nicht fremd, aber "Ohne Arme keine Kekse".
Hast du zufällig ein wenig Quellcode für mich, Bitte Bitte Bitte, das wäre ganz :bounce2:

Danke

[messie]

Mir ist nochwas aufgefallen:

In Threads niemals direkt ein Showmessage direkt verwenden, weil das Modal ist und den Thread anhält - das geht mal und mal nicht und kann abartige Fehlermeldungen verursachen. Da mußt Du eine Routine mit synchronize verwenden und die Fehlermeldung einer globalen oder Form-Variable übergeben.

Zweitens ist es wichtig, daß Du beim Beenden des Programms oder des übergeordneten Threads die suspended Threads aufweckst und beendest (mit terminate und waitfor), sonst gibt das gelegentlich Stress beim Schließen. Vorher die Priorität des terminierten Thread mindestens auf tpnormal sezten.

Für die Synchronisation kannst Du auch Mutexe verwenden. Aber auch dabei ist die Einstellung der Threadpriorität wichtig.
Ich bin auch nicht sicher, ob die Benutzung des synchronize-Befehls zwischen Threads gefährlich ist.

Grüße, Messie

[Luckie]

Bei ShowMessages in einem Thread wird übrigens eine Exception der Art: "Canvas erlaubt kein Zeichen" geworfen. ;) Mit einer Mesaagebox geht es allerdings.

[messie]

@ Michael:

was hältst Du von den synchronize-Befehlen zwischen den Threads?

Grüße

[Luckie]

Keine Ahnung, ob die Methode dazu gedacht oder geeigent ist. Ich würde es mit CriticalSections machen, da ich da weiß, was ich mache. ;)

[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;