Socket: Übertragung "Ruckelt"

teddds

Hallo,

Ich bin dabei Kamerabilder unkomprimiert (nur zur Testzwecken) über eine lokales Gbit Netz su senden. Ich nutze dabei eine Client-Server Struktur, realisiert mittels Socket und der Winapi.

Wenn ich den Server und den Klient lokal starte und die 127.0.0.1 als IP zum connecten verwende funktioniert alles wunderbar. Ich komme dort auch auf die maximale Rate der Kamera von 16 Bilder/Sekunde. Starte ich aber den Klienten auf einen zweiten Rechner, kommt es bei der Übertragung zu einem "Ruckeln". Etwa 10 Bilder kommen flüssig (16Hz) an und danach hängt die Verbindung für etwa eine halbe Sekunde und die Rate geht auf 2 Hz runter. Danach kommen die nächsten 10 Bilder wieder flüssig. Die Verbindung hängt sozusagen kurz. An der Datenmenge kann es nicht liegen, da dieses Phänomen auch bei sehr geringen Auflösungen auftritt (320X240).

Ich denke das hat irgendwie was mit meiner Netzwerkimplementierung zu tun, da dies die einzige Komponente ist die sich bei dem Test mit zwei Rechnern geändert hat. Aber ich wüsste nicht was daran falsch ist. Das ganze läuft nach folgendem Schema ab:
Server schickt kurzes Paket der Länge 30, indem steht wie Groß das kommende Bild ist. Der Klient empfängt das Paket (die Länge oder Buffergröße des kommenden Bildes) und schick eine Mitteilung an den Server du kannst jetzt das Bild losschicken. Der Server schickt das ganze Paket raus und wartet wieder auf eine Antwort. Also ob alles i.O. war oder ob das Paket nochmal geschickt werden soll oder ob zum Beispiel die Verbindung getrennt wurde. Das ganze wiederholt sich ständig.

Vielleicht kennt ihr ja noch einen Tip.

Network::Client

#include "StdAfx.h"
/*
#include "networkClient.h"
#include <Windows.h>
#include <stdio.h>
#include <iostream>
#include <sstream>
#include <string>
*/
#define ERROR "ERROR"
#define OK "OK"

using namespace network;
using namespace std;
//------------------------------------------------------------------------------------------------
//----------------------------------   Client   --------------------------------------------------
//------------------------------------------------------------------------------------------------
Client::Client(void)
{
}

Client::~Client(void)
{
}

bool Client::connect2Server(const char* ip, int port)
{
	//Winsocks starten und mit dem Server verbinden
	TotalDataSend = 0;
	TotalDataReceive = 0;
	long rc;
	WSADATA wsa;
	rc= WSAStartup(MAKEWORD(2,0),&wsa);
	if(rc!=0)
	{
		printf("Fehler: startWinsock, fehler code: %d\n",rc);
		return false;
	}
	else
	{
		printf("Winsock gestartet!\n");
		Socket=socket(AF_INET,SOCK_STREAM,0);

		if(Socket==INVALID_SOCKET)
		{
		  printf("Fehler: Der Socket konnte nicht erstellt werden, fehler code: %d\n",WSAGetLastError());
		  return false;
		}
		else
		{
		  printf("Socket erstellt!\n");
		  memset(&addr,0,sizeof(SOCKADDR_IN)); 

			addr.sin_family=AF_INET;
			addr.sin_port=htons(port); 
			addr.sin_addr.s_addr=inet_addr(ip); 

			rc=connect(Socket,(SOCKADDR*)&addr,sizeof(SOCKADDR));

			if(rc==SOCKET_ERROR)
			{
			  printf("Fehler: connect gescheitert.\n");
			  return false;
			}
			else
			{
				printf("Verbunden mit %s\n",ip);
			}
		}
	}
	return true;
}

void Client::cleanUp()
{
	//Aufräumen
	closesocket(Socket);
	WSACleanup();
}

