Boost.Asio Server und RAII

Question

Ich versuche, eine Netzwerkserveranwendung in C ++ mit Boost.Asio zu implementieren.

Hier sind die Anforderungen, die ich zu erfüllen versuche:

• Die Anwendung erstellt nur eine Instanz von boost::io_context.
• Single io_contextwird run()von einem gemeinsam genutzten Thread-Pool erstellt. Die Anzahl der Threads ist nicht definiert.
• Die Anwendung kann mehrere Serverobjekte instanziieren. Neue Server können jederzeit erzeugt und getötet werden.
• Jeder Server kann Verbindungen von mehreren Clients verarbeiten.

Ich versuche, ein RAII-Muster für die Serverklasse zu implementieren. Was ich garantieren möchte, ist, dass bei der Freigabe des Servers alle Verbindungen vollständig geschlossen werden. Es gibt drei Möglichkeiten, wie jede Verbindung geschlossen werden kann:

1. Der Client antwortet und es ist keine Arbeit mehr in einer Verbindung zu erledigen.
2. Der Server wird freigegeben und alle aktiven Verbindungen werden geschlossen.
3. Die Verbindung wird manuell durch Aufrufen der stop()Methode beendet.

Ich bin zu einer Lösung gekommen, die alle oben genannten Kriterien zu erfüllen scheint, aber da Boost.Asio für mich noch recht neu ist, wollte ich überprüfen, ob das, was ich tue, korrekt ist. Es gibt auch einige Dinge, bei denen ich mir nicht 100% sicher war:

• Ich habe versucht, das mutexaus der Server-Klasse zu entfernen und stattdessen ein strandfür die gesamte Synchronisation zu verwenden, aber ich konnte keinen klaren Weg finden, dies zu tun.
• Da der Thread-Pool nur aus einem Thread bestehen kann und dieser Thread möglicherweise einen Server-Destruktor aufruft, musste ich ihn io_context::poll_one()vom Destruktor aus aufrufen , um allen ausstehenden Verbindungen die Möglichkeit zu geben, das Herunterfahren abzuschließen und einen möglichen Deadlock zu verhindern.
• Ich würde mich über weitere Verbesserungsvorschläge freuen.

Wie auch immer, hier ist der Code mit einigen Unit-Tests (Live-Version auf 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();
}

Mit freundlichen Grüßen, Marek

Answer 1

Ich kenne Asio nicht genug, um die Art von Feedback zu geben, von der ich weiß, dass Sie sie möchten, aber hier sind einige kleinere Aufräumarbeiten, die Sie durchführen könnten:

• Tu es nicht using namespace std. Sie sollten wahrscheinlich auch using namespacealles andere vermeiden , nur aus Gründen der Klarheit.

• virtual ~ConnectionDelegate() { }könnte virtual ~ConnectionDelegate() = default;stattdessen sein. Dies repräsentiert Ihre Absicht ein wenig besser.

• ~Server()sollte sein ~Server() override, um anzuzeigen, dass es eine virtuelle Mitgliedsfunktion überschreibt. Im Allgemeinen sollten Sie overrideüberall dort verwenden, wo es die Sprache physisch zulässt. (Ich denke, du machst es überall richtig, außer bei den Destruktoren.)

