IdUDPServer hat manchmal Datenverlust --> wie macht das Wireshark?

[Poolspieler]

Hallo zusammen,
ich benutze die Indy-Komponenten für einen UDP-Server.
Im Prinzip funktioniert das ganze recht gut. Leider ist aber bei hoher Last (sowohl hohe CPU-Last, als auch viele UDP-Pakete innerhalb kurzer Zeit) immer wieder ein Datenverlust vorhanden. Es landen anscheinend nicht alle UDP-Pakete im Puffer meines UDP-Servers.
Ein Pufferüberlauf ist nicht vorhanden - das habe ich bereits überprüft.
Wenn ich mit dem Wireshark mitlogge, dann kommen die Pakete beim Rechner definitiv an. Nur der UDP-Server hat sie nicht alle im Puffer :(
OK, ich stelle mit den Daten recht viel an:
- Realtimeausgabe der Daten in einem Chart (im Chart (TChart) sind bis zu 500.000 Punkte, die aktualisiert werden müssen)
- gleichzeitig werden die Daten in Textfiles mitgeschrieben

Ich habe den UDP-Server in einem eigenen Thread laufen - und synchronisieren dann die Daten (wenn sie komplett sind) zur Oberfläche. Und es scheint, dass die Datenverluste genau zu diesem Zeitpunkt auftreten - genau kann ich das aber nicht sagen...

Es scheint aber mit der Ausgabe im Chart sehr stark zu tun zu haben. Wenn ich anstatt von TFastlineseries z.B. TLineseries verwende, dann sind die Datenverluste deutlich größer!

Ich weiß schon, dass UDP nicht verbindungsorientiert ist - und es keine Flusskontrolle (ausser IP) gib. Und dass es deshalb auch Datenverluste geben kann.
ABER: WIE macht das der Wireshark (früher Ethereal) --> Der kann Millionen von Paketen mitloggen, parsen und darstellen - OHNE dass auch nur ein einziges Paket verloren geht.

Hat jemand einen Hinweis für mich? Ich würde mich freuen.



Viele Grüße,

Poolspieler

[Zacherl]

Du weißt aber schon, dass bei UDP generell nicht garantiert ist, dass ein Paket ankommt? Nichtmal die korrekte Reihenfolge der Pakete muss zwangsweise eingehalten werden. Wenn du aus irgendeinem Grund auf keinen Fall TCP verwenden willst, musst du dich um Sequenzierung und eventuelles Neu-Anfordern von verlorengegangenen Paketen selbst softwareseitig kümmern.

[Bernhard Geyer]

Zitat:

Zitat von Poolspieler (Beitrag 1288640)

Ich habe den UDP-Server in einem eigenen Thread laufen - und synchronisieren dann die Daten (wenn sie komplett sind) zur Oberfläche. Und es scheint, dass die Datenverluste genau zu diesem Zeitpunkt auftreten - genau kann ich das aber nicht sagen...

Da hast du doch schon den Implementierungsfehler. Wenn du synchronisierst wird den Thread angehalten und die Abarbeitung im Hauptthread weiter gemacht. Während dieser Zeit gehen natürlich alle ankommenden Daten verloren da ja keiner (Dein Thread ist ja durch den Hauptthread blockiert) die Daten annimmt.

Du musst also die Übergabe an die Darstellungsthread besser von deinem Empfangsthread entkoppeln.
Da ich hier wenig Erfahrung habe (bzw. sehr lange her) hier ein (evtl. fehlerhafter) Ansatz:

Statt normal zu synchronisieren triggerst du (z.B. über PostMessage) den Hauptthread. Der holt sich die Daten dann aus einem threadsicheren Datenbereich (das Abholen darf natürlich nicht zu blockieren deines UDP-Threads führen). evtl. einfach nur für jedes vollständige Datenpacket einen "Aufhänger" einen simple TThreadList übergeben.

[Sir Rufo]

Der Wireshark hängt sich gaaanz tief in die Netzwerkkarte (Treiber) rein und er bekommt die Daten quasi bevor die an das System weitergereicht werden.

Um die Verluste zu minimieren solltest du statt

Delphi-Quellcode:

Synchronize

eher

Delphi-Quellcode:

Queue

versuchen. Das kann dann aber auch die Windows-Warteschlange zum Überlaufen bringen.

Ansonsten bleibt dir noch die Daten vom Epfangsthread in einen weiteren Thread zu übergeben, der sich dann um die Weitergabe kümmert. Um die Daten aufzunehmen hat dieser Thread 2 Queues. Eine zum Empfangen und die andere zum Weiterreichen.

Ist die Weiterreich-Queue leer, dann tauscht man die Plätze und die alte Empfangsqueue wird nun die Weiterreichqueue und umgekehrt. Der Tausch geht sehr fix und lässt sich per

Delphi-Quellcode:

TInterlocked.Exchange

threadsicher durchführen. Ein weiteres Locking ist dann nicht erforderlich (wenn es nur einen Empfangsthread gibt).

[Poolspieler]

Hallo Zusammen,
danke für Eure Antworten.

@Zacherl: Ja, das weiß ich. Leider kann ich Systembedingt (und Kundenanforderung) nicht auf TCP umsteigen. Ein zweites Anfordern der Daten ist auch nicht erwünscht bzw. möglich.
Die Sache mit der Reihenfolge tritt IMHO nur auf, wenn Router bzw. mehrere Hops zwischen den Hosts sind. In meinem Fall ist das ausgeschlossen. In den meisten Fällen ist zwischen dem UDP-Server und den Hosts nur ein Switch. Trotzdem danke für Deinen Hinweis!

@Bernhard:
Ich denke in der gleichen Richtung wie Du. Nur bisher hatte ich keinen guten Lösungsansatz, der auch stabil und robust läuft. Bisher habe ich versucht, die Sychnronisation so performant wie möglich zu machen... Die ist natürlich nur ein Workaround...
Das Problem mit dem Hauptthread, der sich die Daten aus einem Threadsicheren Datenbereich holt, ist doch, dass der Server-Thread nicht weiß, wann er die Daten freigeben kann - oder wie geht das?
Gibt es vielleicht ein Beispiel, oder weitere Info dazu im Netz?

@Sir Rufo:
Danke für deinen konkreten Lösungsansatz! Das klingt richtig gut! Ich werde das weiter verfolgen und ggf. später noch konkrete Fragen dazu stellen. :-D
Hast Du eventuell einen Link zu einem Demoprojekt oder weiterer Info?
Falls nicht, kein Problem, dann Google ich gleich selbst...?



Viele Grüße,

Poolspieler

[Sir Rufo]

Ob Google das hat weiß ich nicht, ist mir gerade so eingefallen :stupid:

[jaenicke]

Eine Möglichkeit ist auch ein Ringpuffer.
Kurz gesagt: Da muss der Empfangsthread die meiste Zeit über nur an die durch einen Pointer bezeichnete Stelle zu schreiben und den Pointer inkrementieren (AtomicIncrement). Am Ende des Puffers angekommen, gehts zurück zum Anfang. Und der verarbeitende Thread kann dann einfach lesend darauf zugreifen.

[Poolspieler]

Hallo jaenicke,
das ist auch ein guter Ansatz.
Aber woher weiß der Empfangsthread, ob der verarbeitende Thread die Daten schon abgeholt hat? Stichwort "Pufferüberlauf"...
Das aktualisieren der Pointer muss aber trotzdem synchronisiert erfolgen oder?
Oder sind Pointer threadsave?

Gruß,

Poolspieler

[Sir Rufo]

Hier mal so ein grober Entwurf, der aber schon tut :)

