1
\$\begingroup\$

In fact, this is a throttling proxy. It can work in two modes: "paired" and "multi-Client".

In "pair" mode, two clients connect to each other through a proxy, very similar to a p2p network, this is the BrokerPair class. Data sent by the first client passes through the proxy and is delivered to the second client. The same is true in the opposite direction, data sent by the second client passes through a proxy and is delivered to the first client.

In the "multi-Client" mode, the server first connects to the proxy, then any available number of clients, this is the BrokerMultiple class. What the server sends is sent to all clients. What is sent by the client is sent only to the server.

The testing capabilities themselves are implemented by the Emulator class. It allows you to emulate different emergency situations, for example two simple ones:

  • packet loss, if we are talking about UDP (yes, the code now uses TCP, this is for simplicity);
  • increase ping duration, which means a slowdown in all client-server interactions.

There is also an advantage in the speed of launching testing due to the lack of authorization/authentication logic, which almost all vendors of network p2p (via "NAT") gaming solutions have. Most often, authorization/authentication does not occur instantly.

Uses new "Coroutines TS" for "boost::asio" and therefore C++20.

Here is one key point that concern me:

  1. Is it possible to simplify anything?
  2. Are there any "overcomplicate things" here?

For brevity, it is omitted that #pragma once is used for each header file!

SignalSet.h

class SignalSet {
    boost::asio::signal_set m_signals;

public:
    SignalSet(boost::asio::io_context &io_context) :
        m_signals( io_context, SIGINT, SIGTERM )
    {}
    void ordinary() {
        m_signals.async_wait(
            [&](auto, auto) { 
                auto context = static_cast<boost::asio::io_context *>
                    ( &m_signals.get_executor( ).context( ) );
                context ->stop( );
            });
    }
};

IBroker.h

struct IBrokerPair {
    virtual ~IBrokerPair() {}
    virtual tcp::socket *getPair() const = 0;
    virtual void drop() = 0;
};
struct IBrokerMulti {
    virtual ~IBrokerMulti() {}
    virtual peers_t *getClients() = 0;
    virtual void remove(tcp::socket *) = 0;
    virtual void drop() = 0;
    virtual void setServer(tcp::socket *) = 0;
};

Command.h

struct Command {
    typedef std::array< char, 9 > type;
    // Simple pairs
    struct Pair {
        static constexpr type NoPair = { "NoPair\r\n" };
        static constexpr type NoMore = { "NoMore\r\n" };
    };
    // Multi clients
    struct Multi {
        static constexpr type ImServ = { "ImServ\r\n" };
        static constexpr type ImClie = { "ImClie\r\n" };
        static constexpr type Accept = { "Accept\r\n" };
        static constexpr type Denied = { "Denied\r\n" };
        static constexpr type WaitCl = { "WaitCl\r\n" };
        // the server is in the process of negotiation.
        static constexpr type SrvNeg = { "SrvNeg\r\n" };
    };
};
bool operator == (const data_t data, Command::type const& cmd) {
    return !memcmp( data, &cmd, cmd.size( ) - 1 );
}

Emulation/Config.h

namespace Emulation { 
struct Config {
    std::optional< uint32_t > lossOfEveryN;
    std::optional< uint32_t > longPingDuration_milli;
    std::optional< uint32_t > bandwidthInKBytes;
};
} // namespace Emulation

Emulation/Emulator.h where the "Statistics" class is omitted

namespace Emulation { 
class Emulator {
    config_t m_config;
    tcp::socket::executor_type const& m_executor;
    size_t m_counter = 0;
    bool isBandwidthAffect_(uint32_t) { return false; } // TODO(alex): makeme

public:
    Emulator(config_t config, tcp::socket::executor_type const& executor) : 
        m_config( config )
        , m_executor( executor )
    {}
    boost::asio::awaitable< bool > handle(std::size_t dataInBytes) {
        ++m_counter;
        if ( m_config.lossOfEveryN ) {
            if ( !( m_counter % m_config.lossOfEveryN.value( ) ) ) {
                co_return true;
            }
        }
        // @insp SO c20-coroutines-with-boost-awaitable-custom-suspension
        if ( m_config.longPingDuration_milli ) {
            boost::posix_time::milliseconds duration( m_config.longPingDuration_milli.value( ) );
            boost::asio::deadline_timer timer( m_executor, duration );
            co_await timer.async_wait( use_awaitable );
            co_return false;
        }
        if ( isBandwidthAffect_( dataInBytes ) ) {
            co_return false;
        }
        co_return false;
    }
};
} // namespace Emulation

