Klassendesign: Blockierender Constructor?

[Zacherl]

Zitat von Uwe Raabe:

Wie wäre es denn mit einem Event-gesteuerten Ansatz in der Art Socket.OnDataAvailable := MyHandleDataEvent ?

Das ginge ja in die selbe Richtung wie mein alter HandleData Ansatz. Da aber ja nicht jede Socket Komponente so ein Event besitzt (vor allem nicht mit der selben Parameter-Definition), sollte der Benutzer damals selbst auf ein OnDataAvailable Event reagieren und darin dann HandleData vom IO-Handler aufrufen. Bzw. im Fall von blocking Sockets hat der Benutzer entsprechend nach neuen Daten gepollt und bei Empfang dann HandleData aufgerufen.

Ist natürlich ohne Frage der Ansatz, der dem Benutzer die größte Freiheit lässt.

Was mich einfach stört ist, dass das Senden dann dem IO-Handler obliegt, das Empfangen aber in der Hand des Benutzers ist. Das Receive im Socket Interface wäre damit auf jeden Fall obsolet.

[Namenloser]

Was soll der IOHandler denn genau machen? Sorry, wenn es eine dumme Frage ist, ich kenne den Begriff "IOHandler" von Indy, aber habe Indy das letzte mal vor wahrscheinlich 10 Jahren angerührt, und mich damals auch nicht wirklich mit der Klassenstruktur auseinandergesetzt.

Also eigentlich ist das Paradigma bei blockierenden Sockets ja, dass man immer abwechselnd liest und schreibt. Deswegen ist mir nicht so ganz klar, warum du im Konstruktor bereits anfängst zu lesen (und einen extra Thread für das Senden erzeugst). Denn es hängt ja vom konkreten Protokoll ab, ob zuerst gelesen oder zuerst geschrieben werden muss. Soll IOHandler ein Platzhalter für die Implementierung des Protokolls sein, oder ist es eine Zwischenschicht zwischen Socket und Protokollimplementierung?

Ich entwickle "derzeit" (in Anführungszeichen, weil ich seit Monaten nicht dazu gekommen bin, daran zu weiterarbeiten) auch eine Socket-Lib und bin inzwischen zu dem Ergebnis gekommen, dass blockierend und eventgetrieben (also sowas wie OnDataAvailable) eine grundlegend murxige Kombination ist. Meiner Meinung gibt es zwei lokale Maxima im Raum der möglichen Designs für eine Socket-Lib:
1. Blockierend, imperativ (also eine Receive- und eine Send-Methode, so wie bei dir in ISocket)
2. Nicht-Blockierund und eventgetrieben (Protokollimplementierung als State Machine) - darauf konzentriert sich meine Lib

[OT]Übrigens, weil ich gerade sehe, dass du auch Interfaces für die Sockets verwendest: Es ist schockierend, wie viel Overhead die Referenzzählung bei Interfaces verursacht, zumindest mit dem Code, den Freepascal generiert. Ich hatte mal einen ganz kleinen Test-HTTP-Server, der nur "Hello World" ausgibt, gegen nginx gebenchmarkt, und nginx war selbst mit nur einem Kern schneller als mein Server mit allen vier Kernen. Sogar Apache war schneller! Obwohl mein Server ja nicht mal den Funktionsumfang eines richt HTTP-Servers hat. Habe das dann mal mit Valgrind und Co. analysiert, weil ich dachte, dass ich irgendwas grundlegend falsch oder anders mache, aber es stellte sich raus, dass der einzige Unterschied anscheinend auf das Refcounting zurückzuführen ist, das einen riesen Rattenschwanz erzeugt (Exception-Stack, RelocateThreadVars...). Ansonsten sieht die Aufrufstatistik fast identisch aus wie bei nginx. Ich werde wahrscheinlich deshalb, wenn ich mal wieder dazu komme, bei mir alle Interfaces rausschmeißen.[/OT]

Zur konkreten Frage: Ich würde auch auf jeden Fall im Constructor keine Reads, Writes oder sonstwas ausführen. Dann lieber eine extra Routine wie Start() , wenn es denn sein muss.

