2
\$\begingroup\$

For gaming applications, networking looks strange from the point of view of traditional approaches. Much of the game logic is based on "ticks"; Most often this is a mandatory item of operation. There is also a limitation from the "Epic EOS SDK" in the absence of thread safety, all "C" calls can only be made from one thread. Additionally, there is no explicit separation between client and server; we are dealing with a "p2p" network (transport via UDP protocol, WebRTC).

The main.cpp code provided simply sends and receives a text strings twice.

Here is one key point that concern me:

  1. Have I taken the best approach in terms of performance?

QueueCommands.h

#pragma once
namespace detail_ { template <typename F, typename... Ts> class Action; } // namespace detail_ // forward decl
class QueueCommands {
    static constexpr std::chrono::seconds c_commandTO{ 30 };
    static constexpr auto now = std::chrono::system_clock::now;
    EOS_HPlatform m_platformHandle;
    EOS_HP2P m_p2PHandle;
    struct AvoidPush { bool Yes; } m_avoidPush{ true };

public:
    enum class Direction { Outgoing, Incoming };
    struct ICommand {
        virtual Networking::messageData_t act(const QueueCommands::AvoidPush&) = 0;
        virtual Direction getDirection() const = 0;
        virtual ~ICommand() {}
    };

private:
    typedef std::shared_ptr< ICommand > sptr_t;
    std::queue< sptr_t > m_fifo;

    explicit QueueCommands(EOS_HPlatform platformHandle) :
        m_platformHandle( platformHandle )
        , m_p2PHandle( ::EOS_Platform_GetP2PInterface( platformHandle ) )
    {}
    QueueCommands(const QueueCommands &) = delete;
    QueueCommands(QueueCommands &&) = delete;
    QueueCommands& operator=(const QueueCommands &) = delete;
    QueueCommands& operator=(QueueCommands &&) = delete;

    static QueueCommands& getInstanceImpl_(EOS_HPlatform platformHandle = nullptr) {
        static QueueCommands instance{ platformHandle };
        return instance;
    }
    sptr_t pop_() {
        if ( m_fifo.empty( ) )
            return nullptr;
        auto x = m_fifo.front( );
        m_fifo.pop( );
        return x;
    }
    template <typename, typename...> friend class detail_::Action;
    friend class Receiving;
    friend class Sending;
    void push(const sptr_t &p) {
        m_fifo.push( p );
    }
    void tick_() {
        ::EOS_Platform_Tick( m_platformHandle );
    }

    uint64_t outgoingPacketQueueCurrentSizeBytes_() {
        EOS_P2P_PacketQueueInfo queueInfo = { };
        EOS_P2P_GetPacketQueueInfoOptions options = { EOS_P2P_GETPACKETQUEUEINFO_API_LATEST };
        ::EOS_P2P_GetPacketQueueInfo( m_p2PHandle, &options, &queueInfo ); // error checking has been ommited for "codereview"
        return queueInfo.OutgoingPacketQueueCurrentSizeBytes;
    }

public:
    static void init(EOS_HPlatform platformHandle) {
        getInstanceImpl_( platformHandle );
    }
    static QueueCommands& instance() {
        return getInstanceImpl_( );
    }
    auto ticksAll() {
        std::vector< Networking::messageData_t > allIncomingData;
        sptr_t command;
        while ( command = pop_( ) ) {
            if ( Direction::Outgoing == command ->getDirection( ) ) {
                command ->act( m_avoidPush );
                do {
                    tick_( );
                } while ( outgoingPacketQueueCurrentSizeBytes_( ) );
            }

            auto timeout = now( ) + c_commandTO;
            if ( Direction::Incoming == command ->getDirection( ) ) {
                Networking::messageData_t packet = { };
                do {
                    packet = command ->act( m_avoidPush );
                    if ( packet.size( ) )
                        break;
                    tick_( );
                } while ( now( ) < timeout );
                allIncomingData.push_back( packet );
            }
        }
        return allIncomingData;
    }
};

