Anfängerfrage: Non Blocking TCP Client in einem Thread

[alzaimar]

Moin

Jahaaa, auch Moderatoren sind manchmal richtige Deppen. So wie ich.

Ich muss eine Anwendung fertigstellen, in der ein Thread mit einem TCP-Server kommuniziert. Ich möchte diese Änderungen minimalinvasiv gestalten, d.h. vorhandene Klassen nur austauschen und ganz wenig am Code verändern.

Folgender Code bereitet mir Bauchschmerzen:

Hier der Code des Kommunikationsthreads:

Delphi-Quellcode:

Type
  TComThread = Class (TThread)
  Private
     fSocket : TClientSocket;
  ...
  Public
     Procedure SendData (Const Data : String);
     Property OnDataReceived : TDataReceivedEvent...;
  End;

Procedure TComThread.SendData (Const Data : String);
Begin
  AddDataToOutputQueue(Data)
End;

Procedue TComThread.Execute;
  Procedure WaitForSocketConnected;
  Begin
    Repeat until Timeout or fSocketConnected;
  End;

Begin
  fSocket := TClientSocket.Create(nil);
  fSocket.ClientType := ctNonBlocking;
  fSocket.OnConnected := SocketConnected;
  fSocket.OnRead := ReadDataFromSocket;
  ...
  fSocket.Connect;
  Synchronize (WaitForSocketConnected); // <--- hä? klappt aber nur so

  While not Terminated And fSocketconnected Do
    If DataToSendAvailable Then SendNextChunkOfData;

  fSocket.Free;
End;

Der Thread verwaltet eine Queue, in der andere Threads ihre Daten reinpacken, die verschickt werden sollen.
Der Thread hat einen Event, der gefeuert wird, wenn Daten angekommen sind.

Das Teil hängt sich manchmal auf und ich vermute, das das an diesem komischen Synchronize liegt. Außerdem ist das irgendwie uncool.

Meine Frage lautet nun:
Wie macht man das richtig?
Wie verwendet man einen Non-Blocking Socket in einem Thread? Gibt es irgendwo Codebeispiele?

Oder gibt es bessere/einfachere Lösungen?

Wie gesagt, wichtig ist mir nur das Interface des Threads (SendData, OnDataReceived) und das das Senddata den Aufrufer nicht blockiert.

Da ich bestimmt den Wald vor lauter Bäumen nicht sehe, bin ich gerne bereit, mich auslachen zu lassen.
Oder mit Torten beschmeißen.

[Klaus01]

Hallo alzaimer,

ginge es nicht, wenn man den Thread im suspended Mode erstellt
und dann wenn der Socket sein ConnectEvent schickt den Thread resumed?

Dann natürlich die Einstellungen des Sockets
im Thread.create unterbringen.

So könnte man sich diese while (Time or fSocketConnected) Schleife sparen.

Grüße
Klaus

[wicht]

Wo wird denn fSocketConnected gesetzt? Und wie? In einem Event?

Ändert das hier vielleicht was (falls TClientSocket eine Eigenschaft Connected hat...)?

Code:

  fSocket := TClientSocket.Create(nil);
  fSocket.ClientType := ctNonBlocking;
  fSocket.OnConnected := SocketConnected;
  fSocket.OnRead := ReadDataFromSocket;
  ...
  repeat until fSocket.Connected;

[Astat]

Hallo alzaimar.

Du verwendest einen Asynchronen Socket, die Kommunikation wird mit dem Socket durch den Parameter (ctNonBlocking) auf WindowsMessages festgelegt.

Delphi-Quellcode:

WSAAsyncSelect(Socket, Wnd, Msg, AsyncStyle);

Das heisst, es werden die Socket Events in die Application Message Queue (MSQ), zur Abarbeitung, durch den Mainthread gestellt.
Die Verwendung der TClientSocket Komponente mit dem Parameter (ctNonBlocking) in einem eigenen Thread macht keinen Sinn,
da der Motor, der den Socket am Leben hällt, die MSQ des Mainthreads ist.
Wenn du nicht die Mainthread Message Oueue verwenden willst, musst Du die Komponente umschreiben.
Ersetze das AllocateHwnd der Komponente durch folgenden Code.