char* Client::recvData(int& awnser, int& BufferLength)
{
	// Empfange zunächst die Paketgröße
	char* responseSize = new char[30];
	int bytes = -1;
	int Size = -1;
	int myRecievedBytes = 0;
	int iOld = 0;
	char* response;
	bool sendMeAgain = false;
	string tmp;
	do
	{
		for (int i=0; i<30; i++) 
		{
			responseSize[i] = 0;    
		}

		bytes = recv(Socket, responseSize, 30, 0);
		TotalDataReceive += bytes;
		std::string mySize = responseSize;
		stringstream ss(mySize); 
		ss >> Size;
		if(bytes >= 0)
		{
			if(Size <= 0)
			{
				this->sendERROR();
				sendMeAgain = true;
			}
			else
			{
				this->sendOK();
				sendMeAgain = false;
			}
		}
		else
		{
			sendMeAgain = false;
		}
	}
	while(sendMeAgain != false);
	delete [] responseSize;
	//int round = 0;
	if(bytes == -1)
	{
		awnser = -1;
		return NULL;
	}
	else
	{
		//Erstelle Buffer und empfange komplettes Paket 
		BufferLength = Size;
		response = new char[Size];
		do
		{
			char* partOfResponse = NULL;
			partOfResponse = new char[Size];
			bytes = recv(Socket, partOfResponse, Size, 0);
			TotalDataReceive += bytes;
			if(bytes > 0)
			{

				for(int i=0; i<bytes; i++)
				{
					response[iOld+i] = partOfResponse[i];
				}
				myRecievedBytes += bytes;
				iOld += bytes;
			}
			delete [] partOfResponse;

		}
		while((myRecievedBytes != Size) && (bytes > 0));
		if(bytes == -1)
		{
			// incomplete transfer
			awnser = -1;
			return NULL;
		}
		else if(bytes == 0)
		{
			// connection closed
			awnser = 1;
			return NULL;
		}
		else
		{
			// transfer successful
			this->sendOK();
			awnser = 0;

			return response;
		}
	}
}

int Client::sendData(std::string s)
{
	// Sende String
	return this->sendData(s.c_str(),s.length());
}

int Client::sendData(int value)
{
	// Sende Int
	std::string buf1;
	std::stringstream out1;
	out1 << value;
	out1 >> buf1;
	return this->sendData(buf1.c_str(),buf1.length());
}

int Client::sendData(const char* buff, int buffLength)
{
	// Sende Daten an Server
    int awnser = -2;
	bool SendItAgain = false;
	stringstream ss;
	ss << buffLength;

	// Sende Puffergröße
	do
	{
		TotalDataSend += send(Socket,ss.str().c_str(), 30, 0);
		awnser = this->wait4OK();
		if(awnser == -1)
		{
			SendItAgain = true;
		}		
		else
		{
			SendItAgain = false;
		}
	}
	while(SendItAgain != false);

	if(awnser == 0)
	{
		awnser = -2;
		// Sende Inhalt
		do
		{
			TotalDataSend += send(Socket,buff, buffLength, 0);
			awnser = this->wait4OK();
			if(awnser == -1)
			{
				SendItAgain = true;
			}		
			else
			{
				SendItAgain = false;
			}
		}
		while(SendItAgain != false);
		if(awnser == 0)
		{
			//Alles übertragen
			return 0;
		}
		else if(awnser == 1)
		{
			//Server hat verbindung unterbrochen
			return 1;
		}
	}
	else if(awnser == 1)
	{
		//Server hat verbindung unterbrochen
		return 1;
	}
	else
	{
		//Transfer Incorrect
		return -1;
	}
}

int Client::wait4OK()
{
	// Wartet auf ack vom Server
	char* response = new char[10];
    int bytes = recv(Socket, response, 10, 0);
	TotalDataReceive += bytes;
	if(bytes > 0)
	{
		std::string tmp = response;
		if(tmp.compare("OK") == 0)
		{
			delete []response;
			return 0;
		}
		else
		{
			delete []response;
			return -1;
		}
	}
	else
	{
		delete []response;
		return 1;
	}
}

void Client::sendERROR()
{
	// Sendet Fehler an Server (Bitte um erneutes Senden)
	string tmp = ERROR;	
	TotalDataSend += send(Socket,tmp.c_str(), 10, 0);
}

void Client::sendOK()
{
	// Sendet ack an den Server
	string tmp = OK;	
	TotalDataSend += send(Socket,tmp.c_str(), 10, 0);
}

Network::Server

#include "stdafx.h"

using namespace network;
using namespace std;
//------------------------------------------------------------------------------------------------
//----------------------------------   Server   --------------------------------------------------
//------------------------------------------------------------------------------------------------
Server::Server(void)
{
}

Server::~Server(void)
{
}