Net/Communicator.h

namespace Net { 
class Communicator {
    data_t m_data = { };
    boost::system::error_code m_error; 

protected:
    boost::asio::awaitable< std::size_t > readData(tcp::socket &socket) {
        std::size_t n = 0;
        std::tie( m_error, n ) = co_await socket.async_read_some( buffer( m_data ), c_tuple );
        co_return n;
    }
    boost::asio::awaitable< bool > writeData(tcp::socket &socket, std::size_t n) {
        std::size_t unused_ = 0;
        std::tie( m_error, unused_ ) = co_await async_write( socket, buffer( m_data, n ), c_tuple );
        co_return !m_error;
    }
    auto getError() const {
        return m_error;
    }
    cref_data_t getData() const {
        return m_data;
    }

public:
    static boost::asio::awaitable< boost::system::error_code > writeCommand(tcp::socket &socket, Command::type command) {
        std::size_t unused_ = 0;
        boost::system::error_code error; 
        std::tie( error, unused_ ) = co_await async_write( socket, buffer( command ), c_tuple );
        co_return error;
    }
};
} // namespace Net

Net/Base.h where coding style formatting has been removed to allow the snippet to fit on the screen without scrolling.

namespace Net { 
class Base : public Communicator {
    tcp::socket m_socket; Emulation::Emulator m_emergencyEmulator;
    virtual peers_t *getPeers(cref_data_t, Command::type *) = 0;
protected:
    template<typename T> friend struct PipeSpawner;
    tcp::socket *getSocket() { return &m_socket; }
    awaitable pipe() {
        while ( true ) {
            std::size_t n = co_await readData( m_socket );
            if ( getError( ) )
                break;
            Command::type command;
            peers_t *peers = getPeers( getData( ), &command );
            if ( !peers ->size( ) ) {
                co_await writeCommand( m_socket, command );
                continue;
            }
            if ( co_await m_emergencyEmulator.handle( n ) )
                continue;
            for ( auto &peer : *peers ) 
                if ( !co_await writeData( *peer, n ) ) 
                    break;
        }
        m_socket.close( );
    }
    Base(tcp::socket &&socket, config_t config) : 
        m_socket( std::move( socket ) )
        , m_emergencyEmulator( config, m_socket.get_executor( ) )
    {}
public:
    virtual ~Base() {}
}; } // namespace Net

Net/PipeSpawner.h

namespace Net { 
template<typename T>
struct PipeSpawner {
    tcp::socket *start() {
        auto that = static_cast< T* >( this );
        co_spawn( that ->getSocket( ) ->get_executor( )
                , [self = that ->shared_from_this( )] { 
                    return self ->pipe( );
                }
                , c_detached 
            );
        return that ->getSocket( );
    }
};
} // namespace Net

Behavior/Pair.h

namespace Behavior { 
class Pair final : public Net::Base, public std::enable_shared_from_this< Pair >, public Net::PipeSpawner< Pair > {
    struct Private{};
    IBrokerPair *m_broker = nullptr;
    peers_t m_singlePeer;

    peers_t *getPeers(cref_data_t, Command::type *command) override {
        *command = Command::Pair::NoPair;
        if ( !m_singlePeer.size( ) ) 
            if ( tcp::socket *socket = m_broker ->getPair( ) ) 
                m_singlePeer.push_back( socket );
        return &m_singlePeer;
    }

public:
    Pair(Private, tcp::socket &&socket, tcp::socket *pair, IBrokerPair *broker, config_t config) : 
        Base( std::move( socket ), config )
        , m_broker( broker )
        , m_singlePeer{ pair ?peers_t{ pair } :peers_t{ } }
    {}
    ~Pair() {
        m_broker ->drop( );
    }
    static auto create(tcp::socket &&socket, tcp::socket *pair, IBrokerPair *broker, config_t config) {
        return std::make_shared< Pair >( 
            Private( ), std::move( socket ), pair, broker, config );
    }
};
} // namespace Behavior

Behavior/Multiple/Client.h

