Boost.Asio Server và RAII

Question

Tôi đang cố gắng triển khai ứng dụng máy chủ mạng bằng C ++ bằng cách sử dụng Boost.Asio.

Đây là những yêu cầu tôi đang cố gắng đáp ứng:

• Ứng dụng chỉ tạo một phiên bản của boost::io_context.
• Single io_contextđang được run()chia sẻ Thread Pool. Số lượng chủ đề không được xác định.
• Ứng dụng có thể khởi tạo nhiều đối tượng Server. Máy chủ mới có thể được sinh ra và bị giết bất cứ lúc nào.
• Mỗi Máy chủ có thể xử lý các kết nối từ nhiều máy khách.

Tôi đang cố gắng triển khai mẫu RAII cho lớp Máy chủ. Điều tôi muốn đảm bảo là khi Server được phân bổ, tất cả các kết nối của nó sẽ hoàn toàn đóng. Có 3 cách để đóng mỗi kết nối:

1. Khách hàng phản hồi và không có thêm công việc phải thực hiện trong một kết nối.
2. Máy chủ đang được phân bổ và khiến tất cả các kết nối còn sống đóng lại.
3. Kết nối bị ngắt thủ công bằng stop()phương pháp gọi .

Tôi đã đi đến một giải pháp có vẻ đáp ứng tất cả các tiêu chí trên nhưng vì Boost.Asio vẫn còn khá mới đối với tôi nên tôi muốn xác minh rằng những gì tôi đang làm là chính xác. Ngoài ra, có một số điều mà tôi không chắc chắn 100%:

• Tôi đã cố gắng xóa mutexkhỏi lớp Máy chủ và thay vào đó sử dụng một strandcho tất cả quá trình đồng bộ hóa nhưng tôi không thể tìm thấy cách rõ ràng để thực hiện.
• Bởi vì Thread Pool có thể chỉ bao gồm 1 luồng và luồng này có thể là thứ đang gọi một Trình hủy máy chủ mà tôi đã phải gọi io_context::poll_one()từ Trình hủy để tạo cơ hội cho tất cả các kết nối đang chờ hoàn thành việc tắt và ngăn chặn một bế tắc tiềm ẩn.
• Tôi hoan nghênh bất kỳ đề xuất cải tiến nào khác mà bạn có thể nghĩ ra.

Dù sao, đây là mã với một số bài kiểm tra đơn vị (phiên bản trực tiếp trên 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();
}

Trân trọng, Marek

Answer 1

Tôi không biết đủ Asio để đưa ra loại phản hồi mà tôi biết bạn muốn, nhưng đây là một số thao tác dọn dẹp nhỏ bạn có thể làm:

• Đừng using namespace std. Bạn có lẽ cũng nên tránh using namespacebất cứ điều gì khác, chỉ cho rõ ràng.

• virtual ~ConnectionDelegate() { }có thể được virtual ~ConnectionDelegate() = default;thay thế. Điều này thể hiện ý định của bạn tốt hơn một chút.

• ~Server()nên được ~Server() override, để chỉ ra rằng nó ghi đè một chức năng thành viên ảo. Nói chung, bạn nên sử dụng overridebất cứ nơi nào nó được ngôn ngữ cho phép. (Tôi nghĩ rằng bạn làm điều đó đúng ở mọi nơi ngoại trừ trên các trình hủy.)