Action.h

#pragma once
namespace detail_ {
template <typename F, typename... Ts>
class Action : 
    public QueueCommands::ICommand 
    , public std::enable_shared_from_this< Action< F, Ts... > >
{
    static_assert( !( std::is_rvalue_reference_v<Ts> &&... ) );
    const QueueCommands::Direction m_direction;
    const F m_function;
    const std::tuple<Ts...> m_args;

public:
    template <typename FwdF, typename... FwdTs,
        typename = std::enable_if_t< ( std::is_convertible_v< FwdTs&&, Ts > &&... ) >>
    Action(QueueCommands::Direction direction, FwdF&& func, FwdTs&&... args) : 
        m_direction( direction )
        , m_function( std::forward<FwdF>( func ) )
        , m_args{ std::forward<FwdTs>( args )... }
    {}
    virtual ~Action()
    {}

    [[nodiscard]] Networking::messageData_t act(
        const QueueCommands::AvoidPush& avoidPush = QueueCommands::AvoidPush{ } 
    ) override {
        Networking::messageData_t messageData = std::apply( m_function, m_args );
        if ( !avoidPush.Yes )
            QueueCommands::instance( ).push( shared_from_this( ) );
        return messageData;
    }
    QueueCommands::Direction getDirection() const override {
        return m_direction;
    }
};
template <typename F, typename... Args>
auto make_action(QueueCommands::Direction direction, F&& f, Args&&... args) {
    return std::make_shared
            < Action< std::decay_t< F >, std::remove_cv_t< std::remove_reference_t< Args > >...> >
            ( direction, std::forward< F >( f ), std::forward< Args >( args )... )
        ;
}
} // namespace detail_

aux Ctx

struct Ctx {
    const std::string m_SocketName;
    const EOS_ProductUserId m_LocalUserId;
    const EOS_HPlatform m_PlatformHandle;
    const EOS_ProductUserId m_FriendLocalUserId;
};

Sender/SendText.h

#pragma once
namespace Sender {
class SendText {
    const Ctx m_ctx;
    const EOS_HP2P m_p2PHandle;
    EOS_P2P_SocketId m_sendSocketId;
    EOS_P2P_SendPacketOptions m_options;

    template<typename T>
    void sendPacket_(const T &messageData) {
        const size_t maxDataLengthBytes = Networking::c_MaxDataSizeBytes;
        auto it = messageData.begin( );
        while ( it != messageData.end( ) ) {
            const size_t distance = std::distance( it, messageData.end( ) );
            const size_t dataLengthBytes = std::min( maxDataLengthBytes, distance );

            m_options.DataLengthBytes = static_cast< uint32_t >( dataLengthBytes );
            m_options.Data = std::addressof( *it );
            ::EOS_P2P_SendPacket( m_p2PHandle, &m_options ); // error checking has been ommited for "codereview"
            it += dataLengthBytes;
        }
    }

public:
    SendText(const Ctx &ctx, uint8_t channel) :
        m_ctx( ctx )
        , m_p2PHandle( ::EOS_Platform_GetP2PInterface( ctx.m_PlatformHandle ) )
        , m_sendSocketId{ EOS_P2P_SOCKETID_API_LATEST }
        , m_options{ EOS_P2P_SENDPACKET_API_LATEST }
    {
        strcpy_s( m_sendSocketId.SocketName , m_ctx.m_SocketName.c_str( ) );
        m_options.LocalUserId = m_ctx.m_LocalUserId;
        m_options.RemoteUserId = m_ctx.m_FriendLocalUserId;
        m_options.SocketId = &m_sendSocketId;
        m_options.bAllowDelayedDelivery = EOS_TRUE;
        m_options.Channel = channel;
        m_options.Reliability = EOS_EPacketReliability::EOS_PR_ReliableOrdered;
        m_options.bDisableAutoAcceptConnection = EOS_FALSE;
    }
    void send(const std::string &value) {
        sendPacket_( value );
    }
};
} // namespace Sender

