c++ - 使用 async_accept 处理多个客户端

Question

我正在使用 boost ASIO 和协程编写一个安全的 SSL 回显服务器。我希望此服务器能够为多个并发客户端提供服务，这是我的代码

 try {
    boost::asio::io_service io_service;

    boost::asio::spawn(io_service, [&io_service](boost::asio::yield_context yield) {
      auto ctx = boost::asio::ssl::context{ boost::asio::ssl::context::sslv23 };
      ctx.set_options(
        boost::asio::ssl::context::default_workarounds
        | boost::asio::ssl::context::no_sslv2
        | boost::asio::ssl::context::single_dh_use);
      ctx.use_private_key_file(..); // My data setup
      ctx.use_certificate_chain_file(...); // My data setup

      boost::asio::ip::tcp::acceptor acceptor(io_service,
        boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port));

      for (;;) {

        boost::asio::ssl::stream<boost::asio::ip::tcp::socket> sock{ io_service, ctx };
        acceptor.async_accept(sock.next_layer(), yield);

        sock.async_handshake(boost::asio::ssl::stream_base::server, yield);

        auto ec = boost::system::error_code{};
        char data_[1024];
        auto nread = sock.async_read_some(boost::asio::buffer(data_, 1024), yield[ec]);

        if (ec == boost::asio::error::eof)
          return; //connection closed cleanly by peer
        else if (ec)
          throw boost::system::system_error(ec); //some other error, is this desirable?

        sock.async_write_some(boost::asio::buffer(data_, nread), yield[ec]);

        if (ec == boost::asio::error::eof)
          return; //connection closed cleanly by peer
        else if (ec)
          throw boost::system::system_error(ec); //some other error

        // Shutdown gracefully
        sock.async_shutdown(yield[ec]);
        if (ec && (ec.category() == boost::asio::error::get_ssl_category())
          && (SSL_R_PROTOCOL_IS_SHUTDOWN == ERR_GET_REASON(ec.value())))
        {
          sock.lowest_layer().close();
        }
      }

    });

    io_service.run();
  }
  catch (std::exception& e)
  {
    std::cerr << "Exception: " << e.what() << "\n";
  }

无论如何，我不确定上面的代码是否可行:理论上调用 async_accept 会将控制权返回给 io_service 管理器。

如果一个连接已经被接受，是否会接受另一个连接，即它已经超过了 async_accept 行？

Answer 1

理解您的问题的具体细节有点困难，因为代码不完整(例如，您的 block 中有一个 return，但不清楚该 block 的部分是什么)。

尽管如此，该文档包含一个 example of a TCP echo server using coroutines .看来您基本上需要为其添加 SSL 支持，以使其适应您的需求。

如果您查看 main，它有以下 block :

boost::asio::spawn(io_service,
    [&](boost::asio::yield_context yield)
    {
      tcp::acceptor acceptor(io_service,
        tcp::endpoint(tcp::v4(), std::atoi(argv[1])));

      for (;;)
      {
        boost::system::error_code ec;
        tcp::socket socket(io_service);
        acceptor.async_accept(socket, yield[ec]);
        if (!ec) std::make_shared<session>(std::move(socket))->go();
      }
    });

这个无限循环，并且在每次(成功)调用 async_accept 之后，处理接受下一个连接(同时这个连接和其他连接可能仍然处于事件状态)。

同样，我不确定您的代码，但它包含像这样的循环退出

return; //connection closed cleanly by peer

---

为了说明这一点，这里有两个应用程序。

第一个是 Python 多处理回显客户端，改编自 PMOTW :

import socket
import sys
import multiprocessing

def session(i):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    server_address = ('localhost', 5000)
    print 'connecting to %s port %s' % server_address
    sock.connect(server_address)
    print 'connected'

    for _ in range(300):
        try:

            # Send data
            message = 'client ' + str(i) + ' message'
            print 'sending "%s"' % message
            sock.sendall(message)

            # Look for the response
            amount_received = 0
            amount_expected = len(message)

            while amount_received < amount_expected:
                data = sock.recv(16)
                amount_received += len(data)
                print 'received "%s"' % data

        except:
            print >>sys.stderr, 'closing socket'
            sock.close()

if __name__ == '__main__':
    pool = multiprocessing.Pool(8)
    pool.map(session, range(8))

细节不是那么重要(虽然它是 Python，因此很容易阅读)，但重点是它打开了 8 个进程，并且每个进程都使用相同的 asio echo 服务器(下图)和 300 条消息。

运行时输出

...
received "client 1 message"
sending "client 1 message"
received "client 2 message"
sending "client 2 message"
received "client 3 message"
received "client 0 message"
sending "client 3 message"
sending "client 0 message"
...

显示回声 session 确实是交错的。

现在是回显服务器。我稍微调整了 example from the docs :

#include <cstdlib>
#include <iostream>
#include <memory>
#include <utility>
#include <boost/asio.hpp>

using boost::asio::ip::tcp;

class session :
    public std::enable_shared_from_this<session> {

public:
    session(tcp::socket socket) : socket_(std::move(socket)) {}

    void start() { do_read(); }

private:
    void do_read() {
        auto self(
            shared_from_this());
        socket_.async_read_some(
            boost::asio::buffer(data_, max_length),
            [this, self](boost::system::error_code ec, std::size_t length) {
                 if(!ec)
                     do_write(length);
            });
    }

    void do_write(std::size_t length) {
        auto self(shared_from_this());
        socket_.async_write_some(
            boost::asio::buffer(data_, length),
            [this, self](boost::system::error_code ec, std::size_t /*length*/) {
                if (!ec)
                    do_read();
            });
    }

private:
    tcp::socket socket_;
    enum { max_length = 1024 };
    char data_[max_length];
};

class server {
public:
    server(boost::asio::io_service& io_service, short port) :
            acceptor_(io_service, tcp::endpoint(tcp::v4(), port)),
            socket_(io_service) {
        do_accept();
    }

private:
    void do_accept() {
        acceptor_.async_accept(
            socket_,
            [this](boost::system::error_code ec) {
                if(!ec)
                    std::make_shared<session>(std::move(socket_))->start();

                do_accept();
            });
    }

    tcp::acceptor acceptor_;
    tcp::socket socket_;
};

int main(int argc, char* argv[]) {
    const int port = 5000;
    try {
        boost::asio::io_service io_service;

        server s{io_service, port};

        io_service.run();
    }
    catch (std::exception& e) {
        std::cerr << "Exception: " << e.what() << "\n";
    }
}

这表明这个服务器确实交错了。

请注意，这不是协程版本。虽然我曾经玩过协程版本，但我无法在我当前的盒子上构建它(另外，正如 sehe 在下面的评论中指出的那样，您现在可能更喜欢这个更主流的版本)。

然而，这不是根本区别，w.r.t.你的问题。非协程版本有回调显式显式启动新操作提供下一个回调；协程版本使用了一种更符合顺序的范例。在两个版本中，每次调用都会返回到 asio 的控制循环，它监视所有 当前可以继续的操作。

来自asio coroutine docs :

Coroutines let you create a structure that mirrors the actual program logic. Asynchronous operations don’t split functions, because there are no handlers to define what should happen when an asynchronous operation completes. Instead of having handlers call each other, the program can use a sequential structure.

并不是说顺序结构使所有操作都顺序进行 - 这将消除对 asio 的全部需求。