Дз 2, Костя Недиков, udp client для калькулятора

udp/client/CMakeLists.txt

@@ -0,0 +1,17 @@
+cmake_minimum_required(VERSION 2.8)
+project(calculator)
+set(CMAKE_CXX_STANDARD 17)
+
+file(GLOB SOURCE_FILES "src/*.cpp")
+
+add_executable(calculator ${SOURCE_FILES})
+
+target_include_directories(calculator PRIVATE include)
+
+find_package(Threads REQUIRED)
+if(THREADS_HAVE_PTHREAD_ARG)
+    target_compile_options(calculator PUBLIC "-pthread")
+endif()
+if(CMAKE_THREAD_LIBS_INIT)
+    target_link_libraries(calculator "${CMAKE_THREAD_LIBS_INIT}")
+endif()

udp/client/include/CalcuatorServerDriver.hpp

@@ -0,0 +1,56 @@
+#include <utility>
+
+#pragma once
+
+#include <string>
+#include <queue>
+#include <thread>
+#include <mutex>
+#include <unordered_set>
+#include "requests.hpp"
+#include "socketUtils.hpp"
+#include "ConcurrentQueue.hpp"
+#include "ConcurrentMap.hpp"
+
+class CalcuatorServerDriver {
+public:
+    CalcuatorServerDriver(std::string host, uint16_t port) : m_host_str(std::move(host)), m_port(port) {}
+
+    ~CalcuatorServerDriver();
+
+    void connect();
+
+    bool hasResult();
+
+    CalculatorResponse getResult();
+
+    void factorial(uint32_t id, int64_t arg);
+
+    void sqrt(uint32_t id, int64_t arg);
+
+    CalculatorResponse plus(uint32_t id, int64_t arg1, int64_t arg2);
+
+    CalculatorResponse minus(uint32_t id, int64_t arg1, int64_t arg2);
+
+    CalculatorResponse multiply(uint32_t id, int64_t arg1, int64_t arg2);
+
+    CalculatorResponse divide(uint32_t id, int64_t arg1, int64_t arg2);
+
+    void results();
+
+private:
+    void sendRequest(CalculatorRequest const &request);
+
+    CalculatorResponse getResponse(CalculatorRequest const &request);
+
+    void readingThreadTask();
+
+    ConcurrentQueue<CalculatorResponse> m_longResults;
+    ConcurrentMap<uint32_t, CalculatorResponse> m_instantResults;
+    int m_socket = 0;
+    std::string m_host_str;
+    sockaddr_in m_host;
+    uint16_t m_port;
+    std::thread m_readingThread;
+    std::unordered_set<uint32_t> m_longOperations;
+};

udp/client/include/CalculatorApp.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <iostream>
+#include <sstream>
+#include <iterator>
+#include "CalcuatorServerDriver.hpp"
+#include "socketUtils.hpp"
+
+class CalculatorApp {
+public:
+    CalculatorApp(std::string const &host, uint16_t port) : m_driver(host, port) {}
+
+    void start();
+
+private:
+    void printPrompt(uint32_t computationId);
+
+    void processInput(std::string &line);
+
+    template<typename T>
+    void printResult(uint32_t id, T const &value) {
+        std::cout << "Out [" << id << "]: " << value << std::endl;
+    }
+
+    void printEntryMessage();
+
+    void printLine(std::string const &line = "");
+
+    uint32_t m_currentComputation = 0;
+    CalcuatorServerDriver m_driver;
+
+    void printResponse(const CalculatorResponse &response);
+};

udp/client/include/ConcurrentMap.hpp

@@ -0,0 +1,38 @@
+#pragma once
+
+#include <mutex>
+#include <unordered_map>
+
+template<typename Key, typename Value>
+class ConcurrentMap {
+public:
+    void put(Key const &key, Value const &value) {
+        m_setMutex.lock();
+        m_map[key] = value;
+        m_setMutex.unlock();
+    }
+
+    bool contains(Key const &key) {
+        m_setMutex.lock();
+
+        if (m_map.find(key) != m_map.end()) {
+            m_setMutex.unlock();
+            return true;
+        }
+
+        m_setMutex.unlock();
+        return false;
+    }
+
+    Value pop(Key const &key) {
+        m_setMutex.lock();
+        auto result = m_map[key];
+        m_map.erase(key);
+        m_setMutex.unlock();
+        return result;
+    }
+
+private:
+    std::mutex m_setMutex;
+    std::unordered_map<Key, Value> m_map;
+};

udp/client/include/ConcurrentQueue.hpp

@@ -0,0 +1,33 @@
+#pragma once
+
+#include <mutex>
+#include <queue>
+
+template<typename T>
+class ConcurrentQueue {
+public:
+    void push(T const &response) {
+        m_queueMutex.lock();
+        m_resultsQueue.push(response);
+        m_queueMutex.unlock();
+    }
+
+    T pop() {
+        m_queueMutex.lock();
+        auto result = m_resultsQueue.front();
+        m_resultsQueue.pop();
+        m_queueMutex.unlock();
+        return result;
+    }
+
+    bool empty() {
+        m_queueMutex.lock();
+        bool isEmpty = m_resultsQueue.empty();
+        m_queueMutex.unlock();
+        return isEmpty;
+    }
+
+private:
+    std::mutex m_queueMutex;
+    std::queue<T> m_resultsQueue;
+};

udp/client/include/requests.hpp