Receiver/RecvText.h

#pragma once
namespace Receiver {
class RecvText {
    const Ctx m_ctx;
    const EOS_HP2P m_p2PHandle;
    EOS_P2P_SocketId m_receiveSocketId;
    EOS_P2P_ReceivePacketOptions m_options;
    const uint8_t m_channel = 0;
    // If NULL, we're retrieving the size of the next packet on any channel
    const uint8_t* m_requestedChannel = &m_channel;

public:
    RecvText(const Ctx &ctx, uint8_t channel) :
        m_ctx( ctx )
        , m_p2PHandle( ::EOS_Platform_GetP2PInterface( ctx.m_PlatformHandle ) )
        , m_receiveSocketId{ EOS_P2P_SOCKETID_API_LATEST }
        , m_options{ EOS_P2P_RECEIVEPACKET_API_LATEST }
        , m_channel( channel )
    {
        strcpy_s( m_receiveSocketId.SocketName , m_ctx.m_SocketName.c_str( ) );
        m_options.LocalUserId = m_ctx.m_LocalUserId;
        m_options.MaxDataSizeBytes = Networking::c_MaxDataSizeBytes;
        m_options.RequestedChannel = m_requestedChannel;
    }

    Networking::messageData_t receive(size_t len) {
        EOS_ProductUserId unused_;
        Networking::messageData_t messageData( m_options.MaxDataSizeBytes );
        uint32_t bytesWritten = 0;
        uint8_t outChannel = 0;
        EOS_EResult result = ::EOS_P2P_ReceivePacket( m_p2PHandle, &m_options
            , &unused_, &m_receiveSocketId, &outChannel, messageData.data( ), &bytesWritten );
        // error checking has been ommited for "codereview"
        if ( EOS_EResult::EOS_NotFound == result ) 
            return { };
        messageData.resize( bytesWritten );
        return messageData;
    }
};
} // namespace Receiver

Sending.h

#pragma once
class Sending {
    Ctx m_ctx;
    uint8_t m_channel;

public:
    Sending(const Ctx &ctx, uint8_t channel = 0) : 
        m_ctx( ctx )
        , m_channel( channel )
    {}
    void text(const std::string &text) {
        auto executor = std::make_unique< Sender::SendText >( m_ctx, m_channel );
        auto command = detail_::make_action(
                QueueCommands::Direction::Outgoing
                , [text] (const std::unique_ptr< Sender::SendText > &p) { 
                    p ->send( text );
                    return Networking::messageData_t{ };
                }
                , std::move( executor )
            );
        QueueCommands::instance( ).push( command );
    }
    // auto vector(const Networking::messageData_t &vector) { ... }
};

Receiving.h

#pragma once
class Receiving {
    Ctx m_ctx;
    uint8_t m_channel;

public:
    Receiving(const Ctx &ctx, uint8_t channel = 0) : 
        m_ctx( ctx )
        , m_channel( channel )
    {}
    void text(size_t len) {
        auto executor = std::make_unique< Receiver::RecvText >( m_ctx, m_channel );
        auto accumulator = std::make_unique< Networking::messageData_t >( );
        auto command = detail_::make_action(
                QueueCommands::Direction::Incoming
                , [len] (
                    const std::unique_ptr< Receiver::RecvText > &executor
                    , const std::unique_ptr< Networking::messageData_t > &buf
                ) { 
                    Networking::messageData_t messageData = executor ->receive( len );
                    if ( !messageData.empty( ) && messageData.size( ) != len ) {
                        // accumulate
                        std::copy( messageData.begin( ), messageData.end( ),  std::back_inserter( *buf ) );
                        // until requirement size
                        if ( buf ->size( ) == len )
                            return *buf;
                        return Networking::messageData_t{ };
                    }
                    return messageData;
                }
                , std::move( executor )
                , std::move( accumulator )
            );
        QueueCommands::instance( ).push( command );
    }
    //auto vector(size_t len) { ... }
};

