Array in Thread übergeben

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