Boost.Asio Server i RAII

Question

Próbuję zaimplementować aplikację serwera sieciowego w C ++ przy użyciu Boost.Asio.

Oto wymagania, które staram się spełnić:

• Aplikacja tworzy tylko jedno wystąpienie boost::io_context.
• Pojedynczy io_contextto run()współdzielona pula wątków. Liczba wątków nie jest zdefiniowana.
• Aplikacja może tworzyć instancje wielu obiektów serwera. Nowe serwery można odrodzić i zabić w dowolnym momencie.
• Każdy serwer może obsługiwać połączenia od wielu klientów.

Próbuję zaimplementować wzorzec RAII dla klasy Server. Chcę zagwarantować, że gdy serwer zostanie zwolniony, wszystkie jego połączenia zostaną całkowicie zamknięte. Każde połączenie można zamknąć na 3 sposoby:

1. Klient odpowiada i nie ma więcej pracy do wykonania w połączeniu.
2. Serwer jest zwalniany i powoduje zamknięcie wszystkich aktywnych połączeń.
3. Połączenie jest przerywane ręcznie przez wywołanie stop()metody.

Dotarłem do rozwiązania, które wydaje się spełniać wszystkie powyższe kryteria, ale ponieważ Boost.Asio jest dla mnie wciąż całkiem nowy, chciałem sprawdzić, czy to, co robię, jest poprawne. Jest też kilka rzeczy, których nie byłem pewien w 100%:

• Próbowałem usunąć mutexz klasy Server i zamiast tego użyć stranddo całej synchronizacji, ale nie mogłem znaleźć jasnego sposobu, aby to zrobić.
• Ponieważ Pula wątków może składać się tylko z 1 wątku, a ten wątek może być tym, co wywołuje destruktor serwera, musiałem wywołać io_context::poll_one()z destruktora, aby dać szansę wszystkim oczekującym połączeniom na zakończenie zamknięcia i zapobieżenie potencjalnemu zakleszczeniu.
• Byłbym zadowolony z wszelkich innych propozycji ulepszeń, o których możesz pomyśleć.

Tak czy inaczej, oto kod z niektórymi testami jednostkowymi (wersja na żywo na 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();
}

Z poważaniem Marek

Answer 1

Nie znam wystarczająco Asio, aby przekazać informacje zwrotne, o których wiem, że chcesz, ale oto kilka drobnych porządków, które możesz zrobić:

• Nie using namespace std. Prawdopodobnie powinieneś również unikać using namespaceczegokolwiek innego, dla jasności.

• virtual ~ConnectionDelegate() { }może być virtual ~ConnectionDelegate() = default;zamiast tego. To trochę lepiej oddaje twoje zamiary.

• ~Server()powinien być ~Server() override, aby wskazać, że przesłania on wirtualną funkcję składową. Ogólnie rzecz biorąc, powinieneś używać overridewszędzie tam, gdzie jest to fizycznie dozwolone przez język. (Myślę, że robisz to dobrze wszędzie oprócz destruktorów.)

