From 09457cb6be39794b247fc5fb346c1ebd4e2ab6bc Mon Sep 17 00:00:00 2001 From: Denis Denisov Date: Tue, 14 Feb 2023 22:48:18 +0300 Subject: [PATCH] Added simple TCP socket server to handle requests and TCP socket client to test and measure. --- client/Socket.php | 159 ++++++++++++++++++++++++++++++++++ client/test_client.php | 43 ++++++++++ server/.gitignore | 1 + server/CMakeLists.txt | 23 +++++ server/SynchronousSocket.cpp | 161 +++++++++++++++++++++++++++++++++++ server/SynchronousSocket.h | 29 +++++++ server/ipc_server.cpp | 88 +++++++++++++++++++ server/task.cpp | 23 +++++ server/task.h | 8 ++ 9 files changed, 535 insertions(+) create mode 100644 client/Socket.php create mode 100644 client/test_client.php create mode 100644 server/.gitignore create mode 100644 server/CMakeLists.txt create mode 100644 server/SynchronousSocket.cpp create mode 100644 server/SynchronousSocket.h create mode 100644 server/ipc_server.cpp create mode 100644 server/task.cpp create mode 100644 server/task.h diff --git a/client/Socket.php b/client/Socket.php new file mode 100644 index 0000000..a2b1891 --- /dev/null +++ b/client/Socket.php @@ -0,0 +1,159 @@ +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); + } + +} \ No newline at end of file diff --git a/client/test_client.php b/client/test_client.php new file mode 100644 index 0000000..3518b98 --- /dev/null +++ b/client/test_client.php @@ -0,0 +1,43 @@ +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"); diff --git a/server/.gitignore b/server/.gitignore new file mode 100644 index 0000000..567609b --- /dev/null +++ b/server/.gitignore @@ -0,0 +1 @@ +build/ diff --git a/server/CMakeLists.txt b/server/CMakeLists.txt new file mode 100644 index 0000000..c0b5901 --- /dev/null +++ b/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) + diff --git a/server/SynchronousSocket.cpp b/server/SynchronousSocket.cpp new file mode 100644 index 0000000..c51653a --- /dev/null +++ b/server/SynchronousSocket.cpp @@ -0,0 +1,161 @@ +#include "SynchronousSocket.h" + +#include +#include +#include +#include +#include +#include + +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); +} + + diff --git a/server/SynchronousSocket.h b/server/SynchronousSocket.h new file mode 100644 index 0000000..8fd126e --- /dev/null +++ b/server/SynchronousSocket.h @@ -0,0 +1,29 @@ +#ifndef SYNCHRONOUS_SOCKET_HEADER_FILE_INCLUDED +#define SYNCHRONOUS_SOCKET_HEADER_FILE_INCLUDED + +#include + +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 diff --git a/server/ipc_server.cpp b/server/ipc_server.cpp new file mode 100644 index 0000000..5e7c11f --- /dev/null +++ b/server/ipc_server.cpp @@ -0,0 +1,88 @@ +#include +#include +#include +#include +#include +#include +#include +#include + +#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; +} diff --git a/server/task.cpp b/server/task.cpp new file mode 100644 index 0000000..71b7f40 --- /dev/null +++ b/server/task.cpp @@ -0,0 +1,23 @@ +#include "task.h" +#include +#include + +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(); +} diff --git a/server/task.h b/server/task.h new file mode 100644 index 0000000..6dee665 --- /dev/null +++ b/server/task.h @@ -0,0 +1,8 @@ +#ifndef TASK_HEADER_FILE_INCLUDED +#define TASK_HEADER_FILE_INCLUDED + +#include + +std::string Task(const std::string &message); + +#endif \ No newline at end of file