Boost.Asio Server et RAII

Question

J'essaye d'implémenter une application de serveur de réseau en C ++ en utilisant Boost.Asio.

Voici les exigences que j'essaie de satisfaire:

• L'application ne crée qu'une seule instance de boost::io_context.
• Le single io_contextest run()par un pool de threads partagé. Le nombre de threads n'est pas défini.
• L'application peut instancier plusieurs objets Serveur. De nouveaux serveurs peuvent être créés et tués à tout moment.
• Chaque serveur peut gérer les connexions de plusieurs clients.

J'essaye d'implémenter le modèle RAII pour la classe de serveur. Ce que je veux garantir, c'est que lorsque le serveur est désalloué, toutes ses connexions sont complètement fermées. Il existe 3 façons de fermer chaque connexion:

1. Le client répond et il n'y a plus de travail à faire dans une connexion.
2. Le serveur est en cours de désaffectation et entraîne la fermeture de toutes les connexions actives.
3. La connexion est supprimée manuellement en appelant la stop()méthode.

Je suis arrivé à une solution qui semble répondre à tous les critères ci-dessus mais depuis Boost.Asio est encore assez nouveau pour moi, je voulais vérifier que ce que je fais est correct. Il y a aussi deux choses dont je n'étais pas sûr à 100%:

• J'essayais de supprimer le mutexde la classe Serveur et d'utiliser à la place un strandpour toute la synchronisation, mais je ne pouvais pas trouver un moyen clair de le faire.
• Étant donné que le pool de threads ne peut comprendre qu'un seul thread et que ce thread peut être ce qui appelle un destructeur de serveur, j'ai dû appeler à io_context::poll_one()partir du destructeur pour donner une chance à toutes les connexions en attente de terminer l'arrêt et d'éviter un blocage potentiel.
• Je serais heureux de recevoir toute autre suggestion d’améliorations auxquelles vous pourriez penser.

Bref, voici le code avec quelques tests unitaires (version live sur Coliru: http://coliru.stacked-crooked.com/a/1afb0dc34dd09008 ):

#include <boost/asio/io_context.hpp>
#include <boost/asio/io_context_strand.hpp>
#include <boost/asio/executor.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/asio/dispatch.hpp>
#include <iostream>
#include <string>
#include <vector>
#include <memory>
#include <list>
using namespace std;
using namespace boost::asio;
using namespace std::placeholders;


class Connection;


class ConnectionDelegate
{
public:
    virtual ~ConnectionDelegate() { }
    
    virtual class executor executor() const = 0;
    virtual void didReceiveResponse(shared_ptr<Connection> connection) = 0;
};


class Connection: public enable_shared_from_this<Connection>
{
public:
    Connection(string name, io_context& ioContext)
    : _name(name)
    , _ioContext(ioContext)
    , _timer(ioContext)
    {
    }
    
    const string& name() const
    {
        return _name;
    }
    void setDelegate(ConnectionDelegate *delegate)
    {
        _delegate = delegate;
    }
    
    void start()
    {
        // Simulate a network request
        _timer.expires_from_now(boost::posix_time::seconds(3));
        _timer.async_wait(bind(&Connection::handleResponse, shared_from_this(), _1));
    }
    void stop()
    {
        _timer.cancel();
    }
    
private:
    string _name;
    io_context& _ioContext;
    boost::asio::deadline_timer _timer;
    ConnectionDelegate *_delegate;
    
    void handleResponse(const boost::system::error_code& errorCode)
    {
        if (errorCode == error::operation_aborted)
        {
            return;
        }
        dispatch(_delegate->executor(),
                 bind(&ConnectionDelegate::didReceiveResponse, _delegate, shared_from_this()));
    }
};


class Server: public ConnectionDelegate
{
public:
    Server(string name, io_context& ioContext)
    : _name(name)
    , _ioContext(ioContext)
    , _strand(_ioContext)
    {
    }
    ~Server()
    {
        stop();
        assert(_connections.empty());
        assert(_connectionIterators.empty());
    }
    weak_ptr<Connection> addConnection(string name)
    {
        auto connection = shared_ptr<Connection>(new Connection(name, _ioContext), bind(&Server::deleteConnection, this, _1));
        {
            lock_guard<mutex> lock(_mutex);
            _connectionIterators[connection.get()] = _connections.insert(_connections.end(), connection);
        }
        connection->setDelegate(this);
        connection->start();
        return connection;
    }
    
    vector<shared_ptr<Connection>> connections()
    {
        lock_guard<mutex> lock(_mutex);
        
        vector<shared_ptr<Connection>> connections;
        for (auto weakConnection: _connections)
        {
            if (auto connection = weakConnection.lock())
            {
                connections.push_back(connection);
            }
        }
        return connections;
    }
    void stop()
    {
        auto connectionsCount = 0;
        for (auto connection: connections())
        {
            ++connectionsCount;
            connection->stop();
        }
        
        while (connectionsCount != 0)
        {
            _ioContext.poll_one();
            connectionsCount = connections().size();
        }
    }
    
    // MARK: - ConnectionDelegate
    class executor executor() const override
    {
        return _strand;
    }
    void didReceiveResponse(shared_ptr<Connection> connection) override
    {
        // Strand to protect shared resourcess to be accessed by this method.
        assert(_strand.running_in_this_thread());
        
        // Here I plan to execute some business logic and I need both Server & Connection to be alive.
        std::cout << "didReceiveResponse - server: " << _name << ", connection: " << connection->name() << endl;
    }
    
private:
    typedef list<weak_ptr<Connection>> ConnectionsList;
    typedef unordered_map<Connection*, ConnectionsList::iterator> ConnectionsIteratorMap;
    
    string _name;
    io_context& _ioContext;
    io_context::strand _strand;
    ConnectionsList _connections;
    ConnectionsIteratorMap _connectionIterators;
    mutex _mutex;
    
    void deleteConnection(Connection *connection)
    {
        {
            lock_guard<mutex> lock(_mutex);
            auto iterator = _connectionIterators[connection];
            _connections.erase(iterator);
            _connectionIterators.erase(connection);
        }
        default_delete<Connection>()(connection);
    }
};


void testConnectionClosedByTheServer()
{
    io_context ioContext;
    auto server = make_unique<Server>("server1", ioContext);
    
    auto weakConnection = server->addConnection("connection1");
    assert(weakConnection.expired() == false);
    assert(server->connections().size() == 1);
    
    server.reset();
    assert(weakConnection.expired() == true);
}

void testConnectionClosedAfterResponse()
{
    io_context ioContext;
    auto server = make_unique<Server>("server1", ioContext);
    
    auto weakConnection = server->addConnection("connection1");
    assert(weakConnection.expired() == false);
    assert(server->connections().size() == 1);
    
    while (!weakConnection.expired())
    {
        ioContext.poll_one();
    }
    assert(server->connections().size() == 0);
}

void testConnectionClosedManually()
{
    io_context ioContext;
    auto server = make_unique<Server>("server1", ioContext);
    
    auto weakConnection = server->addConnection("connection1");
    assert(weakConnection.expired() == false);
    assert(server->connections().size() == 1);
    
    weakConnection.lock()->stop();
    ioContext.run();
    
    assert(weakConnection.expired() == true);
    assert(server->connections().size() == 0);
}

void testMultipleServers()
{
    io_context ioContext;
    auto server1 = make_unique<Server>("server1", ioContext);
    auto server2 = make_unique<Server>("server2", ioContext);

    auto weakConnection1 = server1->addConnection("connection1");
    auto weakConnection2 = server2->addConnection("connection2");

    server1.reset();
    assert(weakConnection1.expired() == true);
    assert(weakConnection2.expired() == false);
}

void testDeadLock()
{
    io_context ioContext;
    auto server = make_unique<Server>("server1", ioContext);
    
    auto weakConnection = server->addConnection("connection1");
    assert(weakConnection.expired() == false);
    assert(server->connections().size() == 1);
    
    auto connection = weakConnection.lock();
    server.reset(); // <-- deadlock, but that's OK, i will try to prevent it by other means
}


int main()
{
    testConnectionClosedByTheServer();
    testConnectionClosedAfterResponse();
    testConnectionClosedManually();
    // testDeadLock();
}

Cordialement, Marek

Answer 1

Je ne connais pas assez Asio pour donner le genre de commentaires que je sais que vous voulez, mais voici quelques nettoyages mineurs que vous pourriez faire:

• Ne fais pas ça using namespace std. Vous devriez probablement éviter using namespacetoute autre chose, juste pour plus de clarté.

• virtual ~ConnectionDelegate() { }pourrait être à la virtual ~ConnectionDelegate() = default;place. Cela représente un peu mieux votre intention.

• ~Server()devrait être ~Server() override, pour indiquer qu'il remplace une fonction membre virtuelle. En général, vous devez utiliser overridelà où cela est physiquement autorisé par la langue. (Je pense que vous le faites partout sauf sur les destructeurs.)

• Connection(string name,et les Server(string name,deux font inutilement une copie de string name.

• Tous vos constructeurs devraient l'être explicit, pour dire au compilateur que, par exemple, la paire accolée {"hello world", myIOContext}ne devrait pas être implicitement traitée comme (ou implicitement convertie en) un Serverobjet, pas même par accident.

• Personnellement, je trouve l'utilisation de typedefs pour ConnectionsListet ConnectionsIteratorMapd'être une couche inutile de indirection. Je préférerais voir std::list<std::weak_ptr<Connection>> _connections;juste là en ligne. Si j'ai besoin d'un nom pour ce type, je peux simplement le dire decltype(_connections).

• default_delete<Connection>()(connection)est une façon verbeuse de dire delete connection. Soyez direct.

• class executor executor()est hella déroutant. Le fait que vous ayez dû dire classqu'il aurait dû y avoir un drapeau rouge qui soit executorn'est pas le bon nom pour la classe, soit executor()n'est pas le bon nom pour cette méthode. Pensez à changer le nom de la méthode en get_executor(), par exemple. Je suppose que vous ne pouvez pas changer le nom de class executorcar il n'est pas déclaré dans ce fichier; ça doit venir d'un espace de noms Boost que vous avez usingédité, non? (Pas d' usingespaces de noms!)

---

Vous sautez de nombreuses opportunités pour éviter les copies via des références et / ou déplacer la sémantique. Par exemple, dans Server::connections(), j'aurais écrit:

std::vector<std::shared_ptr<Connection>> connections() {
    std::lock_guard<std::mutex> lock(_mutex);
    std::vector<std::shared_ptr<Connection>> result;
    for (const auto& weakConnection : _connections) {
        if (auto sptr = weakConnection.lock()) {
            result.push_back(std::move(sptr));
        }
    }
    return result;
}

Cela évite de heurter le refcount faible en créant weakConnectionune référence au lieu d'une copie, puis évite de heurter le refcount fort en utilisant déplacer au lieu de copier dans push_back. Quatre opérations atomiques sauvées! (Pas que cela compte dans la vraie vie, probablement, mais bon, bienvenue dans Code Review.)

---

dispatch(_delegate->executor(),
         bind(&ConnectionDelegate::didReceiveResponse, _delegate, shared_from_this()));

Je trouve l'utilisation binddéroutante, mais je ne sais pas avec certitude (et j'espère en fait que quelqu'un va commenter et m'éclairer) - est-il bind nécessaire ici? Il serait certainement plus clair à lire, plus rapide à compiler et pas plus lent à l'exécution à écrire

dispatch(
    _delegate->executor(),
    [self = shared_from_this(), d = _delegate]() {
        d->didReceiveResponse(self);
    }
);

Cela rendrait un peu plus clair ce qui est réellement copié (un shared_ptrmaintien en *thisvie, et un pointeur brut). En fait, je me demande si nous devons même cacher la copie du pointeur brut; pourrions-nous nous en tirer à la place?

dispatch(
    _delegate->executor(),
    [self = shared_from_this()]() {
        self->_delegate->didReceiveResponse(self);
    }
);

Ou pensez-vous que vous entrerez parfois dans le corps de ce lambda avec d != self->_delegateet c'est pourquoi vous avez besoin du pointeur supplémentaire?

---

Je me demande également s'il serait possible d'utiliser à la std::chrono::secondsplace de boost::posix_time::seconds. Boost.Asio peut-il interagir avec C ++ 11 std::chronoces jours-ci?

---

_connectionIterators[connection.get()] = _connections.insert(_connections.end(), connection);

J'ai l'impression que «l'intelligence» est ici du mauvais côté du signe égal. _connections.insert(_connections.end(), connection)semble être une manière verbeuse d'écrire _connections.push_back(connection). Vice versa, j'ai l'habitude de voir les gens remplacer map[k] = vpar map.emplace(k, v)pour la performance et la clarté. Souvenez-vous que les map[k] = vpremières constructions par défaut map[k] , puis y assignez une nouvelle valeur.

Ah, je vois, vous devez utiliser insertparce que insertrenvoie un itérateur et push_backne le fait pas.

Mais cela soulève simplement la question: pourquoi essayez-vous de regrouper deux effets secondaires en une seule ligne? Si nous avons droit à deux lignes, nous faisons simplement push_backet ensuite définissons map.emplace(connection.get(), std::prev(_connections.end())). Ou, à ce moment-là, je ne me plaindrais pas vraiment

auto it = _connections.insert(_connections.end(), connection);
_connectionIterators.emplace(connection.get(), it);

Après avoir repéré le drapeau rouge, creusez plus profondément: quelle est la différence entre le one-liner et le plus clair deux-liner? Ah! La différence est ce qui se passe s'il _connections.insert(...)manque de mémoire et jette. Avec le two-liner, _connectionIteratorsreste intact. Avec le one-liner, vous construisez d' abord par défaut des déchets dangereux _connectionIterators[connection.get()], puis vous propagez l'exception.

Je pense donc qu'il y a un argument raisonnable à faire en faveur du double ligne, uniquement sur des principes généraux.

---

Encore une fois, cette réponse ne répond pas vraiment à votre principale préoccupation concernant RAII, mais j'espère que cela donne quand même matière à réflexion.