UPDATE
Eine kurze Erklärung

Der

Delphi-Quellcode:

TMessageQueueThread<T>

ist selber nicht threadsafe und nur dazu gedacht von einem einzigen Thread gefüttert zu werden. Sollen die Nachrichten von mehreren Threads zusammengeführt werden, so wird zusätzlich noch eine weitere threadsafe Queue benötigt, die dann von den

Delphi-Quellcode:

TMessageQueueThread

gefüttert wird, jetzt aber entkoppelt vom eigentlichen Tread, der die Nachrichten produziert.

Jede Art der Synchronisierung würde den eigentlichen Thread wieder ausbremsen und das gilt es hier ja ausdrücklich zu vermeiden.

Delphi-Quellcode:

unit Unit1;

interface

uses
  System.SysUtils,
  System.Classes,
  System.SyncObjs,
  System.Generics.Collections;

type
  TMessageQueueThread<T> = class( TThread )
  private
    FQueueProc: TProc<T>;
    FInQueue, FOutQueue: TQueue<T>;
    FEvent: TEvent;
  protected
    procedure Execute; override;
    procedure DoProcessQueue( AQueue: TQueue<T> );
    procedure TerminatedSet; override;
  public
    constructor Create( AQueueProc: TProc<T> );
    destructor Destroy; override;

    procedure Enqueue( const AItem: T );
  end;

  TReceiveThread = class( TThread )
  private
    FMessageQueue: TMessageQueueThread<string>;
    FFMTset: TFormatSettings;
  protected
    procedure Execute; override;
  public
    constructor Create( AProc: TProc<string> );
    destructor Destroy; override;
  end;

implementation

{ TRelayQueueThread<T> }

