Boost.Asio Server e RAII

Question

Sto cercando di implementare un'applicazione server di rete in C ++ utilizzando Boost.Asio.

Ecco i requisiti che sto cercando di soddisfare:

• L'applicazione crea solo un'istanza di boost::io_context.
• Single io_contextè run()da un pool di thread condiviso. Il numero di thread non è definito.
• L'applicazione può creare un'istanza di più oggetti Server. Nuovi server possono essere generati e uccisi in qualsiasi momento.
• Ogni server può gestire connessioni da più client.

Sto cercando di implementare il pattern RAII per la classe Server. Quello che voglio garantire è che quando il server viene deallocato tutte le sue connessioni sono completamente chiuse. Ci sono 3 modi in cui ogni connessione può essere chiusa:

1. Il client risponde e non c'è più lavoro da fare in una connessione.
2. Il server viene deallocato e causa la chiusura di tutte le connessioni attive.
3. La connessione viene terminata manualmente invocando il stop()metodo.

Sono arrivato a una soluzione che sembra soddisfare tutti i criteri di cui sopra, ma poiché Boost.Asio è ancora abbastanza nuovo per me, volevo verificare che quello che sto facendo sia corretto. Inoltre ci sono un paio di cose di cui specificamente non ero sicuro al 100%:

• Stavo cercando di rimuovere mutexdalla classe Server e invece utilizzare a strandper tutta la sincronizzazione, ma non sono riuscito a trovare un modo chiaro per farlo.
• Poiché il pool di thread può essere costituito da un solo thread e questo thread potrebbe essere ciò che sta chiamando un distruttore del server, ho dovuto invocare io_context::poll_one()dal distruttore per dare una possibilità a tutte le connessioni in sospeso di completare l'arresto e prevenire un potenziale deadlock.
• Sarei lieto di qualsiasi altro suggerimento di miglioramento a cui potresti pensare.

Comunque, ecco il codice con alcuni unit test (versione live su 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();
}

Cordiali saluti, Marek

Answer 1

Non conosco abbastanza Asio per dare il tipo di feedback che so che desideri, ma ecco alcune piccole pulizie che potresti fare:

• Non farlo using namespace std. Probabilmente dovresti evitare anche using namespacequalsiasi altra cosa, solo per chiarezza.

• virtual ~ConnectionDelegate() { }potrebbe essere virtual ~ConnectionDelegate() = default;invece. Questo rappresenta un po 'meglio il tuo intento.

• ~Server()dovrebbe essere ~Server() override, per indicare che sovrascrive una funzione membro virtuale. In generale, dovresti usare overrideovunque sia fisicamente consentito dalla lingua. (Penso che tu lo faccia ovunque tranne che sui distruttori.)

• Connection(string name,ed Server(string name,entrambi fanno inutilmente una copia di string name.

• Tutti i tuoi costruttori dovrebbero essere explicit, per dire al compilatore che, ad esempio, la coppia controvento {"hello world", myIOContext}non dovrebbe essere implicitamente trattata come (o implicitamente convertita in) un Serveroggetto, nemmeno per sbaglio.

• Personalmente, trovo che l'uso di typedef ConnectionsListe ConnectionsIteratorMapsia un livello di riferimento indiretto non necessario. Preferirei vedere std::list<std::weak_ptr<Connection>> _connections;proprio lì in linea. Se ho bisogno di un nome per quel tipo, posso solo dire decltype(_connections).

• default_delete<Connection>()(connection)è un modo prolisso di dire delete connection. Sii diretto.

• class executor executor()è molto confuso. Il fatto che avessi dovuto dire classche avrebbe dovuto esserci una bandiera rossa che o executornon è il nome giusto per la classe o executor()non è il nome giusto per questo metodo. Considera la possibilità di modificare il nome del metodo in get_executor(), ad esempio. Presumo che non sia possibile modificare il nome di class executorperché non è dichiarato in questo file; deve provenire da uno spazio dei nomi Boost che hai usingscritto, giusto? (Non gli usingspazi dei nomi!)

---

Salti molte opportunità per evitare copie tramite riferimenti e / o spostare la semantica. Ad esempio, in Server::connections(), avrei scritto:

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;
}

Questo evita di superare il conteggio dei rif debole facendo weakConnectionun riferimento invece di una copia, e quindi evita di superare il conteggio dei rif forte usando mossa invece di copiare push_back. Quattro operazioni atomiche salvate! (Non che questo abbia importanza nella vita reale, probabilmente, ma ehi, benvenuto in Code Review.)

---

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

Trovo che l'uso di bindconfusione, ma non lo so per certo (e spero davvero che qualcuno mi commenterà e mi illuminerà) - è bind necessario qui? Sarebbe certamente più chiaro da leggere, più veloce da compilare e non più lento in fase di runtime da scrivere

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

In questo modo sarebbe un po 'più chiaro ciò che è effettivamente in fase di copia (uno shared_ptrtenere *thisvivo, e un puntatore RAW). In effetti, mi chiedo se abbiamo anche bisogno di nascondere la copia del puntatore grezzo; potremmo farla franca invece?

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

O ti aspetti che a volte entrerai nel corpo di quel lambda con d != self->_delegateed è per questo che hai bisogno del puntatore extra?

---

Mi chiedo anche se sia possibile utilizzare al std::chrono::secondsposto di boost::posix_time::seconds. Boost.Asio può interagire con C ++ 11 in std::chronoquesti giorni?

---

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

Mi sento come se la "bravura" qui fosse dalla parte sbagliata del segno di uguale. _connections.insert(_connections.end(), connection)sembra un modo prolisso di scrivere _connections.push_back(connection). Viceversa, sono abituato a vedere le persone sostituire map[k] = vcon map.emplace(k, v)per prestazioni e chiarezza. Ricorda che i map[k] = vprimi costrutti di default map[k] , quindi gli assegna un nuovo valore.

Ah, vedo, devi usare insertperché insertrestituisce un iteratore e push_backnon lo fa.

Ma questo solleva solo la domanda: perché stai cercando di mettere due effetti collaterali in una riga? Se ci sono consentite due righe, basta fare push_backe poi impostare map.emplace(connection.get(), std::prev(_connections.end())). O, diamine, a quel punto non mi lamenterei davvero

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

Dopo aver individuato la bandiera rossa, scavare più a fondo: qual è la differenza tra il one-liner e il two-liner più chiaro? Aha! La differenza è cosa succede se si _connections.insert(...)esaurisce la memoria e si lancia. Con le due linee, _connectionIteratorsrimane intatto. Con il one-liner, prima costruisci per impostazione predefinita alcuni rifiuti pericolosi _connectionIterators[connection.get()]e poi propaghi l'eccezione.

Quindi penso che ci sia un argomento ragionevole da fare a favore del due linee, solo sui principi generali.

---

Ancora una volta, questa risposta in realtà non risponde alla tua principale preoccupazione per la RAII, ma spero che ti dia comunque degli spunti di riflessione.