Mit Datagram Sockets können schwindelerregende Übertragungsraten erreicht werden. Die Spitze sind wieder Datagram Unix Domain Sockets. Mit denen habe ich für die Übertragung von über einer Million (konrekt 1024^2) Paketen mit je einem Byte 1.358 Sekunden gebraucht.
Und jetzt kommt’s: Für die Übertragung von genauso vielen Paketen mit je 1024 Byte (d.h. für die Übertragung von 1GB Daten) waren es 1.430 Sekunden! Offensichtlich spielt der Verbindungsaufbau die größere und der Payload eher eine untergeordnete geordnete Rolle.
Um mich mit meinen Behauptungen nicht zu blamieren, hab ich mir den Fall Datagramm per Unix Domain Sockets etwas genauer angeschaut, auch wenn der Code dadurch etwas unübersichtlicher geworden ist.
- Jeder der übertragenen Puffer wird mit Zufallszahlen gefüllt um eventuelle Caches im TCP-Stack abzuschalten
- Der Overhead für die Generierung der Zufallszahlen wird ermittelt
- Ein kleines Sub-Protokoll wird implementiert um Verwaltungsinformationen vom Client an den Server zu Übermitteln
- Überprüfung ob jedes einzelne Bytes fehlerfrei angekommen ist
Der folgende Ausschnitt zeigt die Ausgabe einer Sitzung in der 16 GB Zufallszahlen übertragen wurden:
Warte auf Daten … Bufferlänge: 16384 Loops: 1048576 Checksum: 1082323705659 vs 1082323705659 1048576 Nachrichten je 16384 Bytes gelesen (16384.0MB) in 143.3s Davon entfallen 106.3s auf die Generierung der 17179869184 Zufallszahlen Reine Transferzeit für 16384.0MB: 37.0s
An einigen Stellen kann man das Ganze bestimmt noch schöner machen. Zum Beispiel mit Programmargumenten insbesondere den Client flexibler gestalten, aber für die beabsichtigten Zwecke reicht es aus.
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <sys/un.h>
#include <unistd.h>
#include <sys/socket.h>
#include <chrono>
#include <arpa/inet.h>
int main(int, char**)
{
int server = socket(AF_UNIX, SOCK_DGRAM, 0); // Datagram-Socket
// Richte den Unix Domain Socket fuer den Server ein
const char* server_socket_path = "server.sock";
int addr_len = sizeof(sockaddr_un);
struct sockaddr_un server_addr;
memset(&server_addr, 0, addr_len);
server_addr.sun_family = AF_UNIX;
strcpy(server_addr.sun_path, server_socket_path);
unlink(server_socket_path);
if( bind(server, (struct sockaddr *) &server_addr, addr_len) < 0 )
{
fprintf(stderr, "bind: %s\n", strerror(errno));
close(server);
exit(1);
}
while( true )
{
printf("Warte auf Daten ...\n");
struct sockaddr_un remote_addr;
long bufferlaenge;
recvfrom(server, &bufferlaenge, sizeof(bufferlaenge), 0, (struct sockaddr *) &remote_addr, (socklen_t*)&addr_len);
bufferlaenge = ntohl(bufferlaenge);
fprintf(stderr, "Bufferlänge: %lu\n", bufferlaenge);
char buffer[bufferlaenge] = {0};
long loops;
recvfrom(server, &loops, sizeof(loops), 0, (struct sockaddr *) &remote_addr, (socklen_t*)&addr_len);
loops = ntohl(loops);
fprintf(stderr, "Loops: %ld\n", loops);
int count = 0;
unsigned long long empfangen = 0LL;
unsigned long long checksum = 0LL;
auto start = std::chrono::high_resolution_clock::now();
while(count++ < loops)
{
int gelesen = recvfrom(server, buffer, bufferlaenge, 0, (struct sockaddr *) &remote_addr, (socklen_t*)&addr_len);
if (gelesen < 0)
{
fprintf(stderr, "recvfrom: %s\n", strerror(errno));
close(server);
exit(1);
}
else
{
empfangen += gelesen;
for(int i = 0; i < gelesen; i++)
{
checksum += buffer[i];
}
}
}
unsigned long long senderCheckSum;
recvfrom(server, &senderCheckSum, sizeof(senderCheckSum), 0, (struct sockaddr *) &remote_addr, (socklen_t*)&addr_len);
senderCheckSum = be64toh(senderCheckSum);
unsigned long long clientoverhead;
recvfrom(server, &clientoverhead, sizeof(clientoverhead), 0, (struct sockaddr *) &remote_addr, (socklen_t*)&addr_len);
clientoverhead = be64toh(clientoverhead);
auto end = std::chrono::high_resolution_clock::now();
double ms = std::chrono::duration_cast<std::chrono::nanoseconds>( std::chrono::high_resolution_clock::now() - start).count()/1000000000.0;
fprintf(stderr, "Checksum: %llu vs %llu\n", checksum, senderCheckSum);
fprintf(stdout, "%d Nachrichten je %ld Bytes gelesen (%.1lfMB) in %.1lfs\n", count-1, bufferlaenge, empfangen/1024.0/1024, ms);
fprintf(stdout, "Davon entfallen %.1lfs auf die Generierung der %llu Zufallszahlen\n", clientoverhead/1024.0/1024/1024.0, empfangen);
fprintf(stdout, "Reine Transferzeit für %.1lfMB: %.1lfs\n", empfangen/1024.0/1024, ms - clientoverhead/1024.0/1024/1024.0);
}
close(server);
return 0;
}
Mit dem hier abgebildeten Client wird etwa 1GB übertragen und die Laufzeit incl. der Generierung der Zufallszahlen und dem Probelauf liegt bei ca. 20 Sekunden.
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <vector>
#include <algorithm>
#include <endian.h>
#include <arpa/inet.h>
#include <chrono>
struct CheckSum
{
long long sum{0};
void operator()(int n) { sum += (long long)n; }
};
int main(void)
{
unsigned long messagelength = 1024;
char message[messagelength] = {0};
long loops = 1024<<10;
unsigned long long checksum = 0;
// Probelauf
fprintf(stderr, "Berechne Overhead...\n");
long dummy = loops;
auto probestart = std::chrono::high_resolution_clock::now();
while(dummy--)
{
std::for_each(message, message+messagelength, [](auto& n) { n = rand()%127; });
CheckSum checkSum = std::for_each(message, message+messagelength, CheckSum());
checksum += checkSum.sum;
}
checksum = 0;
auto probeende = std::chrono::high_resolution_clock::now();
unsigned long long overhead = std::chrono::duration_cast<std::chrono::nanoseconds>( probeende - probestart).count();
int s = socket(AF_UNIX, SOCK_DGRAM, 0);
if (s == -1)
{
fprintf(stderr, "Fehler: %s\n", strerror(errno));
exit(1);
}
struct sockaddr_un server_addr;
memset(&server_addr, 0, sizeof(struct sockaddr_un));
server_addr.sun_family = AF_UNIX;
strcpy(server_addr.sun_path, "server.sock");
fprintf(stderr, "Sende die Größe des Message-Buffers: %lu\n", messagelength);
unsigned long pack = htonl(messagelength);
if( sendto(s, &pack, sizeof(pack) , 0, (struct sockaddr *) &server_addr, sizeof(server_addr)) < 0)
{
fprintf(stderr, "sendto: %s\n", strerror(errno));
close(s);
exit(1);
}
fprintf(stderr, "Sende Anzahl der Loops %ld\n", loops);
pack = htonl(loops);
if( sendto(s, &pack, sizeof(pack) , 0, (struct sockaddr *) &server_addr, sizeof(server_addr)) < 0)
{
fprintf(stderr, "sendto: %s\n", strerror(errno));
close(s);
exit(1);
}
printf("Sende Daten...\n");
while(loops--)
{
// Nachricht mit kleinen Zufallszahlen initialisieren
std::for_each(message, message+messagelength, [](auto& n) { n = rand()%127; });
CheckSum checkSum = std::for_each(message, message+messagelength, CheckSum());
checksum += checkSum.sum;
if( sendto(s, message, messagelength, 0, (struct sockaddr *) &server_addr, sizeof(server_addr)) < 0)
{
fprintf(stderr, "sendto: %s\n", strerror(errno));
close(s);
exit(1);
}
}
checksum = htobe64(checksum);
sendto(s, &checksum, sizeof(checksum), 0, (struct sockaddr *) &server_addr, sizeof(server_addr));
overhead = htobe64(overhead);
sendto(s, &overhead, sizeof(overhead), 0, (struct sockaddr *) &server_addr, sizeof(server_addr));
close(s);
return 0;
}