1
\$\begingroup\$

I've been working on implementing a multi-client socket communication system with a thread pool in C++. The system comprises three main components: logger.h, socket.h, and thread.h, which handle logging, socket operations, and thread pooling respectively. Additionally, I have a test suite tests_multisock.cpp that verifies the functionality of the system.

The code has been designed to facilitate communication between multiple clients and a server using sockets. I'd greatly appreciate your expertise in reviewing my code for any potential issues, optimizations, or areas of improvement. I've outlined a brief summary of the components and included a snippet of the test code for context. If you have any suggestions or feedback, I'm eager to hear them.

Components:

logger.h: A logging class for capturing events and errors. socket.h: A socket class to manage socket communication, supporting functions like creating, binding, listening, accepting, and sending/receiving data. thread.h: A thread pool class to handle concurrent execution of tasks. Test Suite:

tests_multisock.cpp: A suite of tests that validates the functionality of the multi-client socket communication and thread pool system.

include.h

#pragma once

#include <iostream>
#include <vector>
#include <queue>
#include <thread>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <iostream>
#include <ctime>
#include <string>
#include <fstream>
#include <thread>
#include <WinSock2.h>
#include <Ws2tcpip.h>

logger.h

#pragma once

#include "include.h"

enum class LogLevel
{
    DEBUG,
    INFO,
    WARNING,
    ERR
};

class Logger
{
public:
    Logger(LogLevel minLogLevel = LogLevel::INFO, const std::string& fileName = "default_log.txt")
        : minLogLevel(minLogLevel), fileName(fileName)
    {
        SetLogFile(fileName);
    }

    ~Logger()
    {
        if (logFile.is_open())
        {
            logFile.close();
        }
    }

    void SetLogFile(const std::string& filename)
    {
        logFile.open(filename, std::ios::app);
        if (!logFile.is_open())
        {
            std::cerr << "Failed to open log file: " << filename << std::endl;
        }
    }

    void Log(LogLevel level, const char* file, int line, const std::string& message)
    {
        if (level >= minLogLevel)
        {
            std::string logEntry = GetTimeStamp() + " [" + LogLevelToString(level) + "] " + message +
                " [" + file + ":" + std::to_string(line) + "]" + "\n";

            std::cout << logEntry;

            if (logFile.is_open())
            {
                logFile << logEntry;
                logFile.flush();
            }
        }
    }

    std::string GetLogFile() const
    {
        return fileName;
    }

private:
    LogLevel minLogLevel;
    std::ofstream logFile;
    std::string fileName;

    std::string LogLevelToString(LogLevel level) const
    {
        switch (level)
        {
        case LogLevel::DEBUG: return "DEBUG";
        case LogLevel::INFO: return "INFO";
        case LogLevel::WARNING: return "WARNING";
        case LogLevel::ERR: return "ERROR";
        default: return "UNKNOWN";
        }
    }

    std::string GetTimeStamp() const
    {
        std::time_t now = std::time(nullptr);
        char timestamp[20];
        struct tm timeinfo;
#ifdef _WIN32
        localtime_s(&timeinfo, &now);
#else
        localtime_r(&now, &timeinfo);
#endif
        std::strftime(timestamp, sizeof(timestamp), "%Y-%m-%d %H:%M:%S", &timeinfo);
        return timestamp;
    }
};

thread.h

#include "logger.h"

class ThreadPool {
public:
    ThreadPool(int numThreads, Logger& logger) : logger(logger), stop(false) {
        for (int i = 0; i < numThreads; ++i) {
            threads.emplace_back([this]() { ThreadFunction(); });
        }
    }

    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(mutex);
            stop = true;
        }
        condition.notify_all();

        for (std::thread& thread : threads) {
            thread.join();
        }
    }

    void Enqueue(std::function<void()> task) {
        {
            std::unique_lock<std::mutex> lock(mutex);
            tasks.push(task);
        }
        condition.notify_one();
    }

private:
    void ThreadFunction() {
        while (true) {
            std::function<void()> task;
            {
                std::unique_lock<std::mutex> lock(mutex);
                condition.wait(lock, [this]() { return stop || !tasks.empty(); });
                if (stop && tasks.empty()) {
                    return;
                }
                task = tasks.front();
                tasks.pop();
            }

            try {
                logger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Task started.");
                task();
                logger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Task completed.");
            }
            catch (const std::exception& ex) {
                logger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Task error: " + std::string(ex.what()));
            }
        }
    }