• Connection(string name,i Server(string name,oba niepotrzebnie tworzą kopię string name.

• Wszystkie konstruktory powinny być explicit, aby powiedzieć kompilatorowi, że np. Para stężeń {"hello world", myIOContext}nie powinna być niejawnie traktowana (lub niejawnie konwertowana) jako Serverobiekt, nawet przez przypadek.

• Osobiście uważam, że za korzystanie z typedefs ConnectionsListi ConnectionsIteratorMapbyć zbędnym warstwę zadnie. Wolałbym raczej widzieć std::list<std::weak_ptr<Connection>> _connections;w kolejce. Jeśli potrzebuję nazwy dla tego typu, mogę po prostu powiedzieć decltype(_connections).

• default_delete<Connection>()(connection)jest rozwlekłym sposobem powiedzenia delete connection. Bądź bezpośredni.

• class executor executor()jest mylące. Fakt, że musiałeś powiedzieć, classże powinna być czerwona flaga, która albo executornie jest właściwą nazwą dla klasy, albo executor()nie jest właściwą nazwą dla tej metody. Rozważ zmianę nazwy metody get_executor()na przykład na. Zakładam, że nie możesz zmienić nazwy, class executorponieważ nie jest zadeklarowana w tym pliku; musi pochodzić z jakiejś przestrzeni nazw Boost, którą usingstworzyłeś, prawda? (Nie usingprzestrzeni nazw!)

---

Pomijasz wiele okazji, aby uniknąć kopiowania poprzez odwołania i / lub przenoszenie semantyki. Na przykład w Server::connections(), napisałbym:

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

Pozwala to uniknąć uderzania o słabe refcount, tworząc weakConnectionodwołanie zamiast kopii, a następnie unika uderzania silnego refcount przez użycie move zamiast copy in push_back. Uratowano cztery atomowe operacje! (Prawdopodobnie nie ma to znaczenia w prawdziwym życiu, ale hej, witamy w Code Review.)

---

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

Uważam, że użycie jest bindmylące, ale nie wiem na pewno (i faktycznie mam nadzieję, że ktoś mnie skomentuje i oświeci) - jest tu bind potrzebne ? Z pewnością byłby bardziej przejrzysty do odczytania, szybszy w kompilacji i nie wolniejszy w czasie wykonywania

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

Dzięki temu byłoby trochę jaśniej, co jest kopiowane (jeden shared_ptrutrzymuje się *thisprzy życiu, a jeden nieprzetworzony wskaźnik). W rzeczywistości zastanawiam się, czy w ogóle musimy przechowywać kopię surowego wskaźnika; czy możemy zamiast tego uciec?

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

A może spodziewasz się, że czasami dostaniesz się do ciała tej lambdy z d != self->_delegatei dlatego potrzebujesz dodatkowego wskaźnika?

---

Zastanawiam się też, czy można by użyć std::chrono::secondszamiast boost::posix_time::seconds. Czy Boost.Asio może współdziałać z C ++ 11 w std::chronodzisiejszych czasach?

---

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

Wydaje mi się, że „spryt” znajduje się po złej stronie znaku równości. _connections.insert(_connections.end(), connection)wydaje się być rozwlekłym sposobem pisania _connections.push_back(connection). I odwrotnie, jestem przyzwyczajony do tego, że ludzie zastępują map[k] = vsię map.emplace(k, v)wydajnością i przejrzystością. Pamiętaj, że map[k] = vnajpierw konstruuje domyślne map[k] , a następnie przypisuje do niej nową wartość.

Ach, widzę, musisz użyć, insertponieważ insertzwraca iterator, a push_backnie.

Ale to tylko nasuwa pytanie: dlaczego próbujesz połączyć dwa efekty uboczne w jedną linię? Jeśli mamy dwie linie, po prostu robimy, push_backa następnie ustawiamy map.emplace(connection.get(), std::prev(_connections.end())). Albo, do cholery, w tym momencie nie narzekałbym tak naprawdę

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

Widząc czerwoną flagę, sięgnij głębiej: jaka jest różnica między linią jednowarstwową a wyraźniejszą linią dwurzędową? Aha! Różnica polega na tym, co się dzieje, gdy _connections.insert(...)zabraknie pamięci i wyrzuca. Dzięki dwuwarstwowej _connectionIteratorspowłoce pozostaje nietknięta. Z jednowierszowym tekstem najpierw domyślnie konstruujesz jakieś niebezpieczne śmieci, _connectionIterators[connection.get()]a następnie propagujesz wyjątek.

Więc myślę, że jest rozsądny argument przemawiający za dwuliniowością, tylko na ogólnych zasadach.

---

Ponownie, ta odpowiedź tak naprawdę nie odnosi się do twojego głównego problemu związanego z RAII, ale mam nadzieję, że i tak daje trochę do myślenia.