Delphi-Quellcode:

function TCustomWinSocket.GetHandle: HWnd;
begin
  if FHandle = 0 then
    FHandle := AllocateHwnd(WndProc);
  Result := FHandle;
end;

Delphi-Quellcode:

function TCustomWinSocket.GetHandle: HWnd;
begin

  while _hWndMain = 0 do;

  if FHandle = 0 then
    FHandle := _hWndMain; //_hWndMain; //AllocateHwnd(WndProc);

  Result := FHandle;
end;

Delphi-Quellcode:

function MainWndProc(wnd: HWND; Msg: Integer; wp: WPARAM;
  lp: LPARAM): Integer; stdcall; forward;

var
  _Terminated : Boolean = false;
  _hWndMain   : HWND = 0;
  _hThread    : THandle = 0;
  _dwThreadID : dword = 0;

  TASyncMainWindowClass : TWndClass = (style: 0; lpfnWndProc: @MainWndProc;
    cbClsExtra: 0; cbWndExtra: 0; hInstance: 0; hIcon: 0; hCursor: 0;
    hbrBackground: 0; lpszMenuName: nil; lpszClassName: 'TAsyncServerMainWindowClass'
    );

function MainWndProc(wnd: HWND; Msg: Integer; wp: WPARAM; lp: LPARAM): Integer; stdcall;
var
  tMsg: TMessage;
begin
  Result := 0;
  if Msg = WM_CLOSE then DestroyWindow(wnd) else
  if _CustomWinSocket <> nil then begin
    tMsg.Msg := Msg;
    tMsg.WParam   := wp;
    tMsg.LParam   := lp;
    tMsg.WParamLo := LOWORD(wp);
    tMsg.WParamHi := HIWORD(wp);
    tMsg.LParamLo := LOWORD(lp);;
    tMsg.LParamHi := HIWORD(lp);;
    if not _CustomWinSocket.WndProc(tMsg) then
      Result := DefWindowProc(wnd, Msg, wp, lp)
  end else
    Result := DefWindowProc(wnd, Msg, wp, lp);
end;

procedure InitAplication;
begin
  TASyncMainWindowClass.hInstance := hInstance;

  TASyncMainWindowClass.lpszClassName := PChar(CreateClassID);

  if Windows.RegisterClass(TASyncMainWindowClass) = 0 then
    raiselastwin32error;

  _hWndMain := CreateWindowEx(WS_EX_TOOLWINDOW   , TASyncMainWindowClass.lpszClassName,
    '', WS_POPUP   , 0, 0, 0, 0, 0, 0, hInstance, nil);

  if _hWndMain = 0 then raiselastwin32error;
end;

procedure CleanupAplication;
begin
  WSACleanup;
  CloseHandle(_hThread);
  if _hWndMain <> 0 then begin
    DestroyWindow(_hWndMain);
    _hWndMain := 0;
  end;
end;

function RunAplication(p: pointer): integer;
var
  MsgRec : TMsg;
begin
  result := 0;

  SetThreadPriority(_hThread, THREAD_PRIORITY_TIME_CRITICAL);

  WSAStartup($0101, WSAData);

  InitAplication;

  while GetMessage(MsgRec, _hWndMain, 0, 0) do begin
    DispatchMessage(MsgRec)
  end;

  CleanupAplication;
  _Terminated := TRUE;
end;

....

initialization
  _hThread := beginthread(nil, 0, RunAplication, nil, 0, _dwThreadID);
end.

Also, kurz zusammengefasst:

Am einfachsten ist es, wenn du den Socket und die durch die Socket Messages getriggerten Callbacks,
im Mainthread (MainUnit) erstellst. Du verwaltest deine zusätzlich notwendigen Workerthreads von
diesem Punkt aus. Von hier aus (MainUnit) kannst du deine Threads Resumen, Starten Stoppen, und oder
mit Synchgronisationsobjekten (Events, Mutexe, Semaphoren, Critical Sections usw.) arbeiten.

Mach die SendSocket methode Threadsave, dann kann jeder Workerthread zu jeder Zeit gefahlos senden.
Erstelle deine Workerthreads, implementiere wenn notwendig Datenübergabe Prozeduren usw.