private:
    Logger& logger; // Reference to your Logger instance
    std::vector<std::thread> threads;
    std::queue<std::function<void()>> tasks;

    std::mutex mutex;
    std::condition_variable condition;
    bool stop;
};

socket.h

#include "logger.h"

class Socket
{
public:
    Socket(Logger& logger) : logger(logger) {}

    bool Create()
    {
        WSADATA wsaData;
        if (WSAStartup(MAKEWORD(2, 2), &wsaData) != 0)
        {
            logger.Log(LogLevel::WARNING, __FILE__, __LINE__, "WSAStartup failed");
            return false;
        }

        m_socket = socket(AF_INET, SOCK_STREAM, 0);
        if (m_socket == INVALID_SOCKET)
        {
            logger.Log(LogLevel::WARNING, __FILE__, __LINE__, "Failed to create socket: " + std::to_string(WSAGetLastError()));
            return false;
        }

        return true;
    }

    bool Bind(int port)
    {
        sockaddr_in hint;
        hint.sin_family = AF_INET;
        hint.sin_port = htons(port);
        hint.sin_addr.s_addr = INADDR_ANY;
        return bind(m_socket, (sockaddr*)&hint, sizeof(hint)) != SOCKET_ERROR;
    }

    bool Listen()
    {
        return listen(m_socket, SOMAXCONN) != SOCKET_ERROR;
    }

    bool Accept(Socket& clientSocket)
    {
        SOCKET client = accept(m_socket, nullptr, nullptr);
        if (client != INVALID_SOCKET)
        {
            clientSocket.m_socket = client;
            return true;
        }
        return false;
    }

    bool Connect(const char* ipAddress, int port)
    {
        sockaddr_in hint;
        hint.sin_family = AF_INET;
        hint.sin_port = htons(port);

        if (inet_pton(AF_INET, ipAddress, &hint.sin_addr) <= 0)
        {
            // Handle error, unable to convert IP address
            return false;
        }

        return connect(m_socket, (sockaddr*)&hint, sizeof(hint)) != SOCKET_ERROR;
    }

    int Send(const char* data, int dataSize)
    {
        return send(m_socket, data, dataSize, 0);
    }

    int Receive(char* buffer, int bufferSize)
    {
        return recv(m_socket, buffer, bufferSize, 0);
    }

    void Close()
    {
        if (m_socket != INVALID_SOCKET)
        {
            closesocket(m_socket);
            m_socket = INVALID_SOCKET;
            logger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Socket closed.");
        }
    }

    bool SendHttpRequest(const std::string& url, int port, const std::string& httpRequest, std::string& httpResponse)
    {
        if (!Create())
        {
            return false;
        }

        if (!Connect(url.c_str(), port))
        {
            return false;
        }

        if (Send(httpRequest.c_str(), httpRequest.size()) != static_cast<int>(httpRequest.size()))
        {
            return false;
        }

        const int bufferSize = 1024;
        char recvBuffer[bufferSize];
        httpResponse.clear();

        int bytesRead = 0;
        do
        {
            bytesRead = Receive(recvBuffer, bufferSize);
            if (bytesRead > 0)
            {
                httpResponse.append(recvBuffer, bytesRead);
            }
        } while (bytesRead > 0);

        return true;
    }

private:
    SOCKET m_socket;
    Logger& logger; // Reference to the Logger instance
};

tests_multisock.cpp

#include "logger.h"
#include "socket.h"
#include "thread.h"
#include <gtest/gtest.h>

class MultiClientCommunicationTest : public testing::Test {
protected:
    Logger logger;

    void SetUp() override {}
    void TearDown() override {}
};