demo main.cpp

#include "pch.h"
#include "Deferred/QueueCommands.h"
#include "Deferred/Action.h"
#include "Deferred/Sender/SendText.h"
#include "Deferred/Receiver/RecvText.h"
#include "Deferred/Sending.h"
#include "Deferred/Receiving.h"

int main(int argc, char *argv[]) {
    bool isServer = ( argc > 1 );
    Ctx ctx = createContext( isServer );
    QueueCommands::init( ctx.m_PlatformHandle );
    ConnectionRequestListener::AcceptEveryone acceptEveryone( ctx );
    const std::string text0 = "ping1", text1 = "ping2";
    if ( isServer ) {
        Receiving receiving( ctx );
        receiving.text( text0.length( ) );
        receiving.text( text1.length( ) );
        auto incoming = QueueCommands::instance( ).ticksAll( );
        assert( ( text0 == std::string( incoming[ 0 ].begin( ), incoming[ 0 ].end( ) ) ) );
        assert( ( text1 == std::string( incoming[ 1 ].begin( ), incoming[ 1 ].end( ) ) ) );
    } else {
        Sending sending( ctx );
        sending.text( text0 );
        sending.text( text1 );
        QueueCommands::instance( ).ticksAll( );
    }
    return printf( "press [Enter] to exit" ), getchar( );
}

Synchronization of access to methods is intentionally not implemented here; The topic of multithreading support will be discussed later. Intentionally omitted handling of critical call errors and UDP deficiencies. I know that using a template in the class Action (@insp SO) increases the size of the binary file, but willing to sacrifice that for the sake of the usability of the deducer detail_::make_action.

I chose the deferred sending/receiving approach because of the need to cause "ticks" for the network to work, but we had to accumulate sending and receiving commands somewhere. Queue singleton to select to reduce the number of arguments passed between functions, besides, if there is only one tick, then the first thing that comes to mind is the queue in a single copy, although I am against global variables and this review could have been done without them (shared_ptr are passed between all participants...).

Full code on GitHub repo

Regards, Alex0vSky

\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

Separate concerns

Try to separate concerns in your code where possible. For example, your QueueCommands deals with too many things at the same time: it is a queue, it knows about commands, and it deals with network I/O. It's better to have it just be a queue, all other functions can be handled outside the class.

Avoid singletons

While you might only need one command queue in your program, that does not mean it has to be a singleton class. You could just declare a single instance of the queue, either as a global variable, as a local variable of which you pass a reference to other objects and functions as necessary, or perhaps it's best part of another object.

For example, in most places where you call QueueCommands::instance(), you also have a reference to a Ctx. Maybe you can make the queue part of Ctx?

Avoid making everything a class

There is no need in C++ to put everything into a class. In many cases, free functions are perfectly fine. For example, Sending and Receiving are classes with just one member function text(). Why not pass the context and channel to text()? In that case you don't need member variables anymore, and text() can be made a free function. For example:

void sendText(const Ctx &ctx, uint8_t channel, const std::string &text) {
    …
}

void receiveText(const Ctx &ctx, uint8_t channel, std::size_t len) {
    …
}

Classes should not be used for actions, they should be for objects. If anything, you could make a class Channel, that represents a channel as an object, and then have send() and receive() member functions:

class Channel {
    …
public:
    Channel(const Ctx &ctx, uint8_t channel): … {}
    void send(const std::string &text) {…}
    void receive(std::size_t len) {…}
};

You are also needlessly complicating things by having SendText and ReceiveText classes. These might have some value if objects are constructed once, and send() and receive() called many times on them, but they seem just to be used for one-shot sends and receives. Again, why not make them free functions to do the low-level send and receive?

Use std::function

