Boost Asio Server และ RAII

Question

ฉันกำลังพยายามติดตั้งแอปพลิเคชันเซิร์ฟเวอร์เครือข่ายใน C ++ โดยใช้ Boost Asio

นี่คือข้อกำหนดที่ฉันพยายามปฏิบัติตาม:

• แอปพลิเคชันสร้างเพียงอินสแตนซ์เดียวของboost::io_context.
• โสดio_contextอยู่run()โดยเธรดพูลที่ใช้ร่วมกัน ไม่ได้กำหนดจำนวนเธรด
• แอปพลิเคชันสามารถสร้างอินสแตนซ์อ็อบเจ็กต์เซิร์ฟเวอร์หลายตัว เซิร์ฟเวอร์ใหม่สามารถสร้างและฆ่าได้ตลอดเวลา
• เซิร์ฟเวอร์แต่ละเครื่องสามารถจัดการการเชื่อมต่อจากไคลเอนต์หลาย ๆ

ฉันกำลังพยายามใช้รูปแบบ RAII สำหรับคลาสเซิร์ฟเวอร์ สิ่งที่ฉันต้องการรับประกันคือเมื่อเซิร์ฟเวอร์ได้รับการยกเลิกการจัดสรรการเชื่อมต่อทั้งหมดจะถูกปิดอย่างสมบูรณ์ สามารถปิดการเชื่อมต่อแต่ละรายการได้ 3 วิธี:

1. ลูกค้าตอบสนองและไม่มีงานที่ต้องทำในการเชื่อมต่ออีกต่อไป
2. เซิร์ฟเวอร์กำลังถูกยกเลิกการจัดสรรและทำให้การเชื่อมต่อทั้งหมดปิดลง
3. การเชื่อมต่อถูกฆ่าด้วยตนเองโดยการเรียกใช้stop()เมธอด

ฉันมาถึงวิธีแก้ปัญหาที่ดูเหมือนจะตรงตามเกณฑ์ทั้งหมดข้างต้น แต่เนื่องจาก Boost Asio ยังค่อนข้างใหม่สำหรับฉันฉันจึงต้องการตรวจสอบว่าสิ่งที่ฉันทำนั้นถูกต้อง นอกจากนี้ยังมีบางสิ่งที่ฉันไม่แน่ใจ 100% โดยเฉพาะ:

• ฉันพยายามลบออกmutexจากคลาสเซิร์ฟเวอร์และใช้ a strandสำหรับการซิงโครไนซ์ทั้งหมดแทน แต่ฉันไม่พบวิธีที่ชัดเจนในการทำ
• เนื่องจาก Thread Pool ประกอบด้วย 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();
}

ขอแสดงความนับถือ Marek

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วัตถุโดยปริยาย) ไม่ใช่แม้แต่โดยบังเอิญ

• โดยส่วนตัวแล้วฉันพบว่ามีการใช้ typedefs สำหรับ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;
}

นี้หลีกเลี่ยงการชน refcount อ่อนแอโดยการweakConnectionอ้างอิงแทนการคัดลอกแล้วหลีกเลี่ยงการกระแทก refcount 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 แต่ฉันหวังว่ามันจะให้อาหารสำหรับความคิดต่อไป