TEST_F(MultiClientCommunicationTest, MultipleClients) {
    Logger serverLogger(LogLevel::DEBUG, "server_log.txt");
    Logger clientLogger(LogLevel::DEBUG, "client_log.txt");

    // Start a server thread
    std::thread serverThread([&serverLogger]() {
        Socket serverSocket(serverLogger);
        ASSERT_TRUE(serverSocket.Create());
        ASSERT_TRUE(serverSocket.Bind(12345));
        ASSERT_TRUE(serverSocket.Listen());
        serverLogger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Server listening");

        ThreadPool threadPool(50, serverLogger);

        for (int i = 0; i < 50; ++i) {
            threadPool.Enqueue([&serverSocket, i, &serverLogger]() {
                Socket clientSocket(serverLogger);
                ASSERT_TRUE(serverSocket.Accept(clientSocket));
                serverLogger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Server accepted client connection");

                const char* message = "Hello from server!";
                ASSERT_EQ(clientSocket.Send(message, static_cast<int>(strlen(message) + 1)), static_cast<int>(strlen(message) + 1));
                serverLogger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Server sent message to client");
                });
        }
        });

    // Start client threads
    std::vector<std::thread> clientThreads;

    for (int i = 0; i < 50; ++i) {
        clientThreads.emplace_back([&clientLogger]() {
            Socket clientSocket(clientLogger);
            ASSERT_TRUE(clientSocket.Create());
            ASSERT_TRUE(clientSocket.Connect("127.0.0.1", 12345));
            clientLogger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Client socket connected to server");

            char recvBuffer[1024] = { 0 };
            ASSERT_EQ(clientSocket.Receive(recvBuffer, sizeof(recvBuffer)), strlen("Hello from server!") + 1);
            clientLogger.Log(LogLevel::DEBUG, __FILE__, __LINE__, "Client received message from server");

            ASSERT_STREQ(recvBuffer, "Hello from server!");
            });
    }

    // Wait for server and client threads to finish
    serverThread.join();
    for (auto& thread : clientThreads) {
        thread.join();
    }
}

int main(int argc, char** argv) {
    testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}
\$\endgroup\$

2 Answers 2

Reset to default
2
\$\begingroup\$

Use of a thread pool

What was the point of creating a thread pool with 50 threads if you are going to submit exactly 50 tasks to it? You could have just used std::vector<std::thread> like you did for clientThreads.

The thread pool only makes sense if you are going to submit more tasks than you have threads, and you either want to avoid the overhead of creating and destroying lots of threads, or you have CPU-intensive tasks and you want to avoid running more threads concurrently than you have CPU cores. In the case of network communication, it's probably not the latter. But then you have to wonder: what if I have 100 clients connecting, and I only have a thread pool of size 50. The first 50 clients will be accepted, but the next 50 have to wait? What if you did more than just send "Hello from server!", but had a long running task for each connection?

More importantly though, if you only add 50 tasks that each accept one connection, how will you ever be able to service 100 clients? Instead of having each task call Accept(), you probably want to create one thread dedicated to accepting connections, and as soon as a connection is accepted, then create a thread that will further handle that connection.

Partial send()/recv() is not a bug

In your code you are assuming that send() will send the whole message in one go, and that the call to recv() will receive the whole message. However, there is no such guarantee, not even for blocking sockets. Make sure you check the return value (don't just use an assert), and use a loop to call send() repeatedly until either the whole message has been sent, or an error has occured.

For receiving you have to do the same, but it can even be more complicated if you don't know the size of the message you will receive in advance.

Validate the data that you receive

The sender might send something different than you expect. You are calling Receive() with a buffer of 1024 bytes. What if the sender sends 1024 bytes, none of which are a NUL byte? Luckily you only accept a string containing "Hello from server!", but if you didn't know the length and contents of the message, you can no longer trust that the buffer will be NUL-terminated.

\$\endgroup\$
1
\$\begingroup\$

I think include.h is a bad idea - it makes every compilation unit dependent on all the headers used by all the code. Instead, think about which headers are needed by which code, and include only what's needed for each. That will make compilation faster (and reduce the amount of recompilation needed if you update a subset of your libraries).

In the logger, consider using <chrono> rather than <ctime> functions. I find it interesting that you have an #ifdef _WIN32 there, yet the included include.h has unconditional includes of WinSock2.h> and <Ws2tcpip.h> which I'm guessing are only available on Windows targets.

That leads naturally to the Socket class. Most of that looks like Berkeley socket API, so we should be able to conditionally include <net/inet.h> and the rest to get a much more portable implementation.

The header-only structure might work acceptably for this simple test, but it's likely to lead to code bloat when linking several translation units that include the same headers. I recommend moving the implementations of your functions into their own translation units so that programs can use a single copy of each function rather than each translation unit having its own.

\$\endgroup\$

You must log in to answer this question.