Klassendesign: Blockierender Constructor?

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