bool Server::startServer(int port)
{
	// Winsock starten und Server hochfahren
	long rc;
	WSADATA wsa;
	rc=WSAStartup(MAKEWORD(2,0),&wsa);
	if(rc!=0)
	{
		printf("Fehler: startWinsock, fehler code: %d\n",rc);
		return false;
	}
	else
	{
		printf("Winsock gestartet!\n");	

		// Socket erstellen
		acceptSocket=socket(AF_INET,SOCK_STREAM,0);
		if(acceptSocket==INVALID_SOCKET)
		{
			printf("Fehler: Der Socket konnte nicht erstellt werden, fehler code: %d\n",WSAGetLastError());
			return false;
		}
		else
		{
			printf("Socket erstellt!\n");

			memset(&addr,0,sizeof(SOCKADDR_IN));
			addr.sin_family=AF_INET;
			addr.sin_port=htons(port);
			addr.sin_addr.s_addr=INADDR_ANY;
			rc=bind(acceptSocket,(SOCKADDR*)&addr,sizeof(SOCKADDR_IN));
			if(rc==SOCKET_ERROR)
			{
				printf("Fehler: bind, fehler code: %d\n",WSAGetLastError());
				return false;
			}
			else
			{
				printf("Socket an Port %i gebunden\n",port);
				rc=listen(acceptSocket,10);

				if(rc==SOCKET_ERROR)
				{
					printf("Fehler: listen, fehler code: %d\n",WSAGetLastError());
					return false;
				}
				else
				{
					printf("acceptSocket ist im listen Modus....\n"); 
					connectedSocket=accept(acceptSocket,NULL,NULL);
					if(connectedSocket==INVALID_SOCKET)
					{
						printf("Fehler: accept, fehler code: %d\n",WSAGetLastError());
						return false;
					}
					else
					{
						printf("Neue Verbindung wurde akzeptiert!\n");
					}
				}
			}
		}
	}
	return true;
}

void Server::cleanUp()
{
	//Sockets auflösen und Server beenden
	if(connectedSocket != NULL)
	{
		closesocket(connectedSocket);
	}
	if(acceptSocket != NULL)
	{
		closesocket(acceptSocket);
	}
	WSACleanup();
}

char* Server::recvData(int& awnser)
{
	// Empfange zunächst die Paketgröße
	char* responseSize = new char[30];
	int bytes = -1;
	int Size = -1;
	int myRecievedBytes = 0;
	int iOld = 0;
	bool sendMeAgain = false;
	char* response;
	string tmp;
	do
	{
		for (int i=0; i<30; i++) 
		{
			responseSize[i] = 0;    
		}

		bytes = recv(connectedSocket, responseSize, 30, 0);
		std::string mySize = responseSize;
		stringstream ss(mySize); 
		ss >> Size;
		if(bytes >= 0)
		{
			if(Size < 0)
			{
				this->sendERROR();
				sendMeAgain = true;
			}
			else
			{
				this->sendOK();
				sendMeAgain = false;
			}
		}
		else
		{
			sendMeAgain = false;
		}
	}
	while(sendMeAgain != false);
	delete [] responseSize;
	if(bytes == -1)
	{
		awnser = -1;
		return NULL;
	}
	else
	{
		//Erstelle Buffer und empfange komplettes Datenpaket 
		response = new char[Size];
		do
		{
			char* partOfResponse = NULL;
			partOfResponse = new char[Size];
			for (int i=0; i<Size; i++) 
			{
				partOfResponse[i] = NULL;    
			}
			bytes = recv(connectedSocket, partOfResponse, Size, 0);
			if(bytes > 0)
			{
				for(int i=0; i<bytes; i++)
				{
					response[iOld+i] = partOfResponse[i];
				}
				myRecievedBytes += bytes;
				iOld += bytes;
			}
			delete [] partOfResponse;
		}
		while((myRecievedBytes != Size) && (bytes > 0));
		if(bytes == -1)
		{
			// incomplete transfer
			awnser = -1;
			return NULL;
		}
		else if(bytes == 0)
		{
			// connection closed
			awnser = 1;
			return NULL;	
		}
		else
		{
			// transfer successful
			// entpacke Daten wider 
			this->sendOK();
			awnser = 0;

			return response;
		}
	}
}

int Server::sendData(std::string s)
{
	// Sende String
	return this->sendData(s.c_str(),(int)s.length());
}

int Server::sendData(int value)
{
	// Sende Int
	std::string buf1;
	std::stringstream out1;
	out1 << value;
	out1 >> buf1;
	return this->sendData(buf1.c_str(),(int)buf1.length());
}

