2
\$\begingroup\$

I'm writing several local servers which have almost the same code in main.cpp. Appreciate your comments, improvement suggestions and especially notes on potential memory leaks since the services are supposed to run 24/7 and process large requests. Thanks!

#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <unordered_map>

#include "UdsServer.hpp"
#include "RequestManager.hpp"
#include "MSutils.hpp" //MS::log()



void pullRequests();


const std::string pathToSocket = "/var/run/SomeServer.sock";
const std::string SERVICE_NAME = "SomeServer";

RequestManager service; //Does the actual processing of the request


std::unordered_map<int, std::string> requestQueue;
std::mutex requestQueue_mutex;
std::condition_variable processorsThreadSwitch;
bool gotNewRequests = false;



int main()
{
    UdsServer app; //Server listening on a Unix Domain Socket

    try
    {
        app.createServer(pathToSocket);
    }
    catch (const std::string & err)
    {
        MS::log(SERVICE_NAME, "Failed to start the service. Error: " + err, MS::MessageType::FatalException);

        return -1;
    }

    unsigned n_concThreads = std::thread::hardware_concurrency();

    if (!n_concThreads) //if the query failed...
    {
        std::ifstream cpuinfo("/proc/cpuinfo");

        n_concThreads = std::count(std::istream_iterator<std::string>(cpuinfo),
                        std::istream_iterator<std::string>(),
                        std::string("processor"));

        if (!n_concThreads)
            n_concThreads = 6; // ~number of CPU cores. TODO: make the number of worker processes/threads configurable using a config file
    }

    for (int i = 0; i < n_concThreads; ++i)
    {
        std::thread t (pullRequests);
        t.detach();
    }



    while ((int clientConnection = app.newConnectionEstablished()) > -1) //Uses accept() internally
    {
        std::string command = app.getMsg (clientConnection); //Uses read() internally

        if (command.empty())
            app.closeConnection(clientConnection);
        else if (command == "SHUTDOWN")
        {
            app.closeConnection(clientConnection);

            return 0;
        }
        else
        {
            { //Anonymous scope just to get rid of the lock before notifying a thread
                std::lock_guard<std::mutex> writeLock(requestQueue_mutex);

                requestQueue[clientConnection] = std::move(command);

                gotNewRequests = true;
            }

            processorsThreadSwitch.notify_one(); //nothing happens here if all threads are busy
        }
    }
}



void pullRequests()
{
    UnixDomainSocket uds;

    std::unique_lock<std::mutex> writeLock(requestQueue_mutex);

    while (true) //Let the thread run "forever"
    {
        while (!gotNewRequests) 
            processorsThreadSwitch.wait(writeLock);


        std::unordered_map<int, std::string> queueCopy (std::move(requestQueue));
        requestQueue.clear();

        gotNewRequests = false;

        writeLock.unlock(); //Don't let the other threads wait when this threads doesn't need to access the shared data any more        


        if (queueCopy.empty())
            continue;
        else if (queueCopy.size() == 1)
        {
            std::string response = service.pullRequests(queueCopy.cbegin()->second);


            if (response.length())
            {
                auto sendResult = uds.sendMsg(queueCopy.cbegin()->first, response);

                if (!sendResult.isValid())
                    MS::log(SERVICE_NAME, "Could not send the response for request: " + queueCopy.begin()->second, MS::MessageType::Error);
            }

            if (!uds.closeConnection(queueCopy.begin()->first))
                MS::log(SERVICE_NAME, "Could not close the connection.", MS::MessageType::Error);
        }
        else //Multiplex
        {
            std::unordered_map<std::string, std::vector<int>> multiplexedRequests;

            for (auto & request : queueCopy)
                multiplexedRequests[std::move(request.second)].push_back(request.first);

            for (const auto & request : multiplexedRequests)
            {
                std::string response = service.pullRequests(request.first);

                if (response.length())
                    for (auto socket : request.second)
                    {
                        auto sendResult = uds.sendMsg(socket, response);

                        if (!sendResult.isValid())
                            MS::log(SERVICE_NAME, "Could not send the response for request: " + request.first, MS::MessageType::Error);

                        if (!uds.closeConnection(socket))
                            MS::log(SERVICE_NAME, "Could not close the connection.", MS::MessageType::Error);
                    }
            }
        }


        writeLock.lock();
    }
}
\$\endgroup\$
  • 1
    \$\begingroup\$ Have you run it with valgrind to check for memory leaks or race conditions (just wondering)? \$\endgroup\$ – esote Mar 28 at 0:18
  • \$\begingroup\$ If you running a professional site don't write your own. There are many good multi threaded servers out there. \$\endgroup\$ – Martin York Mar 28 at 7:08
  • \$\begingroup\$ It is not a web server. It's a microservice communicating with other local microservices. \$\endgroup\$ – Sceptical Jule Mar 28 at 9:02
  • \$\begingroup\$ @ScepticalJule: That does not affect my comment (or my review) in any way. Apache may be a bit heavy weight then but there are still plenty of already existing server based platforms you can use. Node.js jumps to mind as done engnx. Also why would you not use a standard (HTTP) over HTTPS well documented well understood protocol that already has tools for debugging and maintenance. \$\endgroup\$ – Martin York Mar 28 at 19:31