Apart from having a Direction, an Action is just a reimplementation of std::function. Just use std::function instead, and add whatever you need on top of that. Consider:

struct Command {
    Direction m_direction;
    std::function<Networking::messageData_t(bool)> action;
};

struct Ctx {
    …
    std::queue<Command> commandQueue;

    auto processAll() {
        std::vector<Networking::messageData_t> allIncomingData;

        while (!commandQueue.empty()) {
            auto& command = commandQueue.front();
            if (Direction::Outgoing == command.direction) {
                command.action(m_avoidPush);
                …
            } else {
                …
            }
        }

        return allIncomingData;
    }
    …
};

And to send something:

class Channel {
    Ctx &m_ctx;
    …
public:
    …
    void send(const std::string &text) {
        m_ctx.commandQueue.emplace_back(Direction::Outgoing,
           [&m_ctx, m_channel, text] { sendText(m_ctx, m_channel, text); }
        );
    }
};

And finally, the low-level send function itself can look like this:

void sendText(const Ctx &ctx, uint8_t channel, const std::string &text) {
    EOS_P2P_SendPacketOptions m_options;
    m_options.LocalUserId = m_ctx.m_LocalUserId;
    …
    while (…) {
        …
        ::EOS_P2P_SendPacket(ctx.m_PlatformHandle, m_options);
        …
    }
}

Consider using std::packaged_task

If you are receiving data, you collect the received data into the vector allIncomingData, which is returned from ticksAll(). It's easy if you only receive a few things when calling ticksAll(), but if your code gets more complicated and wants to receive lots of things between calls, or if it doesn't exactly know when ticksAll() gets called, it can become messy to get that data. Wouldn't it be much nicer if you could return the data somehow directly from the send call, but in such a way you get the results asynchronously? That's what std::future is made for. Your calling code could then look like this:

Ctx ctx = …;
…
if (isServer) {
    result0 = ctx.receive(text0.length());
    result1 = ctx.receive(text1.length());
    ctx.processAll();
    assert(text0 == result0.get());
    assert(text1 == result1.get());
}

Instead of storing std::functions in the queue, you should then store std::packaged_tasks instead. The drawback of this is that you can forget to call processAll() before trying to get the result, which would then cause get() to wait indefinitely for the future result to appear. If you had a separate thread processing commands, that would not be an issue.

\$\endgroup\$
4
  • \$\begingroup\$ I really liked the final part, I will try to implement it as soon as possible and first thing. That's what my thoughts revolved around. Most likely, this is exactly what was needed and what was missing. Thanks! \$\endgroup\$
    – Alexovsky
    Nov 26, 2023 at 12:49
  • \$\begingroup\$ Continued in asynchronous. I always strive for "Separate concerns", but I don’t always have enough concentration at the moment when it is not known where the architecture will turn, my mistake. As I wrote in the question, I try to avoid singletons and global variables, I didn’t have enough determination, my mistake. \$\endgroup\$
    – Alexovsky
    Dec 3, 2023 at 11:21
  • \$\begingroup\$ I didnt want to clutter the Ctx; When developing API access, you often need to store many variables in one place, the purpose of which is not yet clear. In my opinion, POD/Aggregate is ideal for this. In the case of Sending/Reception and SendText/ReceiveText I tried to semantically separate: the controversial creation functionality and the clear execution functionality. My mistake is that I did not fully resolve the controversial issues. Regarding for the "free function", I chose classes to reduce the number of call arguments. \$\endgroup\$
    – Alexovsky
    Dec 3, 2023 at 11:22
  • \$\begingroup\$ I also have a habit of avoiding "free function" because at the unitTest stage, the fixture is more accustomed to operating with struct/class and it is easier for me to locate and name tests. About std::function => std::packaged_task is brilliant. It was very helpful and I hope it will work and thank you. \$\endgroup\$
    – Alexovsky
    Dec 3, 2023 at 11:23

Your Answer

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

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