int Server::sendData(const char* buff, int buffLength)
{
	// Sende Daten an Clienten
    int awnser = -2;
	bool SendItAgain = false;

	stringstream ss;
	ss << buffLength;

	// Sende Puffergröße
	do
	{
		send(connectedSocket,ss.str().c_str(), 30, 0);
		awnser = this->wait4OK();
		if(awnser == -1)
		{
			SendItAgain = true;
		}		
		else
		{
			SendItAgain = false;
		}
	}
	while(SendItAgain != false);

	if(awnser == 0)
	{
		awnser = -2;
		// Sende Inhalt
		do
		{
			send(connectedSocket,buff, buffLength, 0); 
			awnser = this->wait4OK();
			if(awnser == -1)
			{
				SendItAgain = true;
			}		
			else
			{
				SendItAgain = false;
			}
		}
		while(SendItAgain != false);
		if(awnser == 0)
		{
			//Alles übertragen
			return 0;
		}
		else if(awnser == 1)
		{
			//Client hat verbindung unterbrochen
			return 1;
		}
	}
	else if(awnser == 1)
	{
		//Client hat verbindung unterbrochen
		return 1;
	}
	else
	{
		//Transfer Incorrect
		return -1;
	}
	return -1;
}

int Server::wait4OK()
{
	// Wartet auf ack vom Clienten
	char* response = new char[10];
    int bytes = recv(connectedSocket, response, 10, 0);
	if(bytes > 0)
	{
		std::string tmp = response;
		if(tmp.compare("OK") == 0)
		{
			delete []response;
			return 0;
		}
		else
		{
			delete []response;
			return -1;
		}
	}
	else
	{
		delete []response;
		return 1;
	}
}

void Server::sendERROR()
{
	// Sendet Fehler an Clienten (Bitte um erneutes Senden)
	string tmp = ERROR_NETWORK;	
	send(connectedSocket,tmp.c_str(), 10, 0);
}

void Server::sendOK()
{
	// Sendet ack an den Clienten
	string tmp = OK_NETWORK;	
	send(connectedSocket,tmp.c_str(), 10, 0);
}

knivil

Was passiert, wenn ein OK-Paket verloren geht?

teddds

Hmm, guter Einwand.

Also theorethisch warten dann beide auf sich gegseitig und das Programm würde komplett hängen. Also Beispiel:
Server schickt Bild, wartet auf OK;
Client bekommt Bild, sendet OK, welches verloren geht.

Server wartet, also immernoch auf das OK;
und Client wartet auf das nächste Paket (die Buffergröße für das darauffolgende Bild)

Also beide würde dann bis zum timeout warten, bzw. bis die Verbindung unterbochen wird.

Das werde ich noch anpassen. Hat aber leider nix mit dem Problem, des Hängen zutun. Denn mein Programm ist so ausgelegt das wenn die Verbindung unterbrochen würde, das Programm sich wieder beendet. Und das tuts ja nicht. Es hängt einfach für ne halbe Sekunde.... Als, wenn er darauf Wartet die minimal Send-Window größe zusammen zukriegen und dann erst das Paket verschickt.
Ich werde jetzt mal protokoll führen. Wie lange jeder Transfer dauert und was das für ein Paket ist. Vlt. hilft mir das ja weiter.

knivil

Ja, sehe gerade, dass du TCP benutzt.

Moegliche Ursachen: Virenscanner, Torrent-Client, Festplattenzugriffe, ...

teddds

Okay, ich habe jetzt ein Protokoll erstellt.
Jetzt weiss ich das der Hänger immer in folgender Situation auftritt:

Alle 5 Bilder wird ein kleiner String vom Klienten an den Server geschickt. In diesem String stehen die Kameraeinstellungen. Der String sieht so aus:

stringstream ss;
	ss  << "# DesparityMax\n"
		<< myParam->maxDisparity
		<< "\n"
		<< "# DesparityMin\n"
		<< myParam->minDisparity
		<< "\n"
		<< "# Narrow 0 Wide 1\n"
		<< myParam->CamSetting
		<< "\n"
		<< "# Solution  0 = 320x240, 1 = 640x480, 2 = 1280x960\n"
		<< myParam->choosenResolution
		<< "\n"
		<< "# TotalColor 240 - 360\n"
		<< myParam->TotalColorSpread
		<< "\n"
		<< "# Use Compression 0 = None, 1 = Difference, 2 = Zip, 3 = ZipAndDifference\n"
                << myParam->usedCompression;

Also nix weltbewegenes.
Laut Protokoll dauert das Senden der Länge des Strings (also nicht der String selber) auf Clientseite 2.73054e-006s, der Empfang auf Serverseite jedoch ~0,2 Sekunden.
Woran kann das liegen?

EDIT: Firewalls, Virenscanner sind deaktiviert, Festplatte ist eine schnelle SSD