Tip:
Um eine gute Skalierung zu erreichen, den Socket alleine im Mainthread werkeln lassen.
Zusätzliche Codeteile in eigene Threads auslagern.


lg. Astat

06810110811210410503210511511603209711003210010110 9032084097103
03211611111604403209711003210010110903210010510103 2108101116122
11610103209010110510810103206711110010103210511003 2068101108112
10410503210310111509910411410510109810111003211910 5114100046

[alzaimar]

Hallo Astat,

So in etwa hatte ich das befürchtet, jedoch insgeheim gehofft, das die MSQ des Threads verwendet wird, den Socket erstellt oder man dem Sockets wenigstens sagen kann, das sie die MSQ des Threads verwenden sollen.

Eigentlich habe ich eine Multi-client-Anwendung, d.h. viele Clients sollen sich mit jeweils einem Server verbinden und mit dem Auatschen.

Du meinst, ich soll einfach die TClientSockets im Hauptprogramm instantiieren und den Threads als Parameter übergeben?
Die Threads biegen dann die Events (OnConnected, OnRead usw. auf ihre eigenen Handler um und das klappt?O)

Das probiere ich mal...

[Astat]

alzaimar hat geschrieben

• So in etwa hatte ich das befürchtet, jedoch insgeheim gehofft, das die MSQ des Threads verwendet wird,
  den Socket erstellt oder man dem Sockets wenigstens sagen kann, das sie die MSQ des Threads verwenden sollen.

Nein, die Komponente verwendet die Main-MessageLoop und nicht die des Threads.
Dh. DispatchMessage wird im Mainthread aufgerufen, und die Messages, an die mit AllocateHwnd angegebene WndProc, des "Fensters" gesendet. Siehe vorherigen Post.

Delphi-Quellcode:

while GetMessage(MsgRec, _hWndMain, 0, 0) do begin
  DispatchMessage(MsgRec)
end;

Um in jedem Thread den du erzeugst, eine Messageloop zu implementieren, musst du folgendes ändern.

Die Globalen Variablen, und WndProc in die TClientSocket-Class integrieren.
Zu jedem neu erzeugten Client Objekt, MakeObjectInstance für die WndProc stdcall methode implementieren, und
einen eigenen Thread, der die Messageloop integriert (DispatchMessage) generieren.

alzaimar hat geschrieben

• Eigentlich habe ich eine Multi-client-Anwendung, d.h. viele Clients sollen sich mit jeweils einem Server
  verbinden und mit dem Quatschen.

Also, wenn es unbedingt multiple Asynchrone Clients sein müssen, die in mehreren Threads unter einem Prozess laufen müssen,
und sich zu ein und demselben Server verbinden, musst Du die Komponeten-Klasse (ScktComp.pas) wie oben, und in den vorhergehenden Thread beschrieben umcodieren.

Normalerweise kann man mit einem Asynchronen Client Socket, wie in der ScktComp.pas Unit implementiert ist,
ganz ordentliche Client Server Anwendungen erstellen. Jedoch haben die Borländer einige Bugs in der ScktComp.pas
reingebaut, desshalb ist das Teil auch nicht Hochlast fähig.

Folgendes musst du ändern.

Delphi-Quellcode:

procedure TCustomWinSocket.Error(const Socket: TCustomWinSocket; const ErrorEvent: TErrorEvent;
  var ErrorCode: Integer);
begin
   Disconnect(FSocket); //-- Astat
   if Assigned(FOnErrorEvent) then FOnErrorEvent(Self, Socket, ErrorEvent, ErrorCode);
end;

function TCustomWinSocket.SendBuf(var Buf; const Count: Integer; var ErrorCode: integer): Integer;
begin
  Result := SOCKET_ERROR; // =0; korrigiert //-- Astat
  ErrorCode := ERROR_SUCCESS; //-- Astat
  if not FConnected then Exit;
  Result := send(FSocket, Buf, Count, 0);
  if Result = SOCKET_ERROR then
  begin
    ErrorCode := WSAGetLastError;
    if (ErrorCode <> WSAEWOULDBLOCK) then
    begin
      Error(Self,eeSend,ErrorCode);
      Disconnect(FSocket);
      if ErrorCode <> 0 then
        raise ESocketError.CreateFmt(sWindowsSocketError,
          [SysErrorMessage(ErrorCode), ErrorCode, 'send']);
    end;
  end;
