Initial commit. Basic test suite

server/.gitignore

@@ -0,0 +1 @@
+build/

server/CMakeLists.txt

@@ -0,0 +1,28 @@
+cmake_minimum_required(VERSION 3.0)
+project(tcp_perf_test_server)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+set(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/../cmake")
+
+find_package(Threads REQUIRED)
+
+## SOURCES  ####################################################################
+set(SOURCE_FILES
+
+tcp_perf_test_server.cpp
+
+task.h
+task.cpp
+
+mutex.h
+mutex.cpp
+
+SynchronousSocket.cpp
+SynchronousSocket.h
+)
+##################################################################################
+
+link_libraries(stdc++ ${CMAKE_THREAD_LIBS_INIT})
+
+add_executable(tcp_perf_test_server ${SOURCE_FILES} ./tcp_perf_test_server.cpp)
+

server/SynchronousSocket.cpp

@@ -0,0 +1,161 @@
+#include "SynchronousSocket.h"
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <netdb.h>
+#include <cstring>
+#include <unistd.h>
+#include <errno.h>
+
+SynchronousSocket::SynchronousSocket()
+{
+	socket = -1;
+}
+
+void SynchronousSocket::Init(int s)
+{
+	socket = s;
+}
+
+bool SynchronousSocket::IsConnected()
+{
+	return (socket != -1);
+}
+
+void SynchronousSocket::Connect(string host, int port)
+{
+	// уже подключены?
+	if (socket != -1) return;
+
+	struct sockaddr_in remoteaddr;
+
+	struct addrinfo *addresses;
+	struct addrinfo hints;
+	memset(&hints, 0, sizeof(hints));
+	hints.ai_socktype = SOCK_STREAM;
+	if (getaddrinfo(host.c_str(), NULL, NULL, &addresses) != 0)
+	{
+		throw (string)("Unable to getaddrinfo");
+		return;
+	}
+	// копипастим себе самый первый адрес (в принципе, нам любой норм)
+	memcpy(&remoteaddr, addresses[0].ai_addr, sizeof(remoteaddr));
+	remoteaddr.sin_port = htons(port);
+	freeaddrinfo(addresses);
+	// создаем сокет
+	if ((socket = ::socket(PF_INET, SOCK_STREAM, 0)) == -1)
+	{
+		throw (string)("Unable to create socket");
+		return;
+	}
+	if (connect(socket, (struct  sockaddr *)(&remoteaddr), sizeof(struct sockaddr_in)) == -1)
+	{
+		if (errno != EINPROGRESS)
+		{
+			socket = -1;
+			throw (string)("Unable to connect");
+			return;
+		}
+	}
+}
+
+void SynchronousSocket::Send(const string & data)
+{
+	int pos = 0;
+	time_t start = time(NULL);
+	while (pos < (int)data.length())
+	{
+		int res = send(socket, &((data.data())[pos]), data.length() - pos, 0);
+		if (res == -1)
+		{
+			if (errno == EAGAIN || errno == EWOULDBLOCK)
+			{
+				if (time(NULL) - start > TIMEOUT)
+				{
+					throw (string)("Timeout");
+				}
+				continue;
+			}
+			else
+			{
+				// сокет поломался - закрываем его
+				throw (string)("Socket closed");
+			}
+		}
+		else
+		{
+			pos += res;
+		}
+	}
+}
+
+string SynchronousSocket::Receive(int amount)
+{
+	string r = "";
+	time_t start = time(NULL);
+	while ((int)r.length() < amount)
+	{
+		char buf[BUFFER_SIZE];
+		int to_recv = min((int)sizeof(buf), amount - (int)r.length());
+		int res = recv(socket, buf, to_recv, 0);
+		if (res == 0)
+		{
+			// бросаем специфический exception, означающий, что сокет закрыт
+			throw (string)("Socket closed");
+		}
+		else if (res == -1)
+		{
+			if (errno == EAGAIN || errno == EWOULDBLOCK)
+			{
+				if (time(NULL) - start > TIMEOUT)
+				{
+					throw (string)("Timeout");
+				}
+				continue;
+			}
+			else
+			{
+				// что-то сломалось, надо закрыть сокет на всякий случай
+				// бросаем специфический exception, означающий, что сокет закрыт
+				throw (string)("Socket closed");
+			}
+		}
+		else
+		{
+			r += string(buf, res);
+		}
+	}
+	return r;
+}
+
+void SynchronousSocket::Disconnect()
+{
+	if (socket == -1) return;
+	close(socket);
+	socket = -1;
+}
+
+string SynchronousSocket::ReceiveMessage()
+{
+	string len_s = Receive(4);
+	int len = 0;
+	for (size_t i = 0; i < len_s.length(); i++)
+	{
+		len |= ((unsigned char)len_s[i]) << (8 * i);
+	}
+	return Receive(len);
+}
+
+void SynchronousSocket::SendMessage(const string & m)
+{
+	int len = m.length();
+	string len_s = "";
+	for (int i = 0; i < 4; i++)
+	{
+		len_s += (char)((len >> (8 * i)) & 0xFF);
+	}
+	Send(len_s);
+	Send(m);
+}
+
+

server/SynchronousSocket.h

@@ -0,0 +1,30 @@
+#ifndef SYNCHRONOUS_SOCKET_HEADER_FILE_INCLUDED
+#define SYNCHRONOUS_SOCKET_HEADER_FILE_INCLUDED
+
+#include <string>
+
+using namespace std;
+
+class SynchronousSocket
+{
+private:
+	const int TIMEOUT = 10; // in seconds
+	const int BUFFER_SIZE = 1048576;
+protected:
+	// собственно сокет
+	int socket;
+public:
+    SynchronousSocket();
+	// Используется для проброса уже готового сокета и работы с ним
+	void Init(int s);
+	bool IsConnected();
+	virtual void Connect(string host, int port);
+	virtual void Send(const string &data);
+	virtual string Receive(int amount);
+	void Disconnect();
+	string ReceiveMessage();
+	void SendMessage(const string &m);
+};
+
+
+#endif

server/mutex.cpp

@@ -0,0 +1,26 @@
+#include <string>
+
+#include "mutex.h"
+
+using namespace std;
+
+void LockMutex(pthread_mutex_t &m)
+{
+	int mres;
+	while ((mres = pthread_mutex_lock(&m)) != 0)
+	{
+		if (mres != EAGAIN)
+		{
+			throw (string)("Fatal error acquiring mutex");
+		}
+	}
+}
+
+void ReleaseMutex(pthread_mutex_t &m)
+{
+	int mres = pthread_mutex_unlock(&m);
+	if (mres != 0)
+	{
+		throw (string)("Error releasing mutex");
+	}
+}

server/mutex.h

@@ -0,0 +1,9 @@
+#ifndef MUTEX_H
+#define MUTEX_H
+
+#include <pthread.h>
+
+void LockMutex(pthread_mutex_t &m);
+void ReleaseMutex(pthread_mutex_t &m);
+
+#endif

server/task.cpp

@@ -0,0 +1,18 @@
+#include "task.h"
+#include <cstring>
+#include <sstream>
+
+using namespace std;
+
+string Task(const string &message)
+{                
+	string result = "";
+	int result_length = message.length() / 8;
+	if (result_length < 1) result_length = 1;
+	for (int i = 0; i < result_length; i++) result += '\x0';
+	for (int i = 0; i < (int)message.length(); i++)
+	{
+		result[i % result_length] ^= message[i];
+	}
+	return result;
+}

server/task.h

@@ -0,0 +1,8 @@
+#ifndef TASK_HEADER_FILE_INCLUDED
+#define TASK_HEADER_FILE_INCLUDED
+
+#include <string>
+
+std::string Task(const std::string &message);
+
+#endif

server/tcp_perf_test_server.cpp

@@ -0,0 +1,120 @@
+#include <netinet/in.h>
+#include <arpa/inet.h>
+#include <string>
+#include <cstring>
+#include <unistd.h>
+#include <sys/file.h>
+#include <errno.h>
+#include <iostream>
+#include <sys/un.h>
+#include <sstream>
+#include <signal.h>
+
+#include "SynchronousSocket.h"
+#include "task.h"
+
+using namespace std;
+
+const int LISTEN_PORT = 19876;
+const int LISTEN_BACKLOG = 10;
+const int EXIT_FAIL = 1;
+
+void CreateThread(void *(*start)(void *), void* arg = nullptr)
+{   
+	pthread_t thread;
+	pthread_attr_t attr;
+	if (pthread_attr_init(&attr) != 0)
+	{                    	
+		throw (string)("pthread_attr_init error");
+	}
+	if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED))
+	{
+		throw (string)("pthread_attr_setdetachstate error");
+	}                 
+	if (pthread_create(&thread, &attr, start, arg) != 0)
+	{                      
+		throw (string)("pthread_create error");
+	}                
+}
+
+void* ProcessRequests(void *arg)
+{
+	int socket_fd = *(int*)arg;
+	SynchronousSocket s;
+	s.Init(socket_fd);
+	try {
+		while (true)
+		{
+			string message = s.ReceiveMessage();
+			s.SendMessage(Task(message));
+		}
+	}
+	catch (string s)
+	{
+		cerr << "Error in socket " << s << "\n";
+	}
+	catch (...)
+	{
+		cerr << "Unknown error in socket\n";
+	}
+	s.Disconnect();
+}
+
+void* IPSocketThread(void* arg)
+{
+	int listensock;
+	if ((listensock = socket(PF_INET, SOCK_STREAM, 0)) == -1)
+	{
+		cerr << "Could not create socket for listening...\n";
+		exit(EXIT_FAIL);
+	}
+	struct sockaddr_in listenaddr;
+	memset(&listenaddr, 0, sizeof(listenaddr));
+	listenaddr.sin_family = AF_INET;
+	listenaddr.sin_addr.s_addr = htonl(INADDR_ANY);
+	listenaddr.sin_port = htons(LISTEN_PORT);
+	if (bind(listensock, (struct  sockaddr *)(&listenaddr), sizeof(struct sockaddr_in)) == -1)
+	{
+		cerr << "Could not bind listening socket...\n";
+		exit(EXIT_FAIL);
+	}
+	if (listen(listensock, LISTEN_BACKLOG) == -1)
+	{
+		cerr << "Could not set socket into listening mode...\n";
+		exit(EXIT_FAIL);
+	}
+	struct sockaddr_in peeraddr;
+	socklen_t paddrlen;
+	paddrlen = sizeof(peeraddr);
+	int recsock;
+	cout << "Accepting connections on port " << LISTEN_PORT << "\n";           
+	while (true)
+	{
+		(recsock = accept(listensock, (struct sockaddr *)(&peeraddr), &paddrlen));
+		if (recsock == -1)
+		{
+			cerr << "Error accepting connection, errno = " << errno << "\n";
+			continue;
+		}
+		cout << "Accepted connection from " << inet_ntoa(peeraddr.sin_addr) << ":" << ntohs(peeraddr.sin_port) << "\n";
+		CreateThread(&ProcessRequests, &recsock);
+	}                                                   
+	cerr << "Accept error...\n";
+	if (close(listensock) == -1)
+	{
+		cerr << "Could not close listening socket...\n";
+		exit(EXIT_FAIL);
+	}
+}
+
+int main()
+{
+	signal(SIGPIPE, SIG_IGN);
+    CreateThread(&IPSocketThread);
+	cout << "Press Ctrl-C for exit\n";
+	while (true)
+	{
+		sleep(10);
+	}
+	return 0;
+}