Response dispatcher for HTTP server

Question

I’m writing a small HTTP server (just for fun and education). One important part is the response dispatcher. Responses for each socket should be sent in the order the requests were received. The idea is:

• There’s an asynchronous connection manager that registers the request (and get id) and sends it to some request handler
• The request handler is a multithreading black box (at least for now) that creates the response and puts it to the response dispatcher.
• The response dispatcher having request sequences for each socket and ready responses fills the queue that stores the responses that should be sent now.

Now, when ResponseDispatcher class works (functionally, it works as supposed to) I'd appreciate any ideas to improve my code.

Here you can find the whole project.

ResponseDispatcher.h

#pragma once

#include <map>
#include <list>

#include "BlockingQueue.h"
#include "winsock2.h"
#include "DataTypes.h"

class ResponseDispatcher
{

public:
    ResponseDispatcher();
    ~ResponseDispatcher();

    void registerRequest( RequestIdType, SOCKET );
    void registerResponse( ResponsePtr );

    ResponseData * pullResponse();

    void removeResponse(RequestIdType);

    void Dump();

    void putTopResponseToQueue(SOCKET);

    SOCKET getSocket(RequestIdType) const;

private:
    std::map< RequestIdType, SOCKET >               m_id2SocketMap;
    std::map< RequestIdType, ResponsePtr >          m_storedResponses;
    std::map< SOCKET, std::list<RequestIdType> >    m_topQueue;

    BlockingQueue<RequestIdType> m_topResponces;

    std::mutex m_Mtx;
};

ResponseDispatcher.cpp

#include "stdafx.h"
#include "ResponseDispatcher.h"

#include<sstream>


ResponseDispatcher::ResponseDispatcher()
{
}


ResponseDispatcher::~ResponseDispatcher()
{
}

void ResponseDispatcher::registerRequest( RequestIdType id, SOCKET sock )
{
    static char pNof[] = __FUNCTION__ ": ";

    if (m_id2SocketMap.find(id) != m_id2SocketMap.end())
    {
        std::stringstream error;
        error << pNof << "Request id duplication #" << id;
        throw std::runtime_error( error.str() );
    }

    std::unique_lock<std::mutex> lck;

    m_id2SocketMap[id] = sock;

    std::list<RequestIdType> & list = m_topQueue[sock];

    auto itFirstSmaller = std::lower_bound( list.begin(), list.end(), id );
    list.insert(itFirstSmaller, id);
}

void ResponseDispatcher::registerResponse( ResponsePtr response )
{
    static char pNof[] = __FUNCTION__ ": ";

    RequestIdType id = response->id;

    std::unique_lock<std::mutex> lck;

    if (m_storedResponses.find(id) != m_storedResponses.end())
    {
        std::stringstream error;
        error << pNof << "Duplicating response id #" << id;
        throw std::runtime_error(error.str());
    }

    auto itID = m_id2SocketMap.find(id);
    if ( itID == m_id2SocketMap.end() )
    {
        std::stringstream error;
        error << pNof << "Can't find socket for the response id #" << id;
        throw std::runtime_error( error.str() );
    }

    SOCKET sock = itID->second;

    m_storedResponses[id] = std::move(response);

    putTopResponseToQueue(sock);
}

SOCKET ResponseDispatcher::getSocket(RequestIdType id) const
{
    static char pNof[] = __FUNCTION__ ": ";

    auto it = m_id2SocketMap.find(id);
    if ( it == m_id2SocketMap.end() )
    {
        std::stringstream error;
        error << pNof << "Can't find socket for the response id #" << id;
        throw std::runtime_error(error.str());
    }

    return it->second;
}

void ResponseDispatcher::putTopResponseToQueue(SOCKET sock)
{
    static char pNof[] = __FUNCTION__ ": ";

    std::unique_lock<std::mutex> lck;

    auto itSock = m_topQueue.find(sock);
    if (itSock == m_topQueue.end() )
    {
        std::stringstream error;
        error << pNof << "Can't find socket #" << sock;
        throw std::runtime_error(error.str());
    }

    RequestIdType topId = itSock->second.front();

    if ( !m_topResponces.contains(topId) )
    {
        m_topResponces.push(topId);
    }
}

ResponseData * ResponseDispatcher::pullResponse()
{
    RequestIdType id = m_topResponces.pull();
    return m_storedResponses[id].get();
}

void ResponseDispatcher::removeResponse( RequestIdType id )
{
    static char pNof[] = __FUNCTION__ ": ";

    std::unique_lock<std::mutex> lck;

    auto itID = m_id2SocketMap.find(id);
    if (itID == m_id2SocketMap.end() )
    {
        std::stringstream error;
        error << pNof << "Can't find socket for the response id #" << id;
        throw std::runtime_error(error.str());
    }

    SOCKET sock = itID->second;

    auto itSock = m_topQueue.find(sock);
    if ( itSock == m_topQueue.end() )
    {
        std::stringstream error;
        error << pNof << "Can't find socket #" << sock;
        throw std::runtime_error(error.str());
    }

    itSock->second.remove(id);

    m_storedResponses.erase( m_storedResponses.find(id) );

    m_id2SocketMap.erase( m_id2SocketMap.find(id) );

    putTopResponseToQueue( sock );
}


