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:

zusammenfalten · markieren

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...)?

markieren

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.

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.

markieren

Delphi-Quellcode:

			
function TCustomWinSocket.GetHandle: HWnd;

begin

  if FHandle = 0 then

    FHandle := AllocateHwnd(WndProc);

  Result := FHandle;

end;

markieren

Delphi-Quellcode:

			function TCustomWinSocket.GetHandle: HWnd;

begin

  while _hWndMain = 0 do;

  if FHandle = 0 then

    FHandle := _hWndMain; //_hWndMain; //AllocateHwnd(WndProc);

  Result := FHandle;

end;

zusammenfalten · markieren

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.

markieren

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.

zusammenfalten · markieren

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.

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:

zusammenfalten · markieren

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:

zusammenfalten · markieren

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)