[Zacherl]

Zitat von Namenloser:

Also eigentlich ist das Paradigma bei blockierenden Sockets ja, dass man immer abwechselnd liest und schreibt. Deswegen ist mir nicht so ganz klar, warum du im Konstruktor bereits anfängst zu lesen (und einen extra Thread für das Senden erzeugst). Denn es hängt ja vom konkreten Protokoll ab, ob zuerst gelesen oder zuerst geschrieben werden muss. Soll IOHandler ein Platzhalter für die Implementierung des Protokolls sein, oder ist es eine Zwischenschicht zwischen Socket und Protokollimplementierung?

Der IO-Handler ist im Grunde ein (De-)Multiplexer, mit dem ich mehrere Streams / Pakete über ein einzelnes Socket tunnele. Es soll aber möglich sein Daten gleichzeitig zu senden und zu empfangen. Aus diesem Grund habe ich einen Thread, der nur pollt und einen weiteren Thread, der sich um das gechunkte Versenden der ausgehenden Daten kümmert. Das klassische Request-Response Prinzip kann ich hier nicht ohne Weiteres anwenden.

Falls du bessere Ideen zur Umsetzung hast, gerne immer her damit

Konkret möchte ich halt mehrere simultane (unterschiedlich priorisierte) Übertragungen über ein einzelnes Socket realisieren, wobei sowohl Client, als auch Server auf Benutzerinteraktion hin eigenständig Datenübertragungen beginnen können. Die Kummunikation folgt also nicht wirklich einem fest definierten linearen Verlauf.

Zitat von Namenloser:

[OT]Übrigens, weil ich gerade sehe, dass du auch Interfaces für die Sockets verwendest: Es ist schockierend, wie viel Overhead die Referenzzählung bei Interfaces verursacht ...[/OT]

Danke für den Hinweis, da werde ich mal drauf achten.

[himitsu]

Overhead?

[Namenloser]

Zitat von Zacherl:

Es soll aber möglich sein Daten gleichzeitig zu senden und zu empfangen.

Ok, verstehe. Ist bei mir ganz ähnlich (bis auf die Priorisierung, das brauche ich bisher nicht). Das war auch bei mir der Grund, weshalb ich angefangen habe, etwas eigenes zu schreiben, weil die herkömmlichen Libs das irgendwie alle nicht unterstützten.

Zitat von Zacherl:

Aus diesem Grund habe ich einen Thread, der nur pollt und einen weiteren Thread, der sich um das gechunkte Versenden der ausgehenden Daten kümmert. Das klassische Request-Response Prinzip kann ich hier nicht ohne Weiteres anwenden.

Falls du bessere Ideen zur Umsetzung hast, gerne immer her damit

Konkret möchte ich halt mehrere simultane (unterschiedlich priorisierte) Übertragungen über ein einzelnes Socket realisieren, wobei sowohl Client, als auch Server auf Benutzerinteraktion hin eigenständig Datenübertragungen beginnen können. Die Kommunikation folgt also nicht wirklich einem fest definierten linearen Verlauf.

Ich weiß nicht, ob ich eine "bessere Idee" habe, aber ich war nach längerer Überlegung zu dem Ergebnis gekommen, dass ich mit blockierenden Sockets in Teufels Küche komme und habe mich für einen nicht-blockierenden Ansatz entscheiden. Ich weiß aber nicht mehr genau warum, weil ich jetzt wie gesagt, mich seit Monaten nicht damit beschäftigt habe, und ziemlich raus bin.

Ich versuche einfach mal grob meinen aktuellen Ansatz zu erklären. Im Grunde besteht die Architektur bei mir aus zwei Schichten. Die unterste Schicht sieht so aus und ist eigentlich nur eine 1:1-Kapselung der BSD-Socket-API:

Delphi-Quellcode:

  { IWorker }

  IWorker = interface(ISuiteObject)
    procedure Work;
    procedure Wake;
  end;

  { IWorkerComponent }

  IWorkerComponent = interface(ISuiteObject)
    procedure AddToWorker(const Worker: IWorker);
    procedure RemoveFromWorker(const Worker: IWorker);
    procedure RemoveFromAllWorkers;
  end;

  { ISocketEventHandler }

  ISocketEventHandler = interface
    procedure HandleCanRecv(const Socket: ISocket; const Worker: IWorker);
    procedure HandleCanSend(const Socket: ISocket; const Worker: IWorker);
    procedure HandleCanAccept(const Socket: ISocket; const Worker: IWorker);
    procedure HandleConnect(const Socket: ISocket; const Worker: IWorker);
    procedure HandleDisconnect(const Socket: ISocket; const Worker: IWorker);
    procedure HandleError(const Socket: ISocket; const Worker: IWorker; const Error: LongInt);
  end;

  { ISocket }

  ISocket = interface(IWorkerComponent)
    function GetEventHandler: ISocketEventHandler;
    procedure SetEventHandler(const AValue: ISocketEventHandler);

    procedure SetBlocking(Blocking: Boolean; out OpResult: TSocketOpResult);
    procedure Shutdown(how: cint; out OpResult: TSocketOpResult); overload;
    procedure Shutdown(out OpResult: TSocketOpResult); overload;

    // Habe ich entfernt, weil ich nach reiflicher Überlegung zu dem Schluss gekommen bin,
    // dass sich bestimmte Race Conditions einfach nicht verhindern lassen. Close wird 
    // automatisch im Destructor aufgerufen.
    //procedure Close(out OpResult: TSocketOpResult); 

    function Send(const Buffer; Size: SizeInt; out OpResult: TSocketOpResult): SizeInt;
    function Receive(var Buffer; Size: SizeInt; out OpResult: TSocketOpResult): SizeInt;
    procedure Bind(const AddrInfo: TAddrInfo; out OpResult: TSocketOpResult);
    procedure Listen(out OpResult: TSocketOpResult);
    procedure Connect(const AddrInfo: TAddrInfo; out OpResult: TSocketOpResult);
    function Accept(out Addr: TSockAddrStorage; out OpResult: TSocketOpResult): ISocket;
    function GetPeerName(out OpResult: TSocketOpResult): TSockAddrStorage;

    procedure SetSockOpt(level:cint; optname:cint; optval:pointer;
                         optlen:tsocklen; out OpResult: TSocketOpResult); overload;
    procedure SetSockOpt(level: cint; optname: cint; optval: boolean;
                         out OpResult: TSocketOpResult); overload;

    property EventHandler: ISocketEventHandler read GetEventHandler write SetEventHandler;
  end;

  { ISocketSuite }

  ISocketSuite = interface
    function CreateWorker(const Logger: ILogger): IWorker;
    function CreateSocket: ISocket; overload;
    function CreateSocket(const Handle: TFdHandle;
      const ConnectionState: TSocketState): ISocket; overload;
    function CreateSocket(Family: cint; SockType: cint; Protocol: cint;
      out OpResult: TSocketOpResult): ISocket; overload;
    function CreateCompatibleSocket(const AddrInfo: TAddrInfo;
      out OpResult: TSocketOpResult): ISocket;
    function CreateTimer: ITimer;
  end;

Man erzeugt also mithilfe der ISocketSuite -Factory einen ISocket , und diesen fügt man dann einem oder mehren IWorkern hinzu. Man muss in einer Schleife IWorker.Work aufrufen, um die anfallenden Events abzuarbeiten (kann und wird man in der Regel natürlich in einem separaten Thread tun. Da gibt es bei mir auch schon eine fertige Klasse für). Die genauen Details lasse ich hier weg, aber letztendlich resultiert das dann darin, dass in IWorker.Work die entsprechenden Methoden des ISocketEventHandler aufgerufen werden.