end;

Was ich derzeit noch nicht verstehe, warum du aus einem Prozess viele Asynchrone Clientverbindungen
zu ein und demselben Server benötigst?

Kann mir eigentlich kein Szenario vorstellen, wo das notwendig ist?

Normalerweise reicht eine Asynchrone Connection um den Server mit allem zu versorgen was dieser benötigt.
Ist ja deshalb ein Asynchroner Socket, in dem ich wie wild Daten schreiben kann, ohne auf ein Handshake wie bei
einer Synchronen Übertragung achten zu müssen.
Sollte es um einen Lasttest gehen, verwende einfach Blockierend Sockets, die du dann ohne Probleme
in multiplen Threads aufrufen kannst.

Delphi-Quellcode:

ClientSocket := Socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)

alzaimar hat geschrieben

• Du meinst, ich soll einfach die TClientSockets im Hauptprogramm instantiieren und den Threads als Parameter übergeben?
  Die Threads biegen dann die Events (OnConnected, OnRead usw. auf ihre eigenen Handler um und das klappt?O)

Nein, die Threads biegen da die Events nicht hin, die Events werden im Mainthread (Mainunit) getriggert.
Ich vertrete hier den Ansatz, dass nur eine Einzige asynchrone Verbindung zum Server besteht.
Wie geschrieben, verstehe ich noch nicht die Notwendigkeit multipler Asynchroner Clientverbindungen,
da man ja alles über eine einzige Asynchrone Verbindung abhandeln kann.
Ich vermute hier mal folgendes Szenario:

Der Server ist als "Blockierend" -->> Socket(AF_INET, SOCK_STREAM, IPPROTO_TCP) implementiert, also wie ein Webserver.
Connectet sich nun ein asynchroner Client, muss dieser das Handshake (Blockierent) nachbauen, ist natürlich möglich,
aber total overdozed. Hier verwendet man "blockierende Sockets" -->> Socket(AF_INET, SOCK_STREAM, IPPROTO_TCP),
dabei sendet man die Daten, und ruft anschließend Recv auf, dies reagiert erst wenn die Gegenstelle antwortet
(Server sendet Response zurück), fertig.

Also ich vermute mal, dass bei deinem Problem Blockierende Sockets die Lösung sein könnten?!


Würde mich interessieren, ob dem so ist?!

Wenn du Asynchrone und oder Synchrone Socket brauchst, sag einfach bescheid, hab da fix und fertige
DLL's (Synchron, Asynchron) Client und Server incl. Source, Hochlast Getestet.
Asynchrone Client Server Anwendung mit ~10000 Client Verbindungen.
Synchrone Client Server Anwendung mit 2 X Quad Xeon mit ca. 3000 Requests/s bei ~ 25Kb Daten in einem 1 GB Netz.

Hoffe etwas geholfen zu haben.

lg. Astat

[alzaimar]

Moin

Ich habe diverse Threads, die die Kommunikation mit jeweils einem Server implementieren und behandeln sollen: Pro Thread ein Server. Also:
1x Prozess, n x (Client[i] -> Server[i]). Eigentlich was stinknormales.

Die Threads verwenden eine TCP-Client-Klasse, die eine Schnittstelle ähnlich der des asynchronen TClientSockets besitzen (SendData, OnDataReceived). Ich bin davon ausgegangen, das das nonblocking sein muss. Wenn nicht, noch besser Hauptsache, ich bekomme etwas Stabiles hin, das o.g. Schnittstelle besitzt und sich aus einem Thread heraus bedienen lässt.

Mittlerweile habe ich ein Fenster (TForm), das eine Liste von TClientsockets verwaltet. Das Fenster brauch ich sowieso, um Statusinformationen der Sockets anzuzeigen.

Jeder TcpClient (ist ja keiner mehr, sondern nur eine Fassade als Thread) bekommt den Index 'seines' Sockets und kann nun über diesen Index Socket-Operationen ausführen. Dabei wird jedoch nicht der Socket direkt aus dem Thread heraus angefasst, sondern alles über Windows-Messages abgewickelt. Wenn ich was von dem Socket will, verpacke ich das in eine Struktur (Record-pointer), schicke die Message per PostMessage raus und kehre in den Thread zurück.