namespace Behavior::Multiple {
class Client final : public Net::Base, public std::enable_shared_from_this< Client >, public Net::PipeSpawner< Client > {
    struct Private{};
    IBrokerMulti *m_broker = nullptr;
    peers_t m_singlePeer;

    peers_t *getPeers(cref_data_t data, Command::type *command) override {
        (void)data, command;
        return &m_singlePeer;
    }

public:
    Client(Private, tcp::socket &&socket, IBrokerMulti *broker, tcp::socket *server, config_t config) : 
        Base( std::move( socket ), config )
        , m_broker( broker )
        , m_singlePeer{ server }
    {}
    ~Client() {
        m_broker ->remove( getSocket( ) );
    }
    static auto create(tcp::socket &&socket, IBrokerMulti *broker, tcp::socket *server, config_t config) {
        return std::make_shared< Client >( 
            Private( ), std::move( socket ), broker, server, config );
    }
};
} // namespace Behavior::Multiple

Behavior/Multiple/Server.h

namespace Behavior::Multiple {
class Server final : public Net::Base, public std::enable_shared_from_this< Server >, public Net::PipeSpawner< Server > {
    struct Private{};
    peers_t *m_peers = nullptr;
    IBrokerMulti *m_broker = nullptr;

    peers_t *getPeers(cref_data_t, Command::type *command) override {
        if ( !m_peers ->size( ) ) 
            return *command = Command::Multi::WaitCl, m_peers;
        return m_peers;
    }

public:
    Server(Private, tcp::socket &&socket, IBrokerMulti *broker, config_t config) : 
        Base( std::move( socket ), config )
        , m_broker( broker )
        , m_peers( broker ->getClients( ) )
    {}
    ~Server() {
        m_broker ->drop( );
    }
    static auto create(tcp::socket &&socket, IBrokerMulti *broker, config_t config) {
        return std::make_shared< Server >( 
            Private( ), std::move( socket ), broker, config );
    }
};
} // namespace Behavior::Multiple

Behavior/Multiple/Negotiator.h

namespace Behavior::Multiple { 
class Negotiator final : public std::enable_shared_from_this< Negotiator >, public Net::Communicator {
    struct Private{};
    bool m_isReady = false;
    tcp::socket m_socket;
    IBrokerMulti *m_broker = nullptr;
    config_t m_config;

    awaitable serverDetection() {
        co_await readData( m_socket );
        if ( getError( ) ) {
            co_return;
        }
        if ( Command::Multi::ImServ != getData( ) ) {
            co_await writeCommand( m_socket, Command::Multi::Denied );
            co_return;
        }
        co_await writeCommand( m_socket, Command::Multi::Accept );
        m_isReady = true;
        tcp::socket *server = Server::create( std::move( m_socket ), m_broker, m_config )
            ->start( );
        m_broker ->setServer( server );
        // waiting for the server to complete
        boost::system::error_code error;
        std::tie( error ) = co_await server ->async_wait( 
            tcp::socket::wait_error, c_tuple );
    }

public:
    typedef std::weak_ptr< Negotiator > serverDetector_t;
    Negotiator(tcp::socket &&socket, IBrokerMulti *broker, config_t config) :
        m_socket( std::move( socket ) )
        , m_broker( broker )
        , m_config( config )
    {}
    ~Negotiator() 
    {}
    void start(Private) {
        co_spawn( m_socket.get_executor( )
                , [self = shared_from_this( )] { 
                    return self ->serverDetection( );
                }
                , c_detached 
            );
    }
    bool isServerReady() {
        return m_isReady;
    }
    static serverDetector_t create(tcp::socket &&socket, IBrokerMulti *broker, config_t config) {
        auto detector = std::make_shared< Negotiator >( std::move( socket ), broker, config );
        detector ->start( { } );
        return detector;
    }
};
} // namespace Behavior::Multiple

BrokerPair.h

class BrokerPair final : public IBrokerPair {
    tcp::socket *m_first = nullptr, *m_second = nullptr;
    tcp::socket *getPair() const override { return m_second; }
    void drop() override { // drop pair
        if ( m_first )
            m_first ->close( );
        if ( m_second )
            m_second ->close( );
        m_first = m_second = nullptr;
    }

public:
    static awaitable listener(tcp::acceptor acceptor, config_t config) {
        BrokerPair broker;
        while ( true ) {
            auto [e, socket] = co_await acceptor.async_accept( c_tuple );
            if ( e ) 
                continue;
            if ( broker.m_second ) {
                co_await Net::Communicator::writeCommand( socket, Command::Pair::NoMore );
                continue;
            }
            if ( !broker.m_first ) 
                broker.m_first = Behavior::Pair::create( 
                    std::move( socket ), nullptr, &broker, config )
                    ->start( );
            else
                broker.m_second = Behavior::Pair::create( 
                    std::move( socket ), broker.m_first, &broker, config )
                    ->start( );
        }
    }
};