• Connection(string name,und Server(string name,beide machen unnötigerweise eine Kopie von string name.

• Alle Ihre Konstruktoren sollten es sein explicit, um dem Compiler mitzuteilen, dass z. B. das geschweifte Paar {"hello world", myIOContext}nicht implizit als ServerObjekt behandelt (oder implizit in ein Objekt konvertiert) werden sollte, auch nicht aus Versehen.

• Persönlich finde ich die Verwendung von typedefs für ConnectionsListund ConnectionsIteratorMapals unnötige Indirektionsebene. Ich würde lieber std::list<std::weak_ptr<Connection>> _connections;genau dort in der Schlange sehen. Wenn ich einen Namen für diesen Typ brauche, kann ich nur sagen decltype(_connections).

• default_delete<Connection>()(connection)ist eine ausführliche Art zu sagen delete connection. Sei direkt.

• class executor executor()ist hella verwirrend. Die Tatsache, dass Sie sagen mussten, dass classes eine rote Fahne geben sollte, die entweder executornicht der richtige Name für die Klasse oder executor()nicht der richtige Name für diese Methode ist. Ändern Sie beispielsweise den Namen der Methode in get_executor(). Ich gehe davon aus, dass Sie den Namen von nicht ändern können, class executorda er in dieser Datei nicht deklariert ist. Es muss aus einem Boost-Namespace stammen, den Sie usingbearbeitet haben, oder? (Keine usingNamespaces!)

---

Sie überspringen viele Möglichkeiten, um Kopien über Referenzen zu vermeiden und / oder die Semantik zu verschieben. Zum Beispiel Server::connections()hätte ich in geschrieben:

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

Dies vermeidet das Anstoßen des schwachen Nachzählens durch Erstellen weakConnectioneiner Referenz anstelle einer Kopie und vermeidet dann das Anstoßen des starken Nachzählens durch Verwenden von Verschieben anstelle des Einkopierens push_back. Vier Atomoperationen gerettet! (Nicht, dass dies im wirklichen Leben wichtig wäre, wahrscheinlich, aber hey, willkommen bei Code Review.)

---

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

Ich finde die Verwendung bindverwirrend, aber ich weiß es nicht genau (und hoffe tatsächlich, dass mich jemand kommentiert und aufklärt) - wird hier bind gebraucht ? Es wäre sicherlich klarer zu lesen, schneller zu kompilieren und nicht langsamer zur Laufzeit zu schreiben

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

Dies würde ein bisschen klarer machen, was tatsächlich kopiert wird (einer shared_ptr, der *thisam Leben bleibt, und ein roher Zeiger). Tatsächlich frage ich mich, ob wir überhaupt die Kopie des Rohzeigers verstauen müssen; könnten wir stattdessen damit durchkommen?

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

Oder erwarten Sie, dass Sie manchmal mit in den Körper dieses Lambda gelangen d != self->_delegateund deshalb den zusätzlichen Zeiger benötigen?

---

Ich frage mich auch, ob es möglich wäre, std::chrono::secondsstattdessen zu verwenden boost::posix_time::seconds. Kann Boost.Asio heutzutage mit C ++ 11 zusammenarbeiten std::chrono?

---

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

Ich habe das Gefühl, dass die "Klugheit" hier auf der falschen Seite des Gleichheitszeichens liegt. _connections.insert(_connections.end(), connection)scheint eine ausführliche Schreibweise zu sein _connections.push_back(connection). Vice versa, ich bin zu sehen , wie Menschen ersetzen verwendet map[k] = vmit map.emplace(k, v)für die Leistung und Klarheit. Denken Sie daran , dass die map[k] = versten Standard-Konstrukte map[k] , und dann ordnet einen neuen Wert hinein.

Ah, ich verstehe, Sie müssen verwenden, insertweil insertein Iterator zurückgegeben wird und push_backnicht.

Aber das wirft nur die Frage auf: Warum versuchen Sie, zwei Nebenwirkungen in einer Zeile zusammenzufassen? Wenn wir zwei Zeilen dürfen, machen wir es einfach push_backund setzen es dann map.emplace(connection.get(), std::prev(_connections.end())). Oder zum Teufel, an diesem Punkt würde ich mich nicht wirklich beschweren

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

Nachdem Sie die rote Fahne entdeckt haben, graben Sie tiefer: Was ist der Unterschied zwischen dem Einzeiler und dem klareren Zweiliner? Aha! Der Unterschied ist, was passiert, wenn _connections.insert(...)der Speicher knapp wird und geworfen wird. Mit dem Zweiliner _connectionIteratorsbleibt unberührt. Mit dem Einzeiler konstruieren Sie zunächst standardmäßig gefährlichen Müll ein _connectionIterators[connection.get()]und verbreiten dann die Ausnahme.

Ich denke, es gibt ein vernünftiges Argument für den Zweiliner, nur nach allgemeinen Grundsätzen.

---

Auch diese Antwort spricht Ihr Hauptanliegen in Bezug auf RAII nicht wirklich an, aber ich hoffe, sie gibt trotzdem Anlass zum Nachdenken.