3
\$\begingroup\$

For this project, I'm using the non-Boost version of the ASIO library.

The basic idea is that it is a relatively simple stand-alone class capable of functioning either as a server or client depending on its needs. It performs all networking in an asynchronous paradigm, and as best as I'm able to determine, it can handle as many simultaneous connections as your operating system is able to maintain.

First, a few utility classes: I rolled my own concurrent versions of std::queue and std::map mostly as a training exercise. I'm using them in this project.

Next, the server itself:

#pragma once
#define ASIO_STANDALONE
#define ASIO_HAS_STD_ADDRESSOF
#define ASIO_HAS_STD_ARRAY
#define ASIO_HAS_CSTDINT
#define ASIO_HAS_STD_SHARED_PTR
#define ASIO_HAS_STD_TYPE_TRAITS
#include<asio.hpp>
#include<asio/placeholders.hpp>
#include<vector>
#include<thread>
#include<string>
#include<iostream>
#include<iomanip>
#include<chrono>
#include<sstream>
#include<random>
#include<functional>
#include<fstream>
#include<memory>
#include<exception>
#include<set>
#include<tuple>
#include<iostream>

#include<array>
#include<map>

#include "Concurrent Queue.h"
#include "Concurrent Map.h"

namespace server {

    constexpr inline unsigned long long power_function(const unsigned long long prime, const unsigned long long iterations) {
        return
            /*iterations >= 1'048'576 ? prime * power_function(prime, 1'048'575) * power_function(prime, iterations - 1'048'576) :
            iterations >= 65'536 ? prime * power_function(prime, 65'535) * power_function(prime, iterations - 65'536) :
            iterations >= 4'096 ? prime * power_function(prime, 4'095) * power_function(prime, iterations - 4'096) :*/
            iterations >= 256 ? prime * power_function(prime, 255) * power_function(prime, iterations - 256) :
            iterations >= 16 ? prime * power_function(prime, 15) * power_function(prime, iterations - 16) :
            iterations > 0 ? prime * power_function(prime, iterations - 1) : 1;
    }

    inline unsigned long long getSeed() {
        return std::chrono::steady_clock::now().time_since_epoch().count();
    }

    union connection_id {
    private:
        static constexpr unsigned long long prime = power_function(-363, 30'000);
        unsigned long long permute_seed(unsigned long long seed) {
            return (seed + 1) * 59 * prime;
        }
    public:
        unsigned long long l;
        unsigned char c[8];

        connection_id() :
            l(0)
        {

        }

        connection_id(unsigned long long seed) :
            l(permute_seed(seed))
        {
        }
        bool operator<(const connection_id & c) const {
            return l < c.l;
        }
        bool operator==(const connection_id & c) const {
            return l == c.l;
        }

        connection_id & operator=(unsigned long long seed) {
            l = permute_seed(seed);
            return *this;
        }

        connection_id & operator=(const connection_id & id) {
            if (this == &id) return *this;
            l = id.l;
            return *this;
        }

        std::string str() const {
            std::stringstream ss;
            static const std::set<int> dash_indexes = {0, 5, 10};
            ss << std::hex;
            ss << std::uppercase;
            for (int i = 0; i < 16; i++) {
                int byte_index = i / 2;
                int byte_half = (i + 1) % 2;
                unsigned int value = (c[byte_index] >> (byte_half * 4)) & 0xF;
                ss << value;
                if (dash_indexes.count(i)) ss << "-";
            }
            return ss.str();
        }

        operator std::string() const {
            return str();
        }

        friend std::ostream & operator<<(std::ostream & os, const connection_id & id);
    };

    std::ostream & operator<<(std::ostream & os, const connection_id & id) {
        return os << id.str();
    }

    typedef std::unique_ptr<asio::ip::tcp::socket> socket_ptr;
    static const size_t MAX_BUFFER_SIZE = 1'048'576;
    typedef std::array<unsigned char, MAX_BUFFER_SIZE> data_buffer_type;
    typedef std::unique_ptr<data_buffer_type> data_buffer;

    class connection {
    private:
        socket_ptr socket;
        connection_id id;
        std::string ip_address;
        std::string port;

    public:
        data_buffer read_buffer;

        connection(unsigned long long seed, asio::io_service& service, std::string ip, std::string _port) :
            id(seed),
            socket(socket_ptr(new asio::ip::tcp::socket(service))),
            ip_address(ip),
            port(_port),
            read_buffer(new data_buffer_type())
        {
        }

        connection(const connection_id & _id, asio::io_service & service, std::string ip, std::string _port) :
            id(_id),
            socket(socket_ptr(new asio::ip::tcp::socket(service))),
            ip_address(ip),
            port(_port),
            read_buffer(new data_buffer_type())
        {
        }

        bool operator<(const connection & c) const {
            return id < c.id;
        }

        bool operator==(const connection & c) const {
            return id == c.id;
        }

        asio::ip::tcp::socket & get_socket() const {
            return *socket;
        }

        std::string get_ip_address() {
            return ip_address;
        }

        std::string get_port() {
            return port;
        }