constructor TMessageQueueThread<T>.Create( AQueueProc: TProc<T> );
begin
  inherited Create( False );
  FEvent := TEvent.Create( nil, False, False, '' );
  FQueueProc := AQueueProc;
  FInQueue := TQueue<T>.Create;
  FOutQueue := TQueue<T>.Create;
end;

destructor TMessageQueueThread<T>.Destroy;
begin
  inherited;
  FInQueue.Free;
  FOutQueue.Free;
  FEvent.Free;
end;

procedure TMessageQueueThread<T>.DoProcessQueue( AQueue: TQueue<T> );
begin
  while AQueue.Count > 0 do
    begin
      FQueueProc( AQueue.Peek );
      AQueue.Dequeue;
    end;
end;

procedure TMessageQueueThread<T>.Enqueue( const AItem: T );
begin
  FInQueue.Enqueue( AItem );
  FEvent.SetEvent;
end;

procedure TMessageQueueThread<T>.Execute;
begin
  inherited;
  while not Terminated do
    begin
      FEvent.WaitFor( );

      if Terminated
      then
        Exit;

      // Queues tauschen
      FOutQueue := TInterlocked.Exchange < TQueue < T >> ( FInQueue, FOutQueue );
      // Queue verarbeiten
      DoProcessQueue( FOutQueue );
      // Queue leeren, nur für alle Fälle, sollte ja eh jetzt leer sein :o)
      FOutQueue.Clear;
    end;
end;

procedure TMessageQueueThread<T>.TerminatedSet;
begin
  inherited;
  FEvent.SetEvent;
end;

{ TReceiveThread }

constructor TReceiveThread.Create( AProc: TProc<string> );
begin
  inherited Create( False );
  FFMTset := TFormatSettings.Create( '' );
  FMessageQueue := TMessageQueueThread<string>.Create( AProc );
end;

destructor TReceiveThread.Destroy;
begin
  inherited;
  FMessageQueue.Free;
end;

procedure TReceiveThread.Execute;
var
  LMsg: string;
begin
  inherited;
  while not Terminated do
    begin
      LMsg := FormatDateTime( 'hh:nn:ss.zzz', Now, FFMTset );
      FMessageQueue.Enqueue( LMsg );
    end;
end;

end.

Der ReceiveThread ist jetzt nur zum Testen des MessageQueueThread da

Und der Test mit der Synchronisierung

Delphi-Quellcode:

unit Form.Main;

interface

uses
  Unit1,

  Winapi.Windows, Winapi.Messages, System.SysUtils, System.Variants, System.Classes, Vcl.Graphics,
  Vcl.Controls, Vcl.Forms, Vcl.Dialogs, Vcl.StdCtrls;

type
  TForm1 = class( TForm )
    Button1: TButton;
    ListBox1: TListBox;
    procedure Button1Click( Sender: TObject );
  private
    FTestThread: TReceiveThread;
    procedure ReceiveMessage( Arg: string );
  public
    { Public-Deklarationen }
  end;

var
  Form1: TForm1;

implementation

{$R *.dfm}

procedure TForm1.Button1Click( Sender: TObject );
begin
  if Assigned( FTestThread )
  then
    FreeAndNil( FTestThread )
  else
    FTestThread := TReceiveThread.Create( ReceiveMessage );
end;

procedure TForm1.ReceiveMessage( Arg: string );
begin
  TThread.Synchronize( nil,
      procedure
    begin
      ListBox1.Items.Add( Arg + ' - ' + FormatDateTime( 'hh:nn:ss.zzz', Now ) );
    end );
end;

end.

Und nach dem Druck auf den Button, relativ schnell wieder den Button drücken, der Thread haut dir gnadenlos die ListBox voll ;)

[OlafSt]

Ich habe so ein ähnliches Problem vor geraumer Zeit gehabt. Als erstes habe ich die Indy-UDP-Komponenten entsorgt und eine Dritt-Implementation von UDP eingesetzt (UdpSockUtil, zu finden auf entwickler-ecke.de).

Im Empfängerthread wird dann eine TQueue mit den Empfangsdaten gefüllt. Nach jedem Empfang wird mit Postmessage der Hauptthread verständigt. Dieser wiederum liest dann solange Daten aus der Queue, bis diese leer ist.

Ich habe die Synchronisation manuell mit TCriticalSections gemacht, so das ich exakte Kontrolle darüber habe, wann wer wie wo was liest oder schreibt - vor allem blockiere ich so die Queue nur für den einzelnen Lese/Schreibvorgang und gebe sie sofort wieder "frei".

Falls die Windows-Messagequeue überzulaufen beginnt, kann man die Postmessages einfach auf jeden zweiten/dritten Eintrag begrenzen. Der Hauptthread arbeitet dann eh alles ab.