Boost.Asio Server y RAII

Question

Estoy tratando de implementar una aplicación de servidor de red en C ++ usando Boost.Asio.

Estos son los requisitos que estoy tratando de cumplir:

• La aplicación crea solo una instancia de boost::io_context.
• Single io_contextestá siendo run()compartido por un Thread Pool. El número de hilos no está definido.
• La aplicación puede crear instancias de varios objetos de servidor. Se pueden generar y eliminar nuevos servidores en cualquier momento.
• Cada servidor puede manejar conexiones de varios clientes.

Estoy tratando de implementar el patrón RAII para la clase Server. Lo que quiero garantizar es que cuando el servidor se desasigne, todas sus conexiones estén completamente cerradas. Hay tres formas de cerrar cada conexión:

1. El cliente responde y no hay más trabajo por hacer en una conexión.
2. El servidor está siendo desasignado y hace que todas las conexiones activas se cierren.
3. La conexión se mata manualmente invocando el stop()método.

Llegué a una solución que parece cumplir con todos los criterios anteriores, pero como Boost.Asio es todavía bastante nuevo para mí, quería verificar que lo que estoy haciendo es correcto. También hay un par de cosas de las que específicamente no estaba 100% seguro:

• Estaba tratando de eliminar el mutexde la clase Server y en su lugar usar un strandpara toda la sincronización, pero no pude encontrar una manera clara de hacerlo.
• Debido a que Thread Pool puede constar de solo 1 subproceso y este subproceso puede ser lo que llama a un destructor de servidor, tuve que invocar io_context::poll_one()desde el destructor para dar una oportunidad a todas las conexiones pendientes para completar el apagado y evitar un posible punto muerto.
• Agradecería cualquier otra sugerencia de mejora que se le ocurra.

De todos modos, aquí está el código con algunas pruebas unitarias (versión en vivo en 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();
}

Saludos cordiales, Marek

Answer 1

No conozco lo suficiente sobre Asio para dar el tipo de retroalimentación que sé que desea, pero aquí hay algunas limpiezas menores que podría hacer:

• No lo hagas using namespace std. Probablemente también debería evitar using namespacecualquier otra cosa, solo por claridad.

• virtual ~ConnectionDelegate() { }podría ser virtual ~ConnectionDelegate() = default;en su lugar. Esto representa tu intención un poco mejor.

• ~Server()debería ser ~Server() override, para indicar que anula una función de miembro virtual. En general, debe usar overridedonde esté físicamente permitido por el idioma. (Creo que lo haces bien en todas partes excepto en los destructores).

• Connection(string name,y Server(string name,ambos hacen innecesariamente una copia de string name.

• Todos sus constructores deberían ser explicit, para decirle al compilador que, por ejemplo, el par entre corchetes {"hello world", myIOContext}no debería tratarse implícitamente (o convertirse implícitamente en) un Serverobjeto, ni siquiera por accidente.

• Personalmente, considero que el uso de typedefs para ConnectionsListy ConnectionsIteratorMapes una capa innecesaria de indirección. Preferiría verlo std::list<std::weak_ptr<Connection>> _connections;en línea. Si necesito un nombre para ese tipo, solo puedo decirlo decltype(_connections).

• default_delete<Connection>()(connection)es una forma prolija de decir delete connection. Sé directo.

• class executor executor()es muy confuso. El hecho de que tuviera que decir classque debería haber habido una señal de alerta de que executorno es el nombre correcto para la clase o executor()no es el nombre correcto para este método. Considere cambiar el nombre del método a get_executor(), por ejemplo. Supongo que no puede cambiar el nombre de class executorporque no está declarado en este archivo; debe provenir de algún espacio de nombres de Boost que usingeditaste, ¿verdad? (¡No usinguses espacios de nombres!)

---

Omite muchas oportunidades para evitar copias a través de referencias y / o semántica de movimientos. Por ejemplo, en Server::connections(), habría escrito:

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

Esto evita golpear el refcount débil haciendo weakConnectionuna referencia en lugar de una copia, y luego evita golpear el refcount fuerte usando mover en lugar de copiar push_back. ¡Cuatro operaciones atómicas salvadas! (No es que esto importe en la vida real, probablemente, pero bueno, bienvenido a Code Review).

---

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

Encuentro el uso de bindconfuso, pero no lo sé con certeza (y de hecho espero que alguien me comente e ilumine) - ¿se bind necesita aquí? Sin duda, sería más claro de leer, más rápido de compilar y no más lento en tiempo de ejecución para escribir

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

Esto aclararía un poco lo que realmente se está copiando (uno que se shared_ptrmantiene *thisvivo y un puntero sin formato). De hecho, me pregunto si necesitamos guardar la copia del puntero sin formato; ¿Podríamos salirse con la nuestra en su lugar?

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

¿O esperas que a veces te metas en el cuerpo de esa lambda d != self->_delegatey es por eso que necesitas el puntero adicional?

---

También me pregunto si sería posible usar en std::chrono::secondslugar de boost::posix_time::seconds. ¿Puede Boost.Asio interoperar con C ++ 11 en std::chronoestos días?

---

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

Siento que la "inteligencia" aquí está en el lado equivocado del signo igual. _connections.insert(_connections.end(), connection)parece una forma prolija de escribir _connections.push_back(connection). Viceversa, yo estoy acostumbrado a ver gente reemplaza map[k] = vcon map.emplace(k, v)el rendimiento y la claridad. Recuerde que map[k] = vprimero se construye por defecto map[k] y luego se le asigna un nuevo valor.

Ah, ya veo, necesitas usar insertporque insertdevuelve un iterador y push_backno lo hace.

Pero eso solo plantea la pregunta: ¿Por qué está tratando de calzar dos efectos secundarios en una sola línea? Si se nos permiten dos líneas, simplemente lo hacemos push_backy luego establecemos map.emplace(connection.get(), std::prev(_connections.end())). O, diablos, en ese momento realmente no me quejaría

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

Habiendo visto la bandera roja, profundice: ¿cuál es la diferencia entre el uno y el dos líneas más claro? ¡Ajá! La diferencia es lo que pasa si se _connections.insert(...)queda sin memoria y tira. Con el de dos liner, _connectionIteratorspermanece intacto. Con el one-liner, primero construye por defecto alguna basura peligrosa _connectionIterators[connection.get()]y luego propaga la excepción.

Así que creo que hay un argumento razonable a favor de la doble línea, solo por principios generales.

---

Una vez más, esta respuesta realmente no aborda su principal preocupación sobre RAII, pero espero que de todos modos dé algo en que pensar.