An so einer DLL wäre ich natürlich brennend interessiert, denn sie würde mir meine Arbeit vermutlich sehr erleichtern.

Ich danke Dir jetzt schon für deine Mühe, denn mir ist das entscheidende Licht -glaube ich- aufgegangen.

[sirius]

Du kannst den Client recht einfach in einen Thread legen, musst halt nur eine MEssageloop dort reinlegen. Z.Bsp: funktioniert folgendes:

Delphi-Quellcode:

uses windows, messages, classes, scktcomp, sysutils,
     syncobjs;


type
  TDataReceivedEvent=procedure(Sender:TObject; const Data:String) of object;

  TClient=class(TThread)
    Constructor Create(CreateSuspended: Boolean); reintroduce;
    Destructor Destroy; override;
   private
    FOnDataReceived: TDataReceivedEvent;
    FCLientSocket:TClientSocket;
    FEvent:TEvent;
    FThreadList:TThreadList;
    FRecvData:String;
    FErrorCode:Integer;
    procedure SetOnDataReceived(const Value: TDataReceivedEvent);
    procedure SocketConnected(Sender:TObject; Socket:TCustomWinSocket);
    procedure SocketError(Sender:TObject; Socket:TCustomWinSocket;
      ErrorEvent: TErrorEvent; var ErrorCode: Integer);
    procedure ReadDataFromSocket(Sender:TObject; Socket:TCustomWinSocket);
   protected
    procedure Execute; override;
    procedure DoOnDatareceived; virtual;
   public
    Procedure SendData(Const Data : String);
    Property OnDataReceived :TDataReceivedEvent read FOnDataReceived write SetOnDataReceived;
    procedure Terminate; reintroduce;
   end;

  TSendData=class
   private
    FData: String;
    procedure SetData(const Value: String);
   public
    property Data:String read FData write SetData;
  end;

implementation

{ TClient }

constructor TClient.Create(CreateSuspended: Boolean);
begin
  inherited;
  FEvent:=TEvent.Create(nil,false,false,'');
  FThreadList:=TThreadList.Create;
end;

destructor TClient.Destroy;
begin
  FEvent.Free;
  FThreadList.Free;
  inherited;
end;

procedure TClient.DoOnDatareceived;
begin
  if assigned(FOnDatareceived) then
    FOnDataReceived(self,FRecvData);
end;

procedure TClient.Execute;
var msg:Tmsg;
    eventhandle:THAndle;
begin
  FClientsocket:=TClientSocket.Create(nil);
  try
    FClientsocket.ClientType:=ctNonBlocking;
    FClientSocket.OnConnect:= SocketConnected;
    FClientSocket.OnRead := ReadDataFromSocket;
    FClientSocket.OnError := SocketError;
    //...
    FClientSocket.Address:='127.0.0.1';
    FClientSocket.Port:=21000;
    FClientSocket.Open;
    eventHandle:=FEvent.Handle;
    repeat
      case MsgWaitForMultipleObjects(1,eventhandle,false,infinite,QS_PostMessage) of
        WAIT_OBJECT_0: //Event fired
          if not terminated then
          begin
            with FThreadList.LockList do
            try
              if Count>0 then
              begin
                FClientSocket.Socket.SendText(
                 (TObject(Extract(First)) as TSendData).Data);
                if Count>0 then FEvent.SetEvent;
              end;
            finally
              FThreadList.UnlockList;
            end;
          end;
        WAIT_OBJECT_0+1: //Message
          while PeekMessage(msg,0,0,0,pm_Remove) do
           Dispatchmessage(msg);
        $FFFFFFFF: //Error
          raise Exception.Create(syserrormessage(getlasterror));
      end;
      if FErrorCode<>0 then  //asynchroner Error (aus Methode SocketError)
        raise Exception.Create(syserrormessage(FErrorCode));
    until terminated;
  finally
    FClientSocket.Free;
  end;
end;

procedure TClient.ReadDataFromSocket(Sender: TObject;
  Socket: TCustomWinSocket);
