ASIO - 如何停止简单的基于协程的服务器？

Question

我有以下简单的基于协程的服务器：

class Server
{
private:
  boost::asio::io_service        Service;
  boost::asio::ip::tcp::acceptor Acceptor;
  boost::asio::ip::tcp::socket   Socket;

private:
  void Accept(boost::asio::yield_context Yield);
  void Write(boost::asio::yield_context Yield);

public:
       Server(): Acceptor(Service), Socket(Service) {}
  void Open(unsigned short PortNum);
  void Run();
  void Stop();
};

void Server::Accept(boost::asio::yield_context Yield)
{
boost::system::error_code ec;

for (;;)
  {
  Socket.close();
  Acceptor.async_accept(Socket,Yield[ec]);
  spawn(Yield,std::bind(&Server::Write,this,Yield[ec]));
  }
}

void Server::Write(boost::asio::yield_context Yield)
{
char InBuffer[1024]= {};
std::size_t Size;
boost::system::error_code ec;
double Data= 6.66;

for (;;)
  {
  boost::asio::streambuf OutBuffer;
  std::ostream os(&OutBuffer);

  Size= Socket.async_read_some(boost::asio::buffer(InBuffer),Yield[ec]);
  if (ec)
    break;
  os.write(reinterpret_cast<const char *>(&Data),sizeof(double));
  Socket.async_write_some(OutBuffer.data(),Yield[ec]);
  if (ec)
    break;
  }
}

void Server::Open(unsigned short PortNum)
{
Acceptor.open(boost::asio::ip::tcp::v4());
Acceptor.bind({{},PortNum});
Acceptor.listen();
}

void Server::Run()
{
spawn(Service,std::bind(&Server::Accept,this,std::placeholders::_1));
Service.run();
}

void Server::Stop()
{
Service.stop();
}

我想运行这个服务器在一个线程上并在主程序即将完成时干净地停止它:

int main()
{
Server s;

s.Open(1024);

std::thread Thread(&Server::Run,&s);

Sleep(10'000);
s.Stop();
Thread.join();
}

不幸的是，如果有连接的套接字，当我调用 Stop 时抛出异常 boost::coroutines::detail::forced_unwind。

我也曾尝试创建一个显式 strand 并在停止前分派 Socket.close()，结果相同。

这种方法有问题吗？

Answer 1

I’m having trouble trying to stop gracefully a similar server ( whosebug.com/questions/50833730/…). – metalfox 4 hours ago

这是一个显示如何处理

的最小更改

关闭会话的退出命令
关闭服务器的关闭命令（因此它停止接受连接并在最后一个会话退出后终止）

Live On Coliru

#include <boost/asio.hpp>
#include <iostream>

using boost::asio::ip::tcp;
using boost::system::error_code;
using boost::asio::streambuf;

int main() {
    boost::asio::io_service svc;

    tcp::acceptor a(svc);
    a.open(tcp::v4());
    a.set_option(tcp::acceptor::reuse_address(true));
    a.bind({{}, 6767}); // bind to port 6767 on localhost
    a.listen(5);

    using session = std::shared_ptr<tcp::socket>;

    std::function<void()>        do_accept;
    std::function<void(session)> do_session;

    do_session = [&](session s) {
        // do a read
        auto buf = std::make_shared<boost::asio::streambuf>();
        async_read_until(*s, *buf, "\n", [&,s,buf](error_code ec, size_t /*bytes*/) {
            if (ec)
                std::cerr << "read failed: " << ec.message() << "\n";
            else {
                std::istream is(buf.get());
                std::string line;
                while (getline(is, line)) // FIXME being sloppy with partially read lines
                {
                    async_write(*s, boost::asio::buffer("Ack\n", 4), [&,s,buf](error_code ec, size_t) {
                        if (ec) std::cerr << "write failed: " << ec.message() << "\n";
                    });
                    if (line == "Exit") {
                        std::cout << "Exit received\n";
                        return;
                    }
                    if (line == "Shutdown") {
                        std::cout << "Server shutdown requested\n";
                        a.close();
                        return;
                    }
                }

                do_session(s); // full duplex, can read while writing, using a second buffer
            }

        });
    };

    do_accept = [&] {
        auto s = std::make_shared<session::element_type>(svc);

        a.async_accept(*s, [&,s](error_code ec) {
            if (ec)
                std::cerr << "accept failed: " << ec.message() << "\n";
            else {
                do_session(s);
                do_accept(); // accept the next
            }
        });
    };

    do_accept(); // kick-off
    svc.run();   // wait for shutdown (Ctrl-C or failure)
}

注意示例会话

echo -en "hello world\nExit\n"     | netcat 127.0.0.1 6767
echo -en "hello world\nShutdown\n" | netcat 127.0.0.1 6767

打印

Ack
Ack
Ack
Ack
Exit received
Server shutdown requested
accept failed: Operation canceled

终止命令

如果您想要 "Terminate" 命令主动关闭所有打开的会话并且关闭服务器，您将必须

保留会话列表
或使用信号

您可以在此处查看两种方法的代码：

与当前示例集成的最简单方法：

Live On Coliru

#include <boost/asio.hpp>
#include <iostream>
#include <list>

using boost::asio::ip::tcp;
using boost::system::error_code;
using boost::asio::streambuf;

int main() {
    boost::asio::io_service svc;

    tcp::acceptor a(svc);
    a.open(tcp::v4());
    a.set_option(tcp::acceptor::reuse_address(true));
    a.bind({{}, 6767}); // bind to port 6767 on localhost
    a.listen(5);

    using session = std::shared_ptr<tcp::socket>;
    using sessref = std::weak_ptr<tcp::socket>;

    std::function<void()>        do_accept;
    std::function<void(session)> do_session;
    std::list<sessref> session_list;

    auto garbage_collect_sessions = [&session_list] {
        session_list.remove_if(std::mem_fn(&sessref::expired));
    };

    do_session = [&](session s) {
        // do a read
        auto buf = std::make_shared<boost::asio::streambuf>();
        async_read_until(*s, *buf, "\n", [&,s,buf](error_code ec, size_t /*bytes*/) {
            if (ec)
                std::cerr << "read failed: " << ec.message() << "\n";
            else {
                std::istream is(buf.get());
                std::string line;
                while (getline(is, line)) // FIXME being sloppy with partially read lines
                {
                    async_write(*s, boost::asio::buffer("Ack\n", 4), [&,s,buf](error_code ec, size_t) {
                        if (ec) std::cerr << "write failed: " << ec.message() << "\n";
                    });
                    if (line == "Exit") {
                        std::cout << "Exit received\n";
                        return;
                    }
                    if (line == "Shutdown") {
                        std::cout << "Server shutdown requested\n";
                        a.close();
                        return;
                    }
                    if (line == "Terminate") {
                        std::cout << "Server termination requested\n";
                        a.close();
                        for (auto wp : session_list) {
                            if (auto session = wp.lock())
                                session->close();
                        }
                        return;
                    }
                }

                do_session(s); // full duplex, can read while writing, using a second buffer
            }

        });
    };

    do_accept = [&] {
        auto s = std::make_shared<session::element_type>(svc);

        a.async_accept(*s, [&,s](error_code ec) {
            if (ec)
                std::cerr << "accept failed: " << ec.message() << "\n";
            else {
                garbage_collect_sessions();

                session_list.push_back(s);
                do_session(s);
                do_accept(); // accept the next
            }
        });
    };

    do_accept(); // kick-off
    svc.run();   // wait for shutdown (Ctrl-C or failure)
}

这显然使用 session_list 来实现 "Terminate" 命令：

if (line == "Terminate") {
    std::cout << "Server termination requested\n";
    a.close();
    for (auto wp : session_list) {
        if (auto session = wp.lock())
            session->close();
    }
    return;
}

ASIO - 如何停止简单的基于协程的服务器？

ASIO - How to stop simple coroutine based server?

c++

boost-asio

boost-coroutine

终止命令