From 59589e4017aa420cf70e3dc96b29210fd1f72817 Mon Sep 17 00:00:00 2001 From: Denis Denisov Date: Sun, 26 Feb 2023 16:12:04 +0300 Subject: [PATCH] Added SharedMemoryChannel and its comparison with other methods --- client/SharedMemoryChannel.php | 80 +++++++++ client/test_client.php | 41 +++++ server/CMakeLists.txt | 5 + server/SharedMemoryChannel.cpp | 319 +++++++++++++++++++++++++++++++++ server/SharedMemoryChannel.h | 88 +++++++++ server/ipc_server.cpp | 134 ++++++++++---- server/mutex.cpp | 26 +++ server/mutex.h | 9 + 8 files changed, 668 insertions(+), 34 deletions(-) create mode 100644 client/SharedMemoryChannel.php create mode 100644 server/SharedMemoryChannel.cpp create mode 100644 server/SharedMemoryChannel.h create mode 100644 server/mutex.cpp create mode 100644 server/mutex.h diff --git a/client/SharedMemoryChannel.php b/client/SharedMemoryChannel.php new file mode 100644 index 0000000..a1f4cc1 --- /dev/null +++ b/client/SharedMemoryChannel.php @@ -0,0 +1,80 @@ +semaphore_send = posix_sem_open($args['send'], 0); + if ($this->semaphore_send === null) { + throw new \Exception("Unable to open send semaphore"); + } + $this->semaphore_receive = posix_sem_open($args['receive'], 0); + if ($this->semaphore_receive === null) { + throw new \Exception("Unable to open receive semaphore"); + } + $this->shared_mem = shmop_open($args['mem'], 'w', 0, 0); + } + + // Предполагалось в деструкторе очищать семаформы и расшаренную память, + // однако это чревато неприятными кейсами, когда мы удаляем семафор, который + // к моменту удаления уже переиспользован. Потому управление этими + // ресурсами - прерогатива исключительно "той" стороны. + // Однако закрыть семафоры и shared-mem было бы неплохо, + // ибо иначе при реконнектах эти ресурсы могут "зависнуть". + public function __destruct() + { + $this->close(); + } + + public function close() + { + if ($this->semaphore_send !== null) { + posix_sem_close($this->semaphore_send); + $this->semaphore_send = null; + } + if ($this->semaphore_receive !== null) { + posix_sem_close($this->semaphore_receive); + $this->semaphore_receive = null; + } + if ($this->shared_mem !== null) { + shmop_close($this->shared_mem); + $this->shared_mem = null; + } + } + + public function sendMessage($s) + { + $len = strlen($s); + if ($len > self::MAX_LENGTH) throw new \Exception("Too long message"); + $lens = chr(($len >> 0) & 0xFF) . chr(($len >> 8) & 0xFF) . chr(($len >> 16) & 0xFF) . chr(($len >> 24) & 0xFF); + shmop_write($this->shared_mem, $lens, 0); + shmop_write($this->shared_mem, $s, 4); + posix_sem_post($this->semaphore_send); + } + + public function receiveMessage() + { + $start = time(); + while (time() < $start + self::TIMEOUT) { + if (posix_sem_timedwait($this->semaphore_receive, time() + 10, 0) === true) { + $lens = shmop_read($this->shared_mem, 0, 4); + $len = ord($lens[0]) << 0; + $len += ord($lens[1]) << 8; + $len += ord($lens[2]) << 16; + $len += ord($lens[3]) << 24; + $s = shmop_read($this->shared_mem, 4, $len); + return $s; + } + } + throw new \Exception("Timeout reading from SharedMemoryChannel"); + } + +} \ No newline at end of file diff --git a/client/test_client.php b/client/test_client.php index dc72dfb..a1cc4b0 100644 --- a/client/test_client.php +++ b/client/test_client.php @@ -1,6 +1,7 @@ connect(); + $s->sendMessage("mem"); + $result = $s->receiveMessage(); + $data = explode(":", $result); + $channel = new SharedMemoryChannel([ + 'send' => $data[0], + 'receive' => $data[1], + 'mem' => $data[2] + ]); + $current = 0; + } + $m = makeRequest($size); + $channel->sendMessage($m); + $r = $channel->receiveMessage(); + $current++; + if ($current >= $reconnect) + { + $s = null; + } + } +} + $requests_amount = $argv[1]; $requests_per_connect = $argv[2]; $request_size = $argv[3]; @@ -78,3 +111,11 @@ $finish = microtime(true); $elapsed = $finish - $start; print('Total: '.round($elapsed, 3).' seconds, '.round($requests_amount / $elapsed, 3).' RPS'."\n"); + +print("Testing shared memory channel:\n"); +$start = microtime(true); +testSharedMemory($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/CMakeLists.txt b/server/CMakeLists.txt index 4e9702f..89587fb 100644 --- a/server/CMakeLists.txt +++ b/server/CMakeLists.txt @@ -14,8 +14,13 @@ ipc_server.cpp task.h task.cpp +mutex.h +mutex.cpp + SynchronousSocket.cpp SynchronousSocket.h +SharedMemoryChannel.cpp +SharedMemoryChannel.h ) ################################################################################## diff --git a/server/SharedMemoryChannel.cpp b/server/SharedMemoryChannel.cpp new file mode 100644 index 0000000..6f55129 --- /dev/null +++ b/server/SharedMemoryChannel.cpp @@ -0,0 +1,319 @@ +#include "SharedMemoryChannel.h" +#include "mutex.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace std; + +bool operator<(const SharedMemoryChannelID &a, const SharedMemoryChannelID &b) +{ + if (a.reuse_at < b.reuse_at) return true; + if (a.reuse_at > b.reuse_at) return false; + return a.internal_id < b.internal_id; +} + +SharedMemoryChannel::SharedMemoryChannel() +{ + semaphore_send = semaphore_receive = nullptr; + shared_mem = -1; + shared = nullptr; + id = id_provider.GetID(); + cerr << "SharedMemoryChannel::GetID " << id.id << "\n"; + { + ostringstream oss; + oss << "/smc_send_" << id.id; + semaphore_send_key = oss.str(); + } + { + ostringstream oss; + oss << "/smc_receive" << id.id; + semaphore_receive_key = oss.str(); + } + char buf[2048]; + memset(buf, 0, sizeof(buf)); + sprintf(buf, "/tmp/smc_%s", id.id.c_str()); + auto t = fopen(buf, "w"); + if (t == nullptr) + { + id_provider.ReleaseID(id); + throw (string)("Unable to create memory channel IPC file"); + } + fclose(t); + sem_unlink(semaphore_receive_key.c_str()); + semaphore_receive = sem_open(semaphore_receive_key.c_str(), O_CREAT, 0666, 1); + if (semaphore_receive == SEM_FAILED) + { + id_provider.ReleaseID(id); + ostringstream oss; + oss << "Unable to init receive semaphore " << errno << " key = " << semaphore_receive_key; + throw oss.str(); + } + sem_unlink(semaphore_send_key.c_str()); + semaphore_send = sem_open(semaphore_send_key.c_str(), O_CREAT, 0666, 1); + if (semaphore_send == SEM_FAILED) + { + id_provider.ReleaseID(id); + ostringstream oss; + oss << "Unable to init send semaphore " << errno << " key = " << semaphore_send_key; + throw oss.str(); + } + // now create shared memory block + shared_mem_key = ftok(buf, 2); + shared_mem = shmget(shared_mem_key, MAX_LEN + 4, IPC_CREAT | 0777); + if (shared_mem == -1) + { + if (semaphore_receive != nullptr) + { + sem_close(semaphore_receive); + } + if (semaphore_send != nullptr) + { + sem_close(semaphore_send); + } + id_provider.ReleaseID(id); + ostringstream oss; + oss << "Unable to get shared mem segment, error = " << errno; + throw oss.str(); + } + shared = shmat(shared_mem, nullptr, 0); + if (shared == (void*)-1) + { + if (semaphore_receive != nullptr) + { + sem_close(semaphore_receive); + } + if (semaphore_send != nullptr) + { + sem_close(semaphore_send); + } + id_provider.ReleaseID(id); + throw (string)("Failed to attach to shared memory"); + } + // acquire both semaphores + // our receive semaphore will be released by client + try { + AcquireSemaphore(semaphore_send); + AcquireSemaphore(semaphore_receive); + } + catch (...){ + shmctl(shared_mem, IPC_RMID, nullptr); + throw; + } +} + +SharedMemoryChannel::~SharedMemoryChannel() +{ + cerr << "SharedMemoryChannel::ReleaseID " << id.id << "\n"; + if (semaphore_receive != nullptr) + { + sem_close(semaphore_receive); + } + if (semaphore_send != nullptr) + { + sem_close(semaphore_send); + } + if (shared_mem != -1) + { + if (shared != nullptr) + { + shmdt(shared); + } + shmctl(shared_mem, IPC_RMID, nullptr); + } + id_provider.ReleaseID(id); +} + +string SharedMemoryChannel::GetSendKey() const +{ + return semaphore_send_key; +} + +string SharedMemoryChannel::GetReceiveKey() const +{ + return semaphore_receive_key; +} + +key_t SharedMemoryChannel::GetSharedMemKey() const +{ + return shared_mem_key; +} + +void SharedMemoryChannel::AcquireSemaphore(sem_t* semaphore) +{ + struct timespec timeout; + auto start = time(NULL); + while (true) + { + clock_gettime(CLOCK_REALTIME, &timeout); + timeout.tv_sec += TIMEOUT_MSEC / 1000; + timeout.tv_nsec += (TIMEOUT_MSEC % 1000) * 1000000; + while (timeout.tv_nsec >= 1000000000) + { + timeout.tv_sec++; + timeout.tv_nsec -= 1000000000; + } + if (sem_timedwait(semaphore, &timeout) == -1) + { + if (errno != EAGAIN && errno != ETIMEDOUT) + { + ostringstream oss; + oss << "AcquireSemaphore failed " << errno; + throw oss.str(); + } + } + else + { + // semaphore operation went well + return; + } + // check timeout + if (time(NULL) - start > CHANNEL_CLEAR_TIMEOUT_SEC) + { + throw (string)("Channel inactive, remove it"); + } + } +} + +void SharedMemoryChannel::ReleaseSemaphore(sem_t* semaphore) +{ + if (sem_post(semaphore) == -1) + { + ostringstream oss; + oss << "ReleaseSemaphore failed " << errno; + throw oss.str(); + } +} + +std::string SharedMemoryChannel::Receive() +{ + AcquireSemaphore(semaphore_receive); + char *sc = (char*)shared; + int len = 0; + for (int i = 3; i >= 0; i--) + { + len = (len << 8) | ((int)(unsigned char)sc[i]); + } + return string(&sc[4], len); +} + +void SharedMemoryChannel::Send(const string &s) +{ + if (s.length() > MAX_LEN) throw (string)("Too long message to pass through shared mem channel"); + char *sc = (char*)shared; + int len = s.length(); + for (int i = 0; i < 4; i++) + { + sc[i] = len & 0xFF; + len >>= 8; + } + memcpy(&sc[4], s.c_str(), s.length()); + ReleaseSemaphore(semaphore_send); +} + +SharedMemoryChannelIDProvider::SharedMemoryChannelIDProvider() +{ + if (pthread_mutex_init(&access_mutex, NULL) != 0) + { + throw (string)("Unable to create access mutex"); + } + FILE *f = fopen("id", "r"); + string id = ""; + if (f == nullptr) + { + for (int i = 0; i < 30; i++) + { + id += (char)(rand() % 26 + 'A'); + } + } + else + { + int c = fgetc(f); + while (c != -1) + { + if (c >= 'A' && c <= 'Z') + { + id += (char)c; + } + c = fgetc(f); + } + fclose(f); + } + + // clear all possible semaphores (in order to make sure that everyone + // waiting for them will be woken up) + DIR* dir = opendir("/dev/shm"); + if (!dir) + { + throw (string)("Could not open semaphores directory"); + } + struct dirent cdir; + struct dirent *p; + if (readdir_r(dir, &cdir, &p)) + { + throw (string)("Error during reading semaphores directory"); + } + string to_check = "smc" + id; + vector to_unlink; + while (p) + { + string cname(cdir.d_name); + if (cname.substr(0, to_check.length()) == to_check) + { + to_unlink.push_back("/dev/shm/" + cname); + } + if (readdir_r(dir, &cdir, &p)) + { + throw (string)("Error during reading semaphores directory"); + } + } + closedir(dir); + for (int i = 0; i < (int)to_unlink.size(); i++) + { + unlink(to_unlink[i].c_str()); + } +} + +SharedMemoryChannelIDProvider SharedMemoryChannel::id_provider; + +SharedMemoryChannelID SharedMemoryChannelIDProvider::GetID() +{ + LockMutex(access_mutex); + if (used_ids.size() > 0) + { + auto t = used_ids.begin(); + if (t->reuse_at <= time(nullptr)) + { + auto result = *t; + used_ids.erase(t); + ReleaseMutex(access_mutex); + return result; + } + } + int id = last_id++; + ReleaseMutex(access_mutex); + ostringstream oss; + oss << prefix << "_" << id; + SharedMemoryChannelID result; + result.internal_id = id; + result.id = oss.str(); + return result; +} + +void SharedMemoryChannelIDProvider::ReleaseID(const SharedMemoryChannelID &id) +{ + LockMutex(access_mutex); + auto t = id; + t.reuse_at = time(nullptr) + ID_REUSE_TIMEOUT; + used_ids.insert(t); + ReleaseMutex(access_mutex); +} diff --git a/server/SharedMemoryChannel.h b/server/SharedMemoryChannel.h new file mode 100644 index 0000000..242cde0 --- /dev/null +++ b/server/SharedMemoryChannel.h @@ -0,0 +1,88 @@ +#ifndef SHAREDMEMORYCHANNEL_H +#define SHAREDMEMORYCHANNEL_H + +#include +#include +#include +#include +#include +#include + +class SharedMemoryChannelID +{ +public: + time_t reuse_at; + int internal_id; + std::string id; +}; + +bool operator<(const SharedMemoryChannelID &a, const SharedMemoryChannelID &b); + +class SharedMemoryChannelIDProvider +{ + // Таймаут, после которого считается, что id канала безопасно переиспользовать. + // В секундах. + const int ID_REUSE_TIMEOUT = 60; + + int last_id; + pthread_mutex_t access_mutex; + std::string prefix; + + std::set used_ids; +public: + SharedMemoryChannelIDProvider(); + SharedMemoryChannelID GetID(); + void ReleaseID(const SharedMemoryChannelID &id); +}; + +/* + * Обмен с PHP-клиентом через shared memory. + * Управляется двумя семафорами. Общий цикл выглядит так: + * 1) Оба семафора залочены, мы ждем semaphore_receive. + * 2) Клиент кладет в shared mem сообщение, разлочивает semaphore_receive. + * 3) Мы получаем semaphore_receive, читаем сообщение. Клиент ждет semaphore_send. + * 4) Разлочим semaphore_send, клиент лочит semaphore_send и получает сообщение, + * + * Итого общий принцип - при получении сообщения получатель + * лочит семафор, ответственный за передачу в его сторону. И + * не разлочивает его, задача разлочивать - на отправителе. + * Отправитель соответственно когда готова отправка разлочивает + * семафор передачи "от него". +*/ +class SharedMemoryChannel +{ + const static int TIMEOUT_MSEC = 100; + // через вот такое количество времени считаем, что канал уже не используется и его можно прибивать + const static int CHANNEL_CLEAR_TIMEOUT_SEC = 10; + // Храним использованные id-шники + // с тем, чтобы разные каналы использовали разные + // id, но при этом набор id-шников оставался + // бы неизменным при повторных запусках. + // Это чинит неприятную проблему, связанную с + // прибиванием процесса, который пользуется семафорами. + static SharedMemoryChannelIDProvider id_provider; + // 4 bytes for length + key_t shared_mem_key; + int shared_mem; + void *shared; + std::string semaphore_send_key; + sem_t* semaphore_send; + std::string semaphore_receive_key; + sem_t* semaphore_receive; + // + void AcquireSemaphore(sem_t *semaphore); + void ReleaseSemaphore(sem_t *semaphore); + // + SharedMemoryChannelID id; +public: + const static int MAX_LEN = 1048076 - 4; + std::string GetSendKey() const; + std::string GetReceiveKey() const; + key_t GetSharedMemKey() const; + SharedMemoryChannel(); + ~SharedMemoryChannel(); + std::string Receive(); + void Send(const std::string &s); +}; + +#endif // SHAREDMEMORYCHANNEL_H diff --git a/server/ipc_server.cpp b/server/ipc_server.cpp index 845ef9d..9e4b187 100644 --- a/server/ipc_server.cpp +++ b/server/ipc_server.cpp @@ -7,7 +7,10 @@ #include #include #include +#include +#include +#include "SharedMemoryChannel.h" #include "SynchronousSocket.h" #include "task.h" @@ -18,9 +21,29 @@ const int LISTEN_BACKLOG = 10; const int EXIT_FAIL = 1; const char* UNIX_SOCKET_PATH = "/tmp/ipc_test_socket"; +const char* SHMEM_SOCKET_PATH = "/tmp/ipc_test_socket_shmem"; -void ProcessRequests(int socket_fd) +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 { @@ -78,7 +101,7 @@ void* IPSocketThread(void* arg) continue; } cout << "Accepted connection from " << inet_ntoa(peeraddr.sin_addr) << ":" << ntohs(peeraddr.sin_port) << "\n"; - ProcessRequests(recsock); + CreateThread(&ProcessRequests, &recsock); } cerr << "Accept error...\n"; if (close(listensock) == -1) @@ -126,7 +149,7 @@ void* UnixSocketThread(void* arg) continue; } cout << "Accepted connection from " << peeraddr.sun_path << "\n"; - ProcessRequests(recsock); + CreateThread(&ProcessRequests, &recsock); } cerr << "Accept error...\n"; if (close(listensock) == -1) @@ -136,46 +159,89 @@ void* UnixSocketThread(void* arg) } } -void CreateSocketThread(void *(*start)(void *)) -{ - pthread_t thread; - pthread_attr_t attr; - if (pthread_attr_init(&attr) != 0) - { - throw (string)("pthread_attr_init error"); +void* ProcessShmemRequests(void *arg) +{ + int socket_fd = *(int*)arg; + SynchronousSocket s; + s.Init(socket_fd); + try { + string initial = s.ReceiveMessage(); + SharedMemoryChannel channel; + ostringstream result; + result << channel.GetReceiveKey() << ":" << + channel.GetSendKey() << ":" << + channel.GetSharedMemKey(); + s.SendMessage(result.str()); + while (true) + { + string message = channel.Receive(); + channel.Send(Task(message)); + } } - if (pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED)) + catch (string s) { - throw (string)("pthread_attr_setdetachstate error"); - } - if (pthread_create(&thread, &attr, start, nullptr) != 0) - { - throw (string)("pthread_create error"); - } + cerr << "Error in socket " << s << "\n"; + } + catch (...) + { + cerr << "Unknown error in socket\n"; + } + s.Disconnect(); } -void CreateUnixSocket() -{ - 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)) +void* SharedMemoryChannelSocketThread(void* arg) +{ + int listensock; + if ((listensock = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { - throw (string)("pthread_attr_setdetachstate error"); - } - if (pthread_create(&thread, &attr, &IPSocketThread, nullptr) != 0) - { - throw (string)("pthread_create error"); - } + cerr << "Could not create socket for listening...\n"; + exit(EXIT_FAIL); + } + struct sockaddr_un listenaddr; + listenaddr.sun_family = AF_UNIX; + strcpy(listenaddr.sun_path, SHMEM_SOCKET_PATH); + unlink(listenaddr.sun_path); + socklen_t len = strlen(listenaddr.sun_path) + sizeof(listenaddr.sun_family); + if (bind(listensock, (struct sockaddr*)&listenaddr, len) != 0) + { + cerr << "Error on binding socket \n"; + perror("bind"); + exit(EXIT_FAIL); + } + if (listen(listensock, LISTEN_BACKLOG) == -1) + { + cerr << "Could not set socket into listening mode...\n"; + exit(EXIT_FAIL); + } + struct sockaddr_un peeraddr; + socklen_t paddrlen; + paddrlen = sizeof(peeraddr); + int recsock; + cout << "Accepting connections on path " << SHMEM_SOCKET_PATH << "\n"; + while (true) + { + (recsock = accept(listensock, (struct sockaddr *)(&peeraddr), &paddrlen)); + if (recsock == -1) + { + cerr << "Error accepting connection, errno = " << errno << "\n"; + continue; + } + CreateThread(&ProcessShmemRequests, &recsock); + } + cerr << "Accept error...\n"; + if (close(listensock) == -1) + { + cerr << "Could not close listening socket...\n"; + exit(EXIT_FAIL); + } } int main() { - CreateSocketThread(&IPSocketThread); - CreateSocketThread(&UnixSocketThread); + signal(SIGPIPE, SIG_IGN); + CreateThread(&IPSocketThread); + CreateThread(&UnixSocketThread); + CreateThread(&SharedMemoryChannelSocketThread); cout << "Press Ctrl-C for exit\n"; while (true) { 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