begin
  FRecvData:=Socket.ReceiveText;
  synchronize(DoOnDataReceived);
end;

procedure TClient.SendData(const Data: String);
var SendData:TSendData;
begin
  SendData:=TSendData.Create;
  SendData.Data:=Data;
  with FThreadList.LockList do
  try
    Add(SendData);
  finally
    FThreadList.UnlockList;
  end;
  FEvent.SetEvent;
end;

procedure TClient.SetOnDataReceived(const Value: TDataReceivedEvent);
begin
  FOnDataReceived := Value;
end;

procedure TClient.SocketConnected(Sender: TObject;
  Socket: TCustomWinSocket);
begin

end;

procedure TClient.SocketError(Sender: TObject; Socket: TCustomWinSocket;
  ErrorEvent: TErrorEvent; var ErrorCode: Integer);
begin
  FErrorCode:=ErrorCode;  //selber behandlen;
  ErrorCode:=0;
end;

procedure TClient.Terminate;
begin
  inherited;
  FEvent.SetEvent;
end;

{ TSendData }

procedure TSendData.SetData(const Value: String);
begin
  FData := Value;
end;

Bei mir allerdings nur mit dem MAinthread als Sender und Empfänger, aber senden kann hier auch jeder andere Thread:

Delphi-Quellcode:

type
  TForm1 = class(TForm)
    Button1: TButton;
    Edit1: TEdit;
    Memo1: TMemo;
    procedure FormCreate(Sender: TObject);
    procedure Button1Click(Sender: TObject);
    procedure FormDestroy(Sender: TObject);
  private
    { Private-Deklarationen }
    FClient:TClient;
    procedure ClientData(Sender:TObject; const Data:String);
    procedure ClientTerminate(Sender:TObject);
  public
    { Public-Deklarationen }
  end;

var
  Form1: TForm1;

implementation

{$R *.dfm}

procedure TForm1.FormCreate(Sender: TObject);
begin
  FClient:=TClient.Create(true);
  FClient.OnDataReceived:=ClientData;
  FClient.OnTerminate:=ClientTerminate;
  FClient.Resume;
end;

procedure TForm1.ClientData(Sender: TObject; const Data: String);
begin
  memo1.lines.add(Data);
end;

procedure TForm1.Button1Click(Sender: TObject);
begin
  FClient.SendData(edit1.text);
end;

procedure TForm1.FormDestroy(Sender: TObject);
begin
  FClient.Terminate;
  FClient.WaitFor;
  FClient.Free;
end;

procedure TForm1.ClientTerminate(Sender: TObject);
begin
  if assigned((Sender as TThread).FatalException) then
  begin
    application.ShowException(
      TThread(Sender).FatalException as Exception);
    close;
  end;
end;

[alzaimar]

So ähnlich war/ist das hier auch.. Wenn ich ein kleines Testprogramm schreibe, und den TCP-Thread aus der Hauptanwendung heraus instantiiere, klappt das. In der Anwendung selbst wird der TCP-Thread von einem Protokollhandler(ein Thread) heraus instantiiert, und da scheint es dann zu klemmen: Die Anwendung friert nach einiger Zeit einfach ein. Zuerst die GUI, aber die Kommunikation selbst funktioniert noch eine Weile, bis dann selbst das nicht mehr geht und der Prozess auf 100% CPU-Last geht. Ich vermute, das eine MSQ voll ist, denn WO es hängt, sehe ich einfach nicht.

Mittlerweile, auch um andere Fehlerursachen auszuschließen, habe ich eine Testanwendung, die einen nur Protokollhandler instantiiert. Das ganze Gedöns drumherum (Datenbank usw) habe ich rausgeschnippelt: Das Protokoll wird abgearbeitet und alles geloggt. Sonst nix... Trotzdem. Die GUI wird immer klebriger...

[sirius]

Wenn der ClientThread nicht aus dem Mainthread gestartet wird, solltest du jegliches Synchronize vermeiden. Ansonsten sollte obiger Code immer funktionieren (wie gesagt, außer das synchronize beim Empfangen (und OnTerminate), da musst du dir etwas eigenes überlegen, das hängt aber von deinem Protokollhandler ab)