Сервер Boost.Asio и RAII

Question

Я пытаюсь реализовать приложение сетевого сервера на C ++ с помощью Boost.Asio.

Вот требования, которым я пытаюсь соответствовать:

• Приложение создает только один экземпляр boost::io_context.
• Сингл io_contextнаходится run()в общем пуле потоков. Количество потоков не определено.
• Приложение может создавать экземпляры нескольких объектов Server. Новые серверы можно создавать и убивать в любое время.
• Каждый Сервер может обрабатывать соединения от нескольких клиентов.

Я пытаюсь реализовать шаблон RAII для класса Server. Я хочу гарантировать, что при освобождении сервера все его соединения полностью закрываются. Каждое соединение можно закрыть тремя способами:

1. Клиент отвечает, и работы с подключением больше не нужно.
2. Сервер освобождается, и все активные соединения закрываются.
3. Соединение разрывается вручную путем вызова stop()метода.

Я пришел к решению, которое, кажется, соответствует всем вышеперечисленным критериям, но, поскольку Boost.Asio все еще является для меня новым, я хотел убедиться, что то, что я делаю, правильно. Также есть пара вещей, в которых я не был уверен на 100%:

• Я пытался удалить mutexиз класса Server и вместо этого использовать strandдля всей синхронизации, но не смог найти четкого способа сделать это.
• Поскольку пул потоков может состоять только из 1 потока, и этот поток может быть тем, что вызывает деструктор сервера, который мне пришлось вызывать io_context::poll_one()из деструктора, чтобы дать возможность всем ожидающим соединениям завершить завершение работы и предотвратить потенциальную взаимоблокировку.
• Я был бы рад любым другим предложениям по улучшению, о которых вы могли бы подумать.

В любом случае, вот код с некоторыми модульными тестами (живая версия на 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();
}

С уважением, Марек

Answer 1

Я недостаточно знаю Asio, чтобы дать вам обратную связь, но вот несколько небольших исправлений, которые вы могли бы сделать:

• Не надо using namespace std. Вам, вероятно, следует избегать и using namespaceвсего остального, просто для ясности.

• virtual ~ConnectionDelegate() { }может быть virtual ~ConnectionDelegate() = default;вместо этого. Это немного лучше отражает ваше намерение.

• ~Server()должно быть ~Server() override, чтобы указать, что он переопределяет виртуальную функцию-член. В общем, вы должны использовать overrideвезде, где это физически разрешено языком. (Я думаю, вы делаете это правильно везде, кроме деструкторов.)

• Connection(string name,и Server(string name,оба без надобности делают копию string name.

• Все ваши конструкторы должны explicitсообщать компилятору, что, например, пара в фигурных скобках {"hello world", myIOContext}не должна неявно обрабатываться как (или неявно преобразовываться в) Serverобъект, даже случайно.

• Лично я считаю , использование определений типов для ConnectionsListи ConnectionsIteratorMapбыть ненужный слой косвенности. Я бы предпочел увидеть std::list<std::weak_ptr<Connection>> _connections;прямо там в очереди. Если мне нужно имя для этого типа, я могу просто сказать decltype(_connections).

• default_delete<Connection>()(connection)это многословный способ сказать delete connection. Будьте прямыми.

• class executor executor()черт возьми, сбивает с толку. Тот факт, что вы должны были сказать, classчто должен был быть красный флаг, который либо executorне является правильным именем для класса, либо executor()не является правильным именем для этого метода. Рассмотрите возможность изменения имени метода get_executor(), например, на. Я предполагаю, что вы не можете изменить имя, class executorпотому что оно не объявлено в этом файле; он должен исходить из какого-то пространства имен Boost, которое вы usingуказали, верно? (Не используйте usingпространства имен!)

---

Вы упускаете множество возможностей избежать копирования через ссылки и / или семантику перемещения. Например, в Server::connections(), я бы написал:

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

Это позволяет избежать увеличения числа слабых ссылок путем создания weakConnectionссылки вместо копии, а затем позволяет избежать увеличения числа сильных ссылок путем использования перемещения вместо копирования внутрь push_back. Спасено четыре атомных операции! (Возможно, это не имеет значения в реальной жизни, но привет, добро пожаловать в Code Review.)

---

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

Я считаю, что bindсбивает с толку, но точно не знаю (и на самом деле надеюсь, что кто-то прокомментирует и просветит меня) - здесь ли bind нужно ? Конечно, было бы легче читать, быстрее компилировать и не медленнее во время выполнения писать

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

Это сделало бы немного яснее , что на самом деле копируется (один shared_ptrдержать в *thisживых, и один сырой указатель). На самом деле, мне интересно, нужно ли нам вообще хранить копию необработанного указателя; мы могли бы вместо этого уйти с этим?

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

Или вы ожидаете, что иногда вы попадете в тело этой лямбды с помощью, d != self->_delegateи поэтому вам понадобится дополнительный указатель?

---

Мне также интересно, можно ли было бы использовать std::chrono::secondsвместо boost::posix_time::seconds. Может ли Boost.Asio взаимодействовать с C ++ 11 в std::chronoнаши дни?

---

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

Мне кажется, что «сообразительность» здесь не по ту сторону знака равенства. _connections.insert(_connections.end(), connection)кажется многословным способом написания _connections.push_back(connection). И наоборот, я привык видеть, как люди заменяются map[k] = vна map.emplace(k, v)для повышения производительности и ясности. Помните, что map[k] = vсначала конструкции по умолчанию map[k] , а затем присваивают ей новое значение.

А, понятно, вам нужно использовать, insertпотому что insertвозвращает итератор, а push_backне возвращает .

Но тут возникает вопрос: почему вы пытаетесь втиснуть два побочных эффекта в одну строку? Если нам разрешено две строки, мы просто делаем, push_backа затем устанавливаем map.emplace(connection.get(), std::prev(_connections.end())). Или, черт возьми, в тот момент я бы не стал жаловаться

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

Заметив красный флаг, копните глубже: в чем разница между однострочным и более четким двухстрочным? Ага! Разница в том, что происходит, если _connections.insert(...)заканчивается память и выкидывает. С двухслойным покрытием _connectionIteratorsостается нетронутым. В однострочном режиме вы сначала создаете по умолчанию опасный мусор, _connectionIterators[connection.get()]а затем распространяете исключение.

Так что я думаю, что есть разумный аргумент в пользу двухстрочного решения, просто исходя из общих принципов.

---

Опять же, этот ответ на самом деле не касается вашей основной проблемы, связанной с RAII, но я надеюсь, что он все равно дает пищу для размышлений.