BrokerMultiple.h

class BrokerMultiple final : public IBrokerMulti {
    tcp::socket *m_server = nullptr;
    peers_t m_clients;
    peers_t *getClients() override {
        return &m_clients;
    }
    void remove(tcp::socket *socket) override {
        auto it = std::find( m_clients.begin( ), m_clients.end( ), socket );
        if ( m_clients.end( ) != it ) 
            m_clients.erase( it );
    }
    // drop all clients
    void drop() override {
        for ( auto &cl : m_clients )
            cl ->close( );
        m_clients.clear( );
    }
    void setServer(tcp::socket *server) override {
        m_server = server;
    }

public:
    static awaitable listener(tcp::acceptor acceptor, config_t config) {
        namespace Multiple = Behavior::Multiple;
        BrokerMultiple broker;
        Multiple::Negotiator::serverDetector_t wptrDetector;
        while ( true ) {
            auto [e, socket] = co_await acceptor.async_accept( c_tuple );
            if ( e ) 
                continue;
            if ( !wptrDetector.expired( ) ) {
                if ( wptrDetector.lock( ) ->isServerReady( ) ) {
                    tcp::socket *other = Multiple::Client::create( 
                        std::move( socket ), &broker, broker.m_server, config )
                        ->start( );
                    broker.m_clients.push_back( other );
                }
                else {
                    co_await Net::Communicator::writeCommand( socket, Command::Multi::SrvNeg );
                }
            } else {
                wptrDetector = Multiple::Negotiator::create( std::move( socket ), &broker, config );
            }
        }
    }
};

main.cpp omitted command line parsing, using hard-code in the structure instead. We can use the class BrokerPair or BrokerMultiple, this is in one comment.

#include "pch.h"
#include "SignalSet.h"
#include "IBroker.h"
#include "Command.h"
#include "Emulation/Config.h"
#include "Emulation/Emulator.h"
#include "Net/Communicator.h"
#include "Net/Base.h"
#include "Net/PipeSpawner.h"
#include "Behavior/Pair.h"
#include "Behavior/Multiple/Client.h"
#include "Behavior/Multiple/Server.h"
#include "Behavior/Multiple/Negotiator.h"
#include "BrokerPair.h"
#include "BrokerMultiple.h"
int wmain(int argc, wchar_t *argv[]) {
    Emulation::Config config{ .longPingDuration_milli = 1000 };
    boost::asio::io_context io_context( 1 );
    SignalSet signalSet( io_context );
    signalSet.ordinary( );
    auto acceptor = tcp::acceptor( io_context, { tcp::v4( ), port } );
    co_spawn( 
            io_context
            , BrokerPair::listener( std::move( acceptor ), config )
            //, BrokerMultiple::listener( std::move( acceptor ), config )
            , c_detached
        );
    io_context.run( );
    return 0;
}

Omitted the well-known pch.h.

aliases.h it's not interesting file with shortens

using awaitable = boost::asio::awaitable<void>;
using tcp = boost::asio::ip::tcp;

static constexpr auto c_detached = boost::asio::detached;
template <typename... Args>
inline auto as_tuple(Args&&... args) {
    return boost::asio::as_tuple( std::forward< Args >( args )... );
}
template <typename... Args>
inline auto buffer(Args&&... args) -> decltype( boost::asio::buffer( std::forward< Args >( args)... ) ) {
    return boost::asio::buffer( std::forward< Args >( args )... );
}
static constexpr auto c_tuple = as_tuple( boost::asio::use_awaitable_t{ } );

static const uint32_t c_dataInBytes = 1024;
typedef char data_t[ c_dataInBytes ];
typedef data_t const& cref_data_t;

typedef std::deque< tcp::socket * > peers_t;
typedef const Emulation ::Config config_t;

Among other things, I would like to receive advice on using coroutines.

Full code on GitHub repo

Regards, Alex0vSky

\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.