        connection_id get_id() const { return id; }
        void regen_id(unsigned long long seed) {
            id = seed;
        }
    };

    typedef std::vector<unsigned char> data_vector;
    typedef std::pair<connection_id, data_vector> data_pair;
    typedef concurrent::queue<data_pair> data_queue;

    typedef std::shared_ptr<connection> connection_ptr;
    typedef concurrent::map<connection_id, connection_ptr> connection_map;
    typedef unsigned long long buffer_id;
    typedef concurrent::map<buffer_id, data_vector> write_buffer_map;
    typedef std::unique_ptr<asio::io_service> service_ptr;
    typedef std::unique_ptr<asio::io_service::work> work_ptr;
    typedef std::unique_ptr<asio::ip::tcp::acceptor> acceptor_ptr;

    class basic_server {
    private:
        service_ptr service;
        work_ptr work;
        acceptor_ptr acceptor;
        connection_id client_connection;
        connection_map connections;
        std::mutex mutex;
        std::condition_variable cond;
        volatile bool running;
        data_queue read_queue;
        write_buffer_map write_buffers;
        buffer_id current_buffer_id;
        std::ofstream lout;

        std::thread service_thread;

        buffer_id get_next_buffer_id() {
            return current_buffer_id++;
        }

        void service_function() {
            asio::error_code ec;
            while (!running) std::this_thread::yield();
            while (running) {
                try {
                    service->run(ec);
                    break;
                }
                catch (std::exception & e) {
                    lout << "ERROR: " << e.what() << std::endl;
                }
            }
        }

        void accept(const asio::error_code & ec, const connection_id & id) {
            if (ec) {
                connections.erase(id);
                return;
            }
            using namespace std::chrono_literals;
            connection_ptr conn = connections[id];
            conn->get_socket().async_read_some(
                asio::buffer(
                    &conn->read_buffer->front(), 
                    MAX_BUFFER_SIZE
                ),
                std::bind(
                    &basic_server::read, 
                    this, 
                    std::placeholders::_1, 
                    std::placeholders::_2,
                    id
                )
            );
            connection_id nid(getSeed());
            while (connections.contains(nid)) {
                std::this_thread::sleep_for(std::chrono::milliseconds(1));
                nid = getSeed();
            }
            connections[nid] = connection_ptr(new connection(nid, *service, conn->get_ip_address(), conn->get_port()));
            connection_ptr nconn = connections[nid];
            acceptor->async_accept(
                nconn->get_socket(), 
                std::bind(
                    &basic_server::accept,
                    this,
                    std::placeholders::_1,
                    nid
                )
            );
        }

        void connect(const asio::error_code & ec, const connection_id & id) {
            if (ec) {
                connections.erase(id);
                return;
            }
            connection_ptr conn = connections[id];
            conn->get_socket().async_read_some(
                asio::buffer(
                    &conn->read_buffer->front(),
                    MAX_BUFFER_SIZE
                ),
                std::bind(
                    &basic_server::read,
                    this,
                    std::placeholders::_1,
                    std::placeholders::_2,
                    id
                )
            );
        }

        void read(const asio::error_code & ec, size_t bytes_read, const connection_id & id) {
            if (ec) {
                connections.erase(id);
                return;
            }
            connection_ptr conn = connections[id];
            data_vector queue_data;
            queue_data.insert(queue_data.end(), conn->read_buffer->begin(), conn->read_buffer->begin() + bytes_read);
            read_queue.push(data_pair(id, queue_data));

            conn->get_socket().async_read_some(
                asio::buffer(
                    &conn->read_buffer->front(),
                    MAX_BUFFER_SIZE
                ),
                std::bind(
                    &basic_server::read,
                    this,
                    std::placeholders::_1,
                    std::placeholders::_2,
                    id
                )
            );
        }

        void write(const asio::error_code & ec, size_t bytes_written, const connection_id & id, buffer_id buf_id) {
            write_buffers.erase(buf_id);
            if (ec) {
                connections.erase(id);
            }
        }
    public:
        basic_server() :
            service(new asio::io_service()),
            work(new asio::io_service::work(*service)),
            acceptor(new asio::ip::tcp::acceptor(*service)),
            running(true),
            service_thread(std::bind(&basic_server::service_function, this)),
            lout("server.log"),
            current_buffer_id(0)
        {
        }

        void start_listening(const std::string & ip_address, const std::string & port) {
            asio::ip::tcp::resolver resolver(*service);
            asio::ip::tcp::resolver::query query(ip_address, port);
            asio::ip::tcp::endpoint endpoint = *(resolver.resolve(query));

            acceptor->open(endpoint.protocol());
            acceptor->bind(endpoint);
            acceptor->listen(asio::socket_base::max_connections);

            connection_id id(getSeed());
            connections[id] = connection_ptr(new connection(id, *service, ip_address, port));
            connection_ptr conn = connections[id];

            acceptor->async_accept(
                conn->get_socket(),
                std::bind(
                    &basic_server::accept,
                    this,
                    std::placeholders::_1,
                    id
                )
            );
        }