zuckerlie

Du könntest mittels Wireshark deinen Netzwerkverkehr auslesen und schauen, ob es dort Verzögerungen gibt.

Aber schau mal hier:

http://support.microsoft.com/kb/214397/en-us
Vielleicht liegt es daran..

Du solltest die Winsock 2,2 Version aufrufen.

Und ich lege dir vorab http://msdn.microsoft.com/en-us/library/windows/desktop/ms740141%28v=vs.85%29.aspx ans Herz. Die wird Dir sicherlich noch nützlich sein...

//Edit:

typedef struct _SYSTEMTIME {
  WORD wYear;
  WORD wMonth;
  WORD wDayOfWeek;
  WORD wDay;
  WORD wHour;
  WORD wMinute;
  WORD wSecond;
  WORD wMilliseconds;
} SYSTEMTIME, *PSYSTEMTIME;

SYSTEMTIME SysTime;
SYSTEMTIME SysTime2;

GetLocalTime(&SysTime); 
recv(..,..,..,..);
GetLocalTime(&SysTime2); //Späterer Zugriff über die SysTime bzw. SysTime2 //Member.

// Da bekommst du die Uhrzeit des Rechners bis Millisekunden.
// Ich würde es jeweils vor und nach dem send bzw. recv anzeigen, da du somit //mittels der Differenz die benötigte Zeit sehen kannst.

teddds

Danke für den Tip (http://support.microsoft.com/kb/214397/en-us)

Tatsächlich hat der kleine String den durchsatz halbiert und mit deaktivierten Neagle Algoritmus, konnte der Durchsatz wieder verdoppelt werden.

Danke nochmal!

PS:
Ich hatte für die Zeitmessung die QueryPerformanceCounter() Funktion der Winapi genommen.

Aber gut zu wissen das Systemtime auch die Millisekunden speichert.

hustbaer

Ein Problem mit deinem Code ist, dass recv() nicht so funktioniert wie du meinst.
Liest nochmal die Doku nach, vielleicht geht dir dann ein Licht auf.

recv() muss in einer Schleife aufgerufen werden.
Ein guter Tip... überprüfe die Rückgabewerte! Das sollte man immer tun...

teddds

hustbaer schrieb:

Ein Problem mit deinem Code ist, dass recv() nicht so funktioniert wie du meinst.
Liest nochmal die Doku nach, vielleicht geht dir dann ein Licht auf.

Hallo,
ich denke du meinst damit das die Recv() nicht immer alle Bytes abholt und deshalb in einer Loop geholt werden sollte. Das Passiert bei mir allerdings beim großen Datenpaket (dem umkomprimierten Bild). Einzig dem kleinem Paket (in dem nur die Dateigröße des Bildes steht) wird bei mir nicht in einer Loop geholt.

Ich bin gerade dabei das ganze System anders aufzubauen. Meine Überlegung ist nun nicht mehr jeder Nachricht zu quittieren sondern in den Paketen Trennbytes einzufügen und den Klienten dann immer nur bis zu den Trennbytes in der Schleife empfangen zu lassen. Während der Sender immer neue Bilder losschickt.

Kann es bei diesem Asynchronen Verhalten dazu kommen das der Kanal "zugestopft" wird wenn der Sender immer Daten verschickt und der Empfänger nicht schnell genug liest? Ich denke ja schon. Aber wie sollte ich das dann verhindern? Also vlt. ein semi-synchrones System? Der Server haut meinetwegen 20 Bilder in den Kanal und wartet dann erstmal auf eine Antwort vom Klienten bevor er die nächsten 20 Bilder losschickt?

Bin für Tips gerne offen

Danke nochmal für die super Hilfe hier.
Echt super Forum!

Genau, was passiert eigentlich, wenn der Sender zB. 1 MB Upload hat, der Empfänger aber nur 128 kB Download?
Sender sendet wie verrückt (kann er ja, oder? Weiß ja nichts vom Downloadspeed des Empfängers), wo landen die ganzen Daten?

-lowbyte- 0

http://en.wikipedia.org/wiki/ICMP_Source_Quench

hustbaer

Bzw. wenn Pakete verloren gehen (was passiert wenn der Sender zu schnell sendet), dann bekommt das der Sender ja auch mit - der Empfänger bestätigt diese Pakete dann ja nicht.

In dem Fall drosselt der Sender auch die Geschwindigkeit.

Socket: Übertragung &quot;Ruckelt&quot;

Socket: Übertragung "Ruckelt"