• Connection(string name,và Server(string name,cả hai đều tạo một bản sao của string name.

• Tất cả các hàm tạo của bạn nên explicitnói với trình biên dịch rằng ví dụ như cặp giằng {"hello world", myIOContext}không nên được coi là (hoặc được chuyển đổi ngầm thành) một Serverđối tượng, thậm chí không phải do ngẫu nhiên.

• Cá nhân, tôi thấy việc sử dụng typedefs cho ConnectionsListvà ConnectionsIteratorMaplà một lớp không cần thiết về mình. Tôi muốn xem std::list<std::weak_ptr<Connection>> _connections;ngay tại đó trong dòng. Nếu tôi cần một cái tên cho loại đó, tôi có thể nói decltype(_connections).

• default_delete<Connection>()(connection)là một cách nói dài dòng delete connection. Hãy trực tiếp.

• class executor executor()hella khó hiểu. Thực tế là bạn phải nói rằng classcó một lá cờ đỏ executorkhông phải là tên phù hợp cho lớp hoặc executor()không phải là tên phù hợp cho phương thức này. get_executor()Ví dụ, hãy xem xét việc thay đổi tên của phương thức thành . Tôi giả sử rằng bạn không thể thay đổi tên của class executorvì nó không được khai báo trong tệp này; nó phải đến từ một không gian tên Boost nào đó mà bạn usingđã chỉnh sửa, phải không? (Đừng có không usinggian tên!)

---

Bạn bỏ qua rất nhiều cơ hội để tránh các bản sao qua tài liệu tham khảo và / hoặc di chuyển ngữ nghĩa. Ví dụ, trong Server::connections(), tôi sẽ viết:

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

Điều này tránh va chạm với số tiền hoàn lại yếu bằng cách tạo weakConnectiontham chiếu thay vì bản sao và sau đó tránh va chạm với số tiền hoàn lại mạnh bằng cách di chuyển thay vì sao chép vào push_back. Bốn hoạt động nguyên tử đã được lưu! (Có lẽ điều này không quan trọng trong cuộc sống thực, nhưng chào mừng bạn đến với Đánh giá mã.)

---

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

Tôi thấy việc sử dụng bindkhó hiểu, nhưng tôi không biết chắc chắn (và thực sự hy vọng ai đó sẽ nhận xét và khai sáng cho tôi) - có bind cần thiết ở đây không? Nó chắc chắn sẽ rõ ràng hơn để đọc, biên dịch nhanh hơn và không chậm hơn trong thời gian chạy để viết

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

Điều này sẽ làm cho nó rõ ràng hơn một chút những gì đang thực sự được sao chép (một con shared_ptrgiữ *thisnguyên và một con trỏ thô). Trên thực tế, tôi tự hỏi liệu chúng ta có cần lưu trữ bản sao của con trỏ thô hay không; thay vào đó chúng ta có thể làm được điều này không?

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

Hoặc bạn có mong đợi rằng đôi khi bạn sẽ đi vào phần thân của lambda d != self->_delegateđó và đó là lý do tại sao bạn cần thêm con trỏ?

---

Tôi cũng tự hỏi liệu nó có thể được sử dụng std::chrono::secondsthay thế không boost::posix_time::seconds. Những ngày này, Boost.Asio có thể tương tác với C ++ 11 std::chronokhông?

---

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

Tôi có cảm giác như "sự thông minh" ở đây là ở phía sai của dấu bằng. _connections.insert(_connections.end(), connection)có vẻ giống như một cách viết dài dòng _connections.push_back(connection). Ngược lại, tôi đã quen với việc mọi người thay thế map[k] = vbằng map.emplace(k, v)cho hiệu suất và sự rõ ràng. Hãy nhớ rằng cấu trúc mặc địnhmap[k] = v đầu tiên , và sau đó gán một giá trị mới cho nó. map[k]

À, tôi hiểu rồi, bạn cần sử dụng insertvì inserttrả về một trình lặp và push_backkhông.

Nhưng điều đó chỉ đặt ra câu hỏi: Tại sao bạn lại cố gắng ghép hai tác dụng phụ vào một dòng? Nếu chúng tôi cho phép hai dòng, chúng tôi chỉ làm push_backvà sau đó thiết lập map.emplace(connection.get(), std::prev(_connections.end())). Hoặc, quái, tại thời điểm đó tôi sẽ không thực sự phàn nàn về

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

Sau khi phát hiện ra lá cờ đỏ, hãy tìm hiểu sâu hơn: sự khác biệt giữa một lớp lót và hai lớp lót rõ ràng là gì? Aha! Sự khác biệt là những gì sẽ xảy ra nếu _connections.insert(...)hết bộ nhớ và ném. Với hai lớp lót, _connectionIteratorsvẫn còn nguyên. Với one-liner, trước tiên bạn mặc định tạo một số loại rác nguy hiểm _connectionIterators[connection.get()]và sau đó tuyên truyền ngoại lệ.

Vì vậy, tôi nghĩ rằng có một lập luận hợp lý được đưa ra để ủng hộ hai lớp lót, chỉ dựa trên các nguyên tắc chung.

---

Một lần nữa, câu trả lời này không thực sự giải quyết mối quan tâm chính của bạn về RAII, nhưng tôi hy vọng nó sẽ cung cấp một số thực phẩm để suy nghĩ.