@@ -0,0 +1,48 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+#pragma pack(push, 1)
+struct CalculatorResponse {
+    uint8_t errorCode;
+    uint8_t operationType;
+    uint32_t computationId;
+    int64_t result;
+};
+#pragma pack(pop)
+
+enum OperationType {
+    FAST = 1,
+    SLOW
+};
+
+enum ErrorCode {
+    OK = 0,
+    WAIT_FOR_RESULT,
+    OVERFLOW,
+    DIV_BY_ZERO,
+    FACT_OF_NEGATIVE,
+    SQRT_OF_NEGATIVE,
+};
+
+std::string errorCodeToString(uint8_t code);
+
+#pragma pack(push, 1)
+struct CalculatorRequest {
+    uint8_t type;
+    uint32_t computationId;
+    int64_t firstOperand;
+    int64_t secondOperand;
+};
+#pragma pack(pop)
+
+enum RequestType {
+    PLUS = 1,
+    MINUS,
+    MULT,
+    DIV,
+    SQRT,
+    FACT,
+    LONG_OP_RESULT
+};

udp/client/include/socketUtils.hpp

@@ -0,0 +1,34 @@
+#pragma once
+
+#include <iostream>
+#include <stdlib.h>
+
+#include <netdb.h>
+#include <netinet/in.h>
+#include <unistd.h>
+#include <strings.h>
+
+void writeToSocket(sockaddr_in &host, int socket, uint8_t const *data, size_t size);
+
+void readFromSocket(sockaddr_in &host, int socket, uint8_t *data, size_t size);
+
+void printError(const std::string &s);
+
+void error(const std::string &type, const std::string &s);
+
+int socketInit();
+
+sockaddr_in hostInit(std::string const &hostname, uint16_t port);
+
+template<typename T>
+void writeObject(sockaddr_in &host, int socket, const T &object) {
+    writeToSocket(host, socket, (uint8_t const *) &object, sizeof(T));
+}
+
+template<typename T>
+T readObject(sockaddr_in &host, int socket) {
+    T object;
+    readFromSocket(host, socket, (uint8_t *) &object, sizeof(T));
+
+    return object;
+}

udp/client/src/CalcuatorServerDriver.cpp

@@ -0,0 +1,93 @@
+#include "CalcuatorServerDriver.hpp"
+
+CalcuatorServerDriver::~CalcuatorServerDriver() {
+    shutdown(m_socket, SHUT_RDWR);
+    close(m_socket);
+}
+
+void CalcuatorServerDriver::connect() {
+    m_socket = socketInit();
+    m_host = hostInit(m_host_str, m_port);
+    m_readingThread = std::thread(&CalcuatorServerDriver::readingThreadTask, this);
+}
+
+void CalcuatorServerDriver::readingThreadTask() {
+    while (true) {
+        auto response = readObject<CalculatorResponse>(m_host, m_socket);
+        if (response.errorCode != WAIT_FOR_RESULT && response.operationType == SLOW) {
+            m_longResults.push(response);
+        }
+        m_instantResults.put(response.computationId, response);
+    }
+}
+
+bool CalcuatorServerDriver::hasResult() {
+    return !m_longResults.empty();
+}
+
+CalculatorResponse CalcuatorServerDriver::getResult() {
+    auto response = m_longResults.pop();
+    m_longOperations.erase(response.computationId);
+    return response;
+}
+
+void CalcuatorServerDriver::factorial(uint32_t id, int64_t arg) {
+    getResponse({FACT, id, arg, 0});
+    m_longOperations.insert(id);
+}
+
+void CalcuatorServerDriver::sqrt(uint32_t id, int64_t arg) {
+    getResponse({SQRT, id, arg, 0});
+    m_longOperations.insert(id);
+}
+
+CalculatorResponse CalcuatorServerDriver::plus(uint32_t id, int64_t arg1, int64_t arg2) {
+    return getResponse({PLUS, id, arg1, arg2});
+}
+
+CalculatorResponse CalcuatorServerDriver::minus(uint32_t id, int64_t arg1, int64_t arg2) {
+    return getResponse({MINUS, id, arg1, arg2});
+}
+
+CalculatorResponse CalcuatorServerDriver::multiply(uint32_t id, int64_t arg1, int64_t arg2) {
+    return getResponse({MULT, id, arg1, arg2});
+}
+
+CalculatorResponse CalcuatorServerDriver::divide(uint32_t id, int64_t arg1, int64_t arg2) {
+    return getResponse({DIV, id, arg1, arg2});
+}
+
+void CalcuatorServerDriver::results() {
+    for (uint32_t id : m_longOperations) {
+        getResponse({LONG_OP_RESULT, id, 0, 0});
+    }
+}
+
+void CalcuatorServerDriver::sendRequest(CalculatorRequest const &request) {
+    writeObject(m_host, m_socket, request);
+}
+
+CalculatorResponse CalcuatorServerDriver::getResponse(CalculatorRequest const &request) {
+    sendRequest(request);
+    int i = 0;
+    int resend = 0;
+
+    while (!m_instantResults.contains(request.computationId)) {
+        if (i == 10) {
+            if (resend == 5) {
+                error("get response", "Connection error.");
+            }
+            sendRequest(request);
+            resend++;
+            i = 0;
+        }
+
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+        i++;
+    }
+
+    return m_instantResults.pop(request.computationId);
+}
+
+
+