#include <iostream>
void ResponseDispatcher::Dump()
{
    auto printList = [&](const std::pair<SOCKET, std::list<RequestIdType>> & item)
    {
        std::cout << item.first << ": ";
        std::for_each(item.second.begin(), item.second.end(), [&](const RequestIdType & i) {std::cout << i << " "; });
        std::cout << std::endl;
    };

    std::for_each(m_topQueue.begin(), m_topQueue.end(), printList);
    std::cout << std::endl;



    std::cout << "id vs sockets: ";
    std::for_each(m_id2SocketMap.begin(),
                    m_id2SocketMap.end(),
                    [&](const auto & t) {std::cout << "[" << t.first << ":" << t.second << "] "; });
    std::cout << std::endl;


    std::cout << "Stored responses: ";
    std::for_each(m_storedResponses.begin(),
                    m_storedResponses.end(),
                    [&](const auto & t) {std::cout << t.first << " "; });
    std::cout << std::endl;


    std::cout << "Top responses: " << m_topResponces.Dump() << std::endl << std::endl;
    std::cout << " ======================== " << std::endl << std::endl;

}

Answer 1

DRY (Don't repeat yourself)

There are lots of snippets that are repeated the same or very similar across many functions:

• Error handling: All cases of error handling have the following form:

  std::stringstream error;
  error << pNof << some_string << some_id;
  throw std::runtime_error( error.str() );
  

  This could easily be extracted into a helper function:

  template<typename ID>
  static void report_error(std::string func_name, std::string error_msg, const ID& id) {
      std::string_stream error;
      error << func_name << ": " << error_msg << id;
      throw std::runtime_error( error.str() );
  }
  

  For C++17 and beyond, I'd use std::string_view instead of std::string in the example above. Actually, I might use a variadic template and a fold expression over << to simply pass every argument (excluding the function name) to error if this pattern is used across multiple files with different usages.

• static char pNof[] = __FUNCTION__ ": ";

  If this really has to be repeated for every member function (because of some requirement), maybe use a macro instead? That way each declaration is uniform, and any possible future changes are limited to one place.

  If there is no requirement, then using __func__ (as __FUNCTION__ is not portable) where appropriate (e.g. when calling the helper function above) might be even better.

• Some of the error messages repeat verbatim in different locations. Maybe extract them (possibly all of them) into some static variables to be reused?

Naming

Some variables have poor names. Examples:

• pNof: What does this even stand for? "pointer to name of function"? function_name (or func_name) might be more readable.

• I get it, list inside registerRequest is a std::list - but what does it represent?

• Similarly, I have no idea what m_topQueue stands for.

• m_topResponces is inconsistenly named (a typo?), m_topResponses would match other usages of Response.

Multithreading issues

None of the std::unique_lock<std::mutex> lck; actually acquire a lock, because no reference to a mutex is passed.

This means that there is no synchronization at all in the whole class!

I guess std::unique_lock<std::mutex> lck{m_Mtx}; was meant instead.

Once that is fixed, there are these issues left:

• Unsychronized access to m_id2SocketMap in registerRequest: Another thread might invalidate any of the iterators returned by find(id) or end() before the comparison is made.

• registerResponse calls putTopResponseToQueue while still holding the lock on m_Mtx, which putTopResponseToQueue will try to acquire - and fail, resulting in a dead-lock.

• Unsynchronized access to m_id2SocketMap in getSocket: Another thread might invalidate any of the iterators returned by find(id) or end() before the comparison is made or before it->second is accessed.

• Unsynchronized access to m_topResponces and m_storedResponses in pullResponse. (Even worse, those accesses might modify the respective containers!)

• removeResponse calls putTopResponseOnQueue while still holding the lock on m_Mtx, which putTopResponseToQueue will try to acquire - and fail, resulting in a dead-lock.

• Dump acquires no lock at all, and at any point the used iterators might be invalidated by other threads.

There seems to be an assumption that calling any const container member function is thread-safe. This is only true if there are no concurrent modifications on the same container! Otherwise, even const member function calls might result in data races, or iterators returned might be invalidated before further use.

---

Fixing all these issues will likely mean that every operation on ResponseDispatcher will rely on locking the same mutex m_Mtx, removing any direct benefit of concurrency inside this class.

Some of that can be improved by introducing one mutex per member variable, so operations on different members can be done in parallel. However, those mutexes would have to be used with care in order to prevent deadlocks due to different order of acquisition in different locations.

Even better if that mutex is similar to the C++17 std::shared_mutex which allows parallel reads but no parallel writes. IIRC boost offered a reader/writer lock for C++14 which could be used.

However, as always if performance is required, measure if it is indeed better, as those results might vary for different use cases.

Other issues

• Any reason std::map is used instead of std::unordered_map? The latter will likely result in better performance, as insertion, lookup and deletion are \$\mathcal{O}(1)\$ (amortized for insertion/deletion) instead of \$\mathcal{O}(\log n)\$.

• In a lot of variable declarations auto could be used instead of long unwindy explicit type names. (See this Guru of the Week post for more info why that might be useful.)

• In removeResponse the second call to m_id2SocketMap.find(id) is unnecessary, the earlier obtained iterator itID could be reused.