Added simple TCP socket server to handle requests and TCP socket client to test and measure.

client/Socket.php

@@ -0,0 +1,159 @@
+<?php
+
+class Socket
+{
+	const RECEIVE_TIMEOUT = 10;
+
+	private $sock;
+	private $throw_exception_on_timeout;
+	private $connected;
+	private $host;
+	private $port;
+
+	public function __construct($ip='', $port='', $throw_exception_on_timeout = false, $auto_connect = true)
+	{
+		$this->sock = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
+		if ($this->sock === FALSE) 
+		{
+			throw new Exception("Error creating socket: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = false;
+		$this->host = $ip;
+		$this->port = $port;
+		if ($auto_connect)
+		{
+			$this->connect();
+		}
+		$this->throw_exception_on_timeout = $throw_exception_on_timeout;
+	}
+
+	public function connect()
+	{
+		if ($this->connected) return;
+		// тушим ошибки, иначе вывалится с PHP Error вместо exception
+		if (@socket_connect($this->sock, $this->host, $this->port) === FALSE) 
+		{
+			throw new Exception("Error connecting: ".socket_last_error()." ".socket_strerror(socket_last_error()));
+		}
+		$this->connected = true;
+	}
+
+	/**
+	 * Закрытие соединения
+	 */
+	public function close()
+	{
+		if ($this->connected) 
+		{
+			socket_close($this->sock);
+			$this->connected = false;
+		}
+	}
+
+	/*
+	 * Пропихивает строку в сокет. Бросит exception если случилась лажа.
+	 */
+
+	public function iterativeSend($s)
+	{
+		while ($s != "") 
+		{
+			$cnt = @socket_send($this->sock, $s, strlen($s), 0);
+			if ($cnt === FALSE) throw new Exception("Sending error: " . socket_strerror(socket_last_error()));
+			$s = substr($s, $cnt);
+		}
+	}
+
+	/*
+	 * Притаскивает из сокета заданное число байт
+	 */
+
+	public function iterativeReceive($cnt, $timeout = 0)
+	{
+		$s = "";
+		$start = microtime(true);
+		while (strlen($s) != $cnt) 
+		{
+			$buf = "";
+			if (socket_recv($this->sock, $buf, $cnt - strlen($s), MSG_DONTWAIT) === FALSE) 
+			{
+				if (socket_last_error() != 11) 
+				{
+					throw new Exception("Receiving error: " . socket_strerror(socket_last_error()));
+				}
+			}
+			if (strlen($buf) == 0) usleep(10 * 1000);
+			$s .= $buf;
+			if ($timeout > 0 && microtime(true) - $start > $timeout) 
+			{
+				if ($this->throw_exception_on_timeout) 
+				{
+					throw new \Exception("Timeout on receiving, wanted " . $cnt . " received so far " . strlen($s));
+				} 
+				else 
+				{
+					return $s;
+				}
+			}
+		}
+		return $s;
+	}
+
+	/*
+	 * Тянет из сокета данные, пока не вытянет заданную строку
+	 */
+
+	public function receiveTerminated($t, $timeout = 0)
+	{
+		$s = "";
+		$start = microtime(true);
+		while (true) 
+		{
+			$s .= $this->iterativeReceive(1, $timeout);
+			if (strlen($t) <= strlen($s) && substr($s, strlen($s) - strlen($t)) == $t) break;
+			if ($timeout > 0 && microtime(true) - $start > $timeout) 
+			{
+				if ($this->throw_exception_on_timeout) 
+				{
+					throw new \Exception("Timeout on terminated receiving, received so far " . strlen($s));
+				} 
+				else 
+				{
+					return $s;
+				}
+			}
+		}
+		return $s;
+	}
+
+
+	/*
+	 * Отправляет строку в сокет согласно протоколу (4 байта длина, потом данные)
+	 */
+
+	public function sendMessage($s)
+	{
+		$len = strlen($s);
+		$lens = chr(($len >> 0) & 0xFF) . chr(($len >> 8) & 0xFF) . chr(($len >> 16) & 0xFF) . chr(($len >> 24) & 0xFF);
+		$this->iterativeSend($lens);
+		$this->iterativeSend($s);
+	}
+
+	/*
+	 * Принимает строку из сокета. Бросит exception если что-то не так
+	 */
+
+	public function receiveMessage()
+	{
+		$lens = $this->iterativeReceive(4, self::RECEIVE_TIMEOUT);
+		if (strlen($lens) < 4) {
+			throw new \Exception("Too short length string received");
+		}
+		$len = ord($lens[0]) << 0;
+		$len += ord($lens[1]) << 8;
+		$len += ord($lens[2]) << 16;
+		$len += ord($lens[3]) << 24;
+		return $this->iterativeReceive($len, 1);
+	}
+
+}

client/test_client.php

@@ -0,0 +1,43 @@
+<?php
+
+require_once('Socket.php');
+
+function makeRequest($size)
+{
+}
+
+function testSocket($amount, $reconnect, $size)
+{
+	$s = null;
+	$current = 0;
+	for ($i = 0; $i < $amount; $i++)
+	{
+		if ($s === null)
+		{
+			$s = new Socket('127.0.0.1', 19876);
+			$s->connect();
+			$current = 0;
+		}
+		$m = makeRequest($size);
+		$s->sendMessage($m);
+		$r = $s->receiveMessage();
+		$current++;
+		if ($current >= $reconnect)
+		{
+			$s = null;
+		}
+	}
+}
+
+$requests_amount = $argv[1];
+$requests_per_connect = $argv[2];
+$request_size = $argv[3];
+
+print('Testing '.$requests_amount.' requests, reconnect after '.
+		$requests_per_connect.' size '.$request_size."\n");
+$start = microtime(true);
+testSocket($requests_amount, $requests_per_connect, $request_size);
+$finish = microtime(true);
+
+$elapsed = $finish - $start;
+print('Total: '.round($elapsed, 3).' seconds, '.round($requests_amount / $elapsed, 3).' RPS'."\n");

server/.gitignore

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

server/CMakeLists.txt

@@ -0,0 +1,23 @@
+cmake_minimum_required(VERSION 3.0)
+project(ipc_test_server)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+set(CMAKE_MODULE_PATH "${CMAKE_SOURCE_DIR}/../cmake")
+
+## SOURCES  ####################################################################
+set(SOURCE_FILES
+
+ipc_server.cpp
+
+task.h
+task.cpp
+
+SynchronousSocket.cpp
+SynchronousSocket.h
+)
+##################################################################################
+
+link_libraries(stdc++)
+
+add_executable(ipc_test_server ${SOURCE_FILES} ./ipc_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[1024];
+		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
+		{
+			for (int i = 0; i < res; i++) r += buf[i];
+		}
+	}
+	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,29 @@
+#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
+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/ipc_server.cpp

@@ -0,0 +1,88 @@
+#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 "SynchronousSocket.h"
+#include "task.h"
+
+using namespace std;
+
+const int LISTEN_PORT = 19876;
+const int LISTEN_BACKLOG = 10;
+const int EXIT_FAIL = 1;
+
+void ProcessRequests(int socket_fd)
+{
+	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();
+}
+
+int main()
+{
+	int listensock;
+	if ((listensock = socket(PF_INET, SOCK_STREAM, 0)) == -1)
+	{
+		cerr << "Could not create socket for listening...\n";
+		return 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";
+		return EXIT_FAIL;
+	}
+	if (listen(listensock, LISTEN_BACKLOG) == -1)
+	{
+		cerr << "Could not set socket into listening mode...\n";
+		return EXIT_FAIL;
+	}
+	struct sockaddr_in peeraddr;
+	socklen_t paddrlen;
+	paddrlen = sizeof(peeraddr);
+	int recsock;
+	cout << "Press Ctrl-C for exit\n";
+	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";
+		ProcessRequests(recsock);
+	}                                                   
+	cerr << "Accept error...\n";
+	if (close(listensock) == -1)
+	{
+		cerr << "Could not close listening socket...\n";
+		exit(EXIT_FAIL);
+	}
+	return 0;
+}

server/task.cpp

@@ -0,0 +1,23 @@
+#include "task.h"
+#include <cstring>
+#include <sstream>
+
+using namespace std;
+
+int cnt[256];
+
+string Task(const string &message)
+{                
+	memset(cnt, 0, sizeof(cnt));
+	for (int i = 0; i < (int)message.length(); i++)
+	{
+		cnt[(int)(unsigned char)message[i]]++;
+	}
+	ostringstream result;
+	for (int i = 0; i < 256; i++)
+	{
+		if (cnt[i] == 0) continue;
+		result << i << ":" << cnt[i] << ";";
+	}
+	return result.str();
+}

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