4
\$\begingroup\$

Overview

You over-allocate the number of child threads. Remember you should save one thread (the main thread for listening for new messages).

// Remember master thread takes a CPU while listening for connections.
unsigned n_concThreads = std::thread::hardware_concurrency() - 1;

Using multiple threads to handle incoming connection is not the way to go. The problem here is that the parallelization is limited by the number of threads (which is usually in the single digits low double digits if you are lucky).

A single thread can handle thousands of connections simultaneously because most of the time is spent waiting for data on the port. Thus your thread will be idle for most of the time it is handling a request waiting for input. If handled correctly, this idle time could be used to handle other requests.

A good example is the Node.js server. It is single-threaded and will easily handle thousands of incoming connections.

If you want to do this properly you need to use a library like LibEvent (or you can do it manually with select(), pselect() or ppoll()). This allows you to handle multiple sockets with the same thread.

But to be blunt. Doing this is not trivial and not for the beginner. I would opt for a server that already does this part of the work for you. Apache for heavyweight, or nginx for lightweight, or if you want to use another language Node.js. All these servers allow you to write your own request handling code but do the complex stuff for you.

Code Review

Bad form to have global variables.

std::unordered_map<int, std::string> requestQueue;
std::mutex requestQueue_mutex;
std::condition_variable processorsThreadSwitch;
bool gotNewRequests = false;

Would be much nicer to wrap this in a class and pass a reference to each string. That would make the application much more expandable in the long run.


OK. This is speculation. But I find it strange that something would throw a std::string! It would be more normal to throw something derived from std::runtime_error.

    try
    {
        app.createServer(pathToSocket);
    }
    catch (const std::string & err)
    {
        MS::log(SERVICE_NAME, "Failed to start the service. Error: " + err, MS::MessageType::FatalException);

        return -1;
    }

Also I would not return -1. I would rethrow the exception. This gives the calling function more information about the issue.


OK. A bit of a hack. But clever way of finding the number of processors.

    unsigned n_concThreads = std::thread::hardware_concurrency();

    if (!n_concThreads) //if the query failed...
    {
        std::ifstream cpuinfo("/proc/cpuinfo");

        n_concThreads = std::count(std::istream_iterator<std::string>(cpuinfo),
                        std::istream_iterator<std::string>(),
                        std::string("processor"));

        if (!n_concThreads)
            n_concThreads = 6; // ~number of CPU cores. TODO: make the number of worker processes/threads configurable using a config file
    }

But it is OS-specific. So I would wrap it in a function that is OS specific. That way on Linux you can do your clever check. On Windows it defaults to 6 (and the next person can simply replace the Windows version with a Windows specific hack without having to understand your Linux hack).


Not a fan of detaching the thread.

    for (int i = 0; i < n_concThreads; ++i)
    {
        std::thread t (pullRequests);
        t.detach();
    }

You lose all control of the thread. You also lose any warnings when you have a bug in your code and you accidentally shut down while one of the threads is still executing. Using detach() should be a last resource when all neat ways of doing something have been tried and failed.


The locking in pullRequests() seems highly roundabout. Move all the locking into a separate function. That way your scope lock behaves like it should.

I would have written it like this:

std::unordered_map<int, std::string> getNextTask()
{
     std::unique_lock<std::mutex> writeLock(requestQueue_mutex);

     // note this can be written more succinctly.
     // But I am to lazy to look up condition variables at the moment.
     // The following three lines should be a one liner
     // with wait taking a lambda
     while (!gotNewRequests) {
         processorsThreadSwitch.wait(writeLock);
     }

     std::unordered_map<int, std::string> result = std::move(requestQueue);
     return result;
}

void pullRequests()
{
    while(/*test*/)
    {
        // STUFF
        std::unordered_map<int, std::string> queueCopy = getNextTask();
        // STUFF
    }
}

Seems like some repeated code:

            if (!sendResult.isValid())
                    MS::log(SERVICE_NAME, "Could not send the response for request: " + queueCopy.begin()->second, MS::MessageType::Error);
            }

            if (!uds.closeConnection(queueCopy.begin()->first))
                MS::log(SERVICE_NAME, "Could not close the connection.", MS::MessageType::Error);

If you find repeated code then put it in a function.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.