        void start_connecting(const std::string & ip_address, const std::string & port) {
            asio::ip::tcp::resolver resolver(*service);
            asio::ip::tcp::resolver::query query(ip_address, port);
            asio::ip::tcp::endpoint endpoint = *(resolver.resolve(query));

            connection_id id(getSeed());
            connections[id] = connection_ptr(new connection(id, *service, ip_address, port));
            connection_ptr conn = connections[id];
            client_connection = id;

            conn->get_socket().async_connect(
                endpoint,
                std::bind(
                    &basic_server::connect,
                    this,
                    std::placeholders::_1,
                    id
                )
            );
        }

        ~basic_server() {
            stop();
            service_thread.join();
        }

        bool read_from_queue(data_pair & data) {
            return read_queue.try_pop(data);
        }

        void write_to_connection(const connection_id & id, const data_vector & data) {
            if (connections.contains(id)) {
                buffer_id bid = get_next_buffer_id();
                write_buffers[bid] = data;
                data_vector & saved_buffer = write_buffers[bid];
                connection_ptr conn = connections[id];
                conn->get_socket().async_write_some(
                    asio::buffer(
                        &saved_buffer.front(), 
                        saved_buffer.size()
                    ),
                    std::bind(
                        &basic_server::write,
                        this,
                        std::placeholders::_1,
                        std::placeholders::_2,
                        id,
                        bid
                    )
                );
            }
        }

        void write_to_connection(const connection_id & id, const void * data, size_t size_of_data) {
            data_vector vector(size_of_data);
            const unsigned char * data_ptr = static_cast<const unsigned char *>(data);
            std::copy(data_ptr, data_ptr + size_of_data, vector.begin());
            write_to_connection(id, vector);
        }

        void client_write(const data_vector & data) {
            connection_id id = client_connection;
            write_to_connection(id, data);
        }

        void stop() {
            running = false;
            work.reset();
            service->stop();
            read_queue.wake_all();
        }

        std::set<connection_id> get_all_connection_ids() const {
            return connections.getKeySet();
        }

        bool is_connected(const connection_id & id) const {
            return connections.contains(id);
        }

        const connection_ptr get_connection(const connection_id & id) const {
            return connections.at(id);
        }
    };
}

The public-facing functions are as follows:

  • start_listening(string ip_address, string port): If this is meant to be used as a server, it'll start waiting for clients to connect on the given IP address and port. I promise to add a stop_listening() command at some point, but this version doesn't have that.
  • start_connecting(string ip_address, string port): Does the same thing as start_listening, but from the other direction. Has a few semantics that differ as a result. To my knowledge, there's no reason this function would fail if a user attempted to connect to multiple servers simultaneously.
  • read_from_queue(data_pair & data): The server/client stores an internally managed queue of reads from all of its connections. This stores in the provided reference object a pair of the connection_id and the data received from that connection.
  • write_to_connection(const connection_id & id, const data_vector & data): The server will check if the provided connection_id is valid, and if it is, it will attempt to send the data to that connection.
  • write_to_connection(const connection_id & id, const void * data, size_t size_of_data): Does the same thing as the other write_to_connection function, but allows the user to just point to a raw pointer and provide a size of the pointer. Useful as a shorthand for "wrap this data in a vector and send it".
  • client_write(const data_vector & data): If this is operating as a client, it will attempt to send to the last connected server. This is mainly just a convenience method to simplify the server<-->client model.

I've also written two basic programs to showcase the server working correctly. These files aren't part of the code review, but they show how my model works.

Server Main simply waits for connections, and if it receives connections, writes any output from that connection into a file named after its connection_id. Send File Main simply connects to a server, and attempts to send it the contents of an arbitrary file it picks up. In my example, I have an (exactly) 20,000,000 byte file full of garbage (that I named bigfile.dat) that I use to test and verify that it is being sent correctly.

So what I'm looking for are several things:

  • Are there substantial security concerns with the code as I've written it here, particularly in Basic Server.h?
  • Is there a better way for me to be doing error handling?
  • Are there any race conditions I'm not accounting for? The expectation is that only a single thread interacts with the basic_server object at any given moment, but that doesn't preclude the possibility of the internal service_thread clashing with the owner thread. Bear in mind that most of the internal data structures are using my concurrent data types.
  • Have I made any egregious errors in terms of readability, code structure, best-practice, etc.?
\$\endgroup\$
  • \$\begingroup\$ Could you edit your question to embed the important parts of the code in the question? \$\endgroup\$ – David Morris Feb 17 '16 at 22:32
  • \$\begingroup\$ @DavidMorris Would putting Basic Server.h be sufficient? The two concurrent data structure classes are just simple wrappers around STL containers, and the two .cpp files are just example code, not part of the server itself. \$\endgroup\$ – Xirema Feb 17 '16 at 22:35
  • \$\begingroup\$ that seems reasonable to me. I think the basic idea is that any (Question, Answer) pair should be able to stand alone, without needing external references to make sense of it. If the code you would like reviewed is principally in Basic Server.h, then including it should satisfy that property. \$\endgroup\$ – David Morris Feb 17 '16 at 22:41

Your Answer

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

Browse other questions tagged or ask your own question.