Es liegt dann beim Implementierer des ISocketEventHandler , wie er auf die Events reagiert. Wenn z.B. HandleCanRecv aufgerufen wurde, dann weiß ich nur, dass irgendwann in der Zwischenzeit mal Daten angekommen sind bzw. sein könnten, aber ich bekomme sie noch nicht direkt – ich kann dann Socket.Receive aufrufen, um die Daten (sofern vorhanden) abzurufen, oder einfach nichts tun, oder ganz was anderes machen. Allerdings werde ich erst wieder erneut benachrichtigt, nachdem ich Socket.Receive aufgerufen habe. Analog läuft es bei den anderen Methoden.

Die zweite Schicht sieht bei mir so aus:

Delphi-Quellcode:

TCustomConnection = class(TWorkerComponent, ISocketEventHandler, ITimerHandler)
  private
    FSocket: ISocket;
    { ... }

    // ISocketEventHandler
    procedure SocketCanRecv(const Socket: ISocket; const Worker: IWorker);
    procedure SocketCanSend(const Socket: ISocket; const Worker: IWorker);
    procedure SocketConnect(const Socket: ISocket; const Worker: IWorker);
    procedure SocketDisconnect(const Socket: ISocket; const Worker: IWorker);
    procedure SocketCanAccept(const Socket: ISocket; const Worker: IWorker);
    procedure SocketError(const Socket: ISocket; const Worker: IWorker; const Error: LongInt);

    procedure ISocketEventHandler.HandleCanRecv = SocketCanRecv;
    procedure ISocketEventHandler.HandleCanSend = SocketCanSend;
    procedure ISocketEventHandler.HandleConnect = SocketConnect;
    procedure ISocketEventHandler.HandleDisconnect = SocketDisconnect;
    procedure ISocketEventHandler.HandleCanAccept = SocketCanAccept;
    procedure ISocketEventHandler.HandleError = SocketError;
  protected
    { ... }

    // TWorkerComponent
    procedure DoAddToWorker(const Worker: IWorker); override;
    procedure DoRemoveFromWorker(const Worker: IWorker); override;

    procedure HandleCanRecv(const Worker: IWorker); virtual;
    procedure HandleCanSend(const Worker: IWorker); virtual;
    procedure HandleConnect(const Worker: IWorker); virtual;
    procedure HandleDisconnect(const Worker: IWorker); virtual;
    procedure HandleError(const Worker: IWorker; const Error: LongInt); virtual;
    
    { ... }

    procedure Shutdown(how: cint; out OpResult: TSocketOpResult); overload;
    procedure Shutdown(out OpResult: TSocketOpResult);  overload;
    procedure Close(out OpResult: TSocketOpResult);

    function Send(const Buffer; Size: SizeInt; out OpResult: TSocketOpResult): SizeInt;
    function Receive(var Buffer; Size: SizeInt; out OpResult: TSocketOpResult): SizeInt;

    function GetPeerName(out OpResult: TSocketOpResult): TSockAddrStorage;

    procedure SetSockOpt(level:cint; optname:cint; optval:pointer;
                         optlen:tsocklen; out OpResult: TSocketOpResult); overload;
    procedure SetSockOpt(level: cint; optname: cint; optval: boolean;
                         out OpResult: TSocketOpResult);  overload;
    procedure Connect(const Address: String; const Port: Word; out OpResult: TSocketOpResult);
  public
    constructor Create(const Suite: ISocketSuite); overload;
    constructor Create(const Socket: ISocket); overload;
    destructor Destroy; override;
  end;

Diese Klasse ist eigentlich nur eine Kapselung um den Socket, stellt aber noch Zusatzfunktionen bereit, die nicht direkt Teil der Socket-API sind, z.B. Timer/Timeouts (gehe ich hier nicht näher drauf ein). Das ist die Klasse, von der man sich seine eigenen Implementierungen für sein jeweiliges Protokoll ableiten sollte. Zumindest ist es so gedacht. Man kann natürlich auch ISocket direkt verwenden, aber TCustomConnection bietet mehr Komfort.

Eine solche Implementierung sieht dann z.B. in meinem Fall so aus (das ist nicht mehr Teil der Lib sondern der Anwendung):

Delphi-Quellcode:

  { IConnection }

  IConnection = interface(IWorkerComponent)
    procedure SendEvent(const Event: TEvent; out OpResult: TSocketOpResult);

    { ... }

    property OnReceiveEvent: TConnectionOnReceiveEvent read GetOnReceiveEvent write SetOnReceiveEvent;

    { ... }

    property OnConnect: TConnectionOnConnect read GetOnConnect write SetOnConnect;
    property OnDisconnect: TConnectionOnDisconnect read GetOnDisconnect write SetOnDisconnect;
    property OnError: TConnectionOnError read GetOnError write SetOnError;

    { ... }
  end;


  { TConnection }

  TConnection = class(TCustomConnection, IConnection)
  protected
    FOnReceiveEvent: TConnectionOnReceiveEvent;
    { ... }
    FOnConnect: TConnectionOnConnect;
    FOnDisconnect: TConnectionOnDisconnect;
    FOnError: TConnectionOnError;

    FSendBuffer: String;
    FRecvBuffer: String;
    FSendMutex: TCriticalSection;
    FRecvMutex: TCriticalSection;

    procedure HandleConnect(const Worker: IWorker); override;
    procedure HandleCanRecv(const Worker: IWorker); override;
    procedure HandleCanSend(const Worker: IWorker); override;
    procedure HandleDisconnect(const Worker: IWorker); override;
    procedure HandleError(const Worker: IWorker; const Error: LongInt); override;
    procedure HandleRecvIdleTimeout(const Worker: IWorker); override;
    procedure HandleSendIdleTimeout(const Worker: IWorker); override;

    { ... }

    procedure SendEvent(const Event: TEvent; out OpResult: TSocketOpResult);
    { ... }
  public
    constructor Create(const Suite: ISocketSuite); overload;
    constructor Create(const Socket: ISocket); overload;
    destructor Destroy; override;
  end;

Wenn ich also IConnection.SendEvent(Event) aufrufe, dann wird Event in einen String konvertiert und an den internen FSendBuffer angehängt (man könnte statt Buffer auch Queue sagen). Der FSendBuffer wird dann peu à peu abgearbeitet, immer dann, wenn durch den Worker wieder HandleCanSend aufgerufen wird.

Umgekehrt läuft es beim Empfangen ab: In HandleCanRecv wird erst Recv() aufgerufen, und dann die empfangenen Daten an FRecvBuffer angehängt. Immer wenn eine Nachricht vollständig ist, wird OnReceiveEvent aufgerufen.

(Strings als Buffer sind natürlich nicht effizient, aber für diese konkrete Anwendung ist das egal. Vielleicht mache ich das irgendwann mal richtig, aber andere Dinge sind wichtiger...)

Ich hoffe, es war nicht zu verwirrend. Ich versuche es noch mal kurz zusammenzufassen. Also:

• Daten kommen an -> HandleCanRecv() -> Recv() -> empfangene Daten werden an FRecvBuffer angehängt -> Wenn Nachricht vollständig, OnReceiveEvent()
• SendEvent() -> Daten werden an FSendBuffer angehängt -> Send() -> ... -> HandleCanSend() -> (falls noch nicht alle Daten versendet:) Send() -> ... -> HandleCanSend() usw.

Nachteil bei dieser Implementierung, so wie sie hier ist: Die Buffer von TConnection können theoretisch beliebig groß werden, wenn die Daten entweder zu schnell reinkommen oder nicht schnell genug versandt werden können. Das ist mir bewusst, aber bei dieser konkreten Anwendung kein ernsthaftes Problem. Eine allgemeine Lösung gibt es dafür eh nicht, das kommt immer auf die Anwendung an. Man könnte in so einem Fall blockieren, oder Nachrichten verwerfen, oder einen Fehler zurückgeben, oder oder oder. Es ist also keine Beschränkung von meiner Architektur, sondern ich habe mir da bisher einfach keine Mühe gemacht, weil es nicht notwendig war.

Analog zu TCustomConnection und IConnection gibt es natürlich auch noch TCustomListener und IListener für den Server-Teil. Und noch einige andere Sachen, die ich hier weggelassen habe... Der Beitrag ist länger geworden, als ich wollte.