From 4c52d620c56a740800c83585f31b7fea4bfbd870 Mon Sep 17 00:00:00 2001 From: Denis Denisov Date: Tue, 14 Mar 2023 20:56:59 +0300 Subject: [PATCH] Initial commit. Basic test suite --- LICENSE | 13 +++ README.md | 6 + client/Socket.php | 197 ++++++++++++++++++++++++++++++++ client/test_client.php | 50 ++++++++ server/.gitignore | 1 + server/CMakeLists.txt | 28 +++++ server/SynchronousSocket.cpp | 161 ++++++++++++++++++++++++++ server/SynchronousSocket.h | 30 +++++ server/mutex.cpp | 26 +++++ server/mutex.h | 9 ++ server/task.cpp | 18 +++ server/task.h | 8 ++ server/tcp_perf_test_server.cpp | 120 +++++++++++++++++++ 13 files changed, 667 insertions(+) create mode 100644 LICENSE create mode 100644 README.md 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/mutex.cpp create mode 100644 server/mutex.h create mode 100644 server/task.cpp create mode 100644 server/task.h create mode 100644 server/tcp_perf_test_server.cpp diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..9affcdd --- /dev/null +++ b/LICENSE @@ -0,0 +1,13 @@ +Copyright 2023 Denis "ftc" Denisov + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..4de1ba8 --- /dev/null +++ b/README.md @@ -0,0 +1,6 @@ +This is a test project written to compare methods of +communicating between PHP and C++ code. + +It contains a server, written in C++ and client, written +in PHP. The client sends several requests to server via +all implemented mechanisms and measures time taken. diff --git a/client/Socket.php b/client/Socket.php new file mode 100644 index 0000000..39e37ab --- /dev/null +++ b/client/Socket.php @@ -0,0 +1,197 @@ +connected) + { + socket_close($this->sock); + $this->connected = false; + } + } + + /* + * Пропихивает строку в сокет. + */ + + 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); + } + +} + +class IPSocket extends Socket +{ + 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; + } +} + +class UnixSocket extends Socket +{ + private $path; + + public function __construct($path='', $throw_exception_on_timeout = false, $auto_connect = true) + { + $this->sock = socket_create(AF_UNIX, SOCK_STREAM, 0); + if ($this->sock === FALSE) + { + throw new Exception("Error creating socket: ".socket_last_error()." ".socket_strerror(socket_last_error())); + } + $this->connected = false; + $this->path = $path; + 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->path) === FALSE) + { + throw new Exception("Error connecting: ".socket_last_error()." ".socket_strerror(socket_last_error())); + } + $this->connected = true; + } +} \ No newline at end of file diff --git a/client/test_client.php b/client/test_client.php new file mode 100644 index 0000000..9a7b08a --- /dev/null +++ b/client/test_client.php @@ -0,0 +1,50 @@ +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"); + + +print("Testing IP socket:\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..e4871c0 --- /dev/null +++ b/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) + diff --git a/server/SynchronousSocket.cpp b/server/SynchronousSocket.cpp new file mode 100644 index 0000000..4b32552 --- /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[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); +} + + diff --git a/server/SynchronousSocket.h b/server/SynchronousSocket.h new file mode 100644 index 0000000..6fcaee4 --- /dev/null +++ b/server/SynchronousSocket.h @@ -0,0 +1,30 @@ +#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 + 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 diff --git a/server/mutex.cpp b/server/mutex.cpp new file mode 100644 index 0000000..7904686 --- /dev/null +++ b/server/mutex.cpp @@ -0,0 +1,26 @@ +#include + +#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"); + } +} diff --git a/server/mutex.h b/server/mutex.h new file mode 100644 index 0000000..a302cf9 --- /dev/null +++ b/server/mutex.h @@ -0,0 +1,9 @@ +#ifndef MUTEX_H +#define MUTEX_H + +#include + +void LockMutex(pthread_mutex_t &m); +void ReleaseMutex(pthread_mutex_t &m); + +#endif diff --git a/server/task.cpp b/server/task.cpp new file mode 100644 index 0000000..e09a8a0 --- /dev/null +++ b/server/task.cpp @@ -0,0 +1,18 @@ +#include "task.h" +#include +#include + +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; +} 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 diff --git a/server/tcp_perf_test_server.cpp b/server/tcp_perf_test_server.cpp new file mode 100644 index 0000000..8ebf493 --- /dev/null +++ b/server/tcp_perf_test_server.cpp @@ -0,0 +1,120 @@ +#include +#include +#include +#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 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; +}