Boost.Asio Server e RAII

Question

Estou tentando implementar um aplicativo de servidor de rede em C ++ usando Boost.Asio.

Aqui estão os requisitos que estou tentando atender:

• O aplicativo cria apenas uma instância de boost::io_context.
• Single io_contextestá sendo run()por um Thread Pool compartilhado. O número de threads não está definido.
• O aplicativo pode instanciar vários objetos de servidor. Novos servidores podem ser gerados e eliminados a qualquer momento.
• Cada servidor pode lidar com conexões de vários clientes.

Estou tentando implementar o padrão RAII para a classe Server. O que quero garantir é que, quando o servidor for desalocado, todas as suas conexões serão completamente fechadas. Cada conexão pode ser fechada de 3 maneiras:

1. O cliente responde e não há mais trabalho a ser feito em uma conexão.
2. O servidor está sendo desalocado e faz com que todas as conexões ativas sejam fechadas.
3. A conexão é encerrada manualmente invocando o stop()método.

Cheguei a uma solução que parece atender a todos os critérios acima, mas como Boost.Asio ainda é muito novo para mim, queria verificar se o que estou fazendo está correto. Além disso, há algumas coisas sobre as quais eu não estava 100% certo:

• Eu estava tentando remover o mutexda classe Servidor e, em vez disso, usar um strandpara toda a sincronização, mas não consegui encontrar uma maneira clara de fazer isso.
• Porque Thread Pool pode consistir em apenas 1 thread e este thread pode ser o que está chamando um destruidor de servidor, eu tive que invocar a io_context::poll_one()partir do destruidor para dar uma chance para todas as conexões pendentes completarem o desligamento e prevenir um possível deadlock.
• Eu agradeceria qualquer outra sugestão de melhorias que você possa imaginar.

Enfim, aqui está o código com alguns testes de unidade (versão ao vivo no 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();
}

Atenciosamente, Marek

Answer 1

Não conheço Asio o suficiente para dar o tipo de feedback que sei que você quer, mas aqui estão algumas pequenas limpezas que você pode fazer:

• Não using namespace std. Você provavelmente também deve evitar using namespacequalquer outra coisa, apenas para maior clareza.

• virtual ~ConnectionDelegate() { }poderia ser em virtual ~ConnectionDelegate() = default;vez disso. Isso representa um pouco melhor sua intenção.

• ~Server()deve ser ~Server() override, para indicar que ele substitui uma função de membro virtual. Em geral, você deve usar overridesempre que for fisicamente permitido pelo idioma. (Eu acho que você faz isso certo em qualquer lugar, exceto nos destruidores.)

• Connection(string name,e Server(string name,ambos fazem uma cópia desnecessária de string name.

• Todos os seus construtores devem ser explicit, para dizer ao compilador que, por exemplo, o par de chaves {"hello world", myIOContext}não deve ser tratado implicitamente como (ou convertido implicitamente em) um Serverobjeto, nem mesmo por acidente.

• Pessoalmente, acho o uso de typedefs para ConnectionsListe ConnectionsIteratorMapuma camada desnecessária de indireção. Eu prefiro ver std::list<std::weak_ptr<Connection>> _connections;ali na linha. Se eu precisar de um nome para esse tipo, posso apenas dizer decltype(_connections).

• default_delete<Connection>()(connection)é uma forma prolixa de dizer delete connection. Seja direto.

• class executor executor()é muito confuso. O fato de você ter que dizer classque deveria haver um sinalizador vermelho que ou executornão é o nome certo para a classe, ou executor()não é o nome certo para este método. Considere alterar o nome do método para get_executor(), por exemplo. Presumo que você não pode alterar o nome de class executorporque ele não está declarado neste arquivo; deve vir de algum namespace Boost que você usingeditou, certo? (Não usinguse namespaces!)

---

Você pula muitas oportunidades para evitar cópias por meio de referências e / ou semântica de movimentação. Por exemplo, em Server::connections(), eu teria 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;
}

Isso evita colidir com o refcount fraco fazendo weakConnectionuma referência em vez de uma cópia e, em seguida, evita colidir com o refcount forte usando mover em vez de copiar em push_back. Quatro operações atômicas salvas! (Não que isso importe na vida real, provavelmente, mas hey, bem-vindo à revisão do código.)

---

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

Acho o uso de bindconfuso, mas não sei ao certo (e na verdade espero que alguém comente e me esclareça) - é bind necessário aqui? Certamente seria mais claro para ler, mais rápido para compilar e não mais lento no tempo de execução para escrever

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

Isso deixaria um pouco mais claro o que está realmente sendo copiado (um shared_ptrmantendo *thisvivo e um ponteiro bruto). Na verdade, eu me pergunto se precisamos mesmo esconder a cópia do ponteiro bruto; poderíamos escapar impunes em vez disso?

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

Ou você espera que às vezes entre no corpo daquele lambda com d != self->_delegatee é por isso que precisa do ponteiro extra?

---

Também me pergunto se seria possível usar em std::chrono::secondsvez de boost::posix_time::seconds. O Boost.Asio pode interoperar com o C ++ 11 std::chronoatualmente?

---

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

Eu sinto que a "inteligência" aqui está do lado errado do sinal de igual. _connections.insert(_connections.end(), connection)parece uma forma prolixa de escrever _connections.push_back(connection). Vice-versa, eu estou acostumado a ver as pessoas substituir map[k] = vcom map.emplace(k, v)o desempenho e clareza. Lembre-se de que map[k] = vprimeiro constrói o padrão e map[k] , em seguida, atribui um novo valor a ele.

Ah, entendo, você precisa usar insertporque insertretorna um iterador e push_backnão retorna .

Mas isso apenas levanta a questão: por que você está tentando juntar dois efeitos colaterais em uma linha? Se temos permissão para duas linhas, apenas fazemos push_backe depois configuramos map.emplace(connection.get(), std::prev(_connections.end())). Ou, diabos, nesse ponto eu não iria reclamar

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

Depois de localizar a bandeira vermelha, vá mais fundo: qual é a diferença entre o one-liner e o mais claro two-liner? Aha! A diferença é o que acontece se _connections.insert(...)ficar sem memória e jogar. Com o two-liner, _connectionIteratorspermanece intocado. Com o one-liner, você primeiro constrói por padrão algum lixo perigoso _connectionIterators[connection.get()]e depois propaga a exceção.

Portanto, acho que há um argumento razoável a ser feito a favor das duas linhas, apenas em princípios gerais.

---

Novamente, esta resposta realmente não aborda sua principal preocupação sobre RAII, mas espero que dê algum motivo para reflexão de qualquer maneira.