boost::asio::async_write 写入 ssl::stream 成功但服务器未获取

Question

我写了一个ssl服务器，客户端运行一个pingpang进程，过了一会儿，客户端说发送数据成功，但服务器没有收到。 client 运行多线程，单线程时，正常。我尝试添加计时器以添加握手，然后服务器可以获取所有数据，但我希望它可以运行正确地握手任何人都可以帮助找出问题所在。

这是我的服务器

#include <cstdlib>
#include <functional>
#include <iostream>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/bind/bind.hpp>

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

class session : public std::enable_shared_from_this<session> {
public:
    session(tcp::socket socket, boost::asio::ssl::context &context)
        : socket_(std::move(socket), context), m_strand(socket.get_executor()) {
    }

    void start() {
        do_handshake();
    }

private:
    void do_handshake() {
        auto self(shared_from_this());
        socket_.async_handshake(boost::asio::ssl::stream_base::server,
                                [this, self](const boost::system::error_code &error) {
                                    if (!error) {
                                        do_read();
                                    }
                                });
    }

    void do_read() {
        auto self(shared_from_this());

        socket_.async_read_some(boost::asio::buffer(data_),
                                [this, self](const boost::system::error_code &ec, std::size_t length) {
                                    if (!ec) {
                                        std::cout << "get <";
                                        std::cout.write(data_, length);
                                        std::cout << std::endl;
                                        do_write(length);
                                    }
                                });
    }

    void do_write(std::size_t length) {
        auto self(shared_from_this());
        std::cout << "send <";
        std::cout.write(data_, length);
        std::cout << std::endl;
        boost::asio::async_write(socket_, boost::asio::buffer(data_, length),
                                 [this, self](const boost::system::error_code &ec,
                                              std::size_t /*length*/) {
                                     if (!ec) {
                                         do_read();
                                     }
                                 });
    }

    boost::asio::ssl::stream<tcp::socket> socket_;
    boost::asio::strand<boost::asio::ip::tcp::socket::executor_type> m_strand;
    char data_[1024];
};

class server {
public:
    server(boost::asio::io_context &io_context, unsigned short port)
        : acceptor_(io_context, tcp::endpoint(tcp::v4(), port)),
          context_(boost::asio::ssl::context::sslv23) {
        context_.set_options(
            boost::asio::ssl::context::default_workarounds
            | boost::asio::ssl::context::no_sslv2
            | boost::asio::ssl::context::single_dh_use);
        context_.set_password_callback(std::bind(&server::get_password, this));
        context_.use_certificate_chain_file("server.pem");
        context_.use_private_key_file("server.pem", boost::asio::ssl::context::pem);
        context_.use_tmp_dh_file("dh2048.pem");

        do_accept();
    }

private:
    std::string get_password() const {
        return "test";
    }

    void do_accept() {
        acceptor_.async_accept(
            [this](const boost::system::error_code &error, tcp::socket socket) {
                if (!error) {
                    std::make_shared<session>(std::move(socket), context_)->start();
                }

                do_accept();
            });
    }

    tcp::acceptor acceptor_;
    boost::asio::ssl::context context_;
};

int main(int argc, char *argv[]) {
    try {
        if (argc != 2) {
            std::cerr << "Usage: server <port>\n";
            return 1;
        }

        boost::asio::io_context io_context;

        using namespace std; // For atoi.
        server s(io_context, atoi(argv[1]));

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

    return 0;
}

和下一个客户

#include <cstdlib>
#include <cstring>
#include <functional>
#include <iostream>
#include <boost/asio.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/thread.hpp>

using boost::asio::ip::tcp;
using std::placeholders::_1;
using std::placeholders::_2;

enum {
    max_length = 1024
};

class client {
public:
    client(boost::asio::io_context &io_context,
           boost::asio::ssl::context &context,
           const tcp::resolver::results_type &endpoints)
        : socket_(io_context, context), strand_(io_context.get_executor()) {
        socket_.set_verify_mode(boost::asio::ssl::verify_peer);
        socket_.set_verify_callback(
            std::bind(&client::verify_certificate, this, _1, _2));

        connect(endpoints);
    }

private:
    bool verify_certificate(bool preverified,
                            boost::asio::ssl::verify_context &ctx) {
        char subject_name[256];
        X509 *cert = X509_STORE_CTX_get_current_cert(ctx.native_handle());
        X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
        std::cout << "Verifying " << subject_name << "\n";

        return true;
    }

    void connect(const tcp::resolver::results_type &endpoints) {
        boost::asio::async_connect(socket_.lowest_layer(), endpoints,
                                   [this](const boost::system::error_code &error,
                                          const tcp::endpoint & /*endpoint*/) {
                                       if (!error) {
                                           handshake();
                                       } else {
                                           std::cout << "Connect failed: " << error.message() << "\n";
                                       }
                                   });
    }

    void handshake() {
        socket_.async_handshake(boost::asio::ssl::stream_base::client,
                                [this](const boost::system::error_code &error) {
                                    if (!error) {
                                        send_request("hello ssl");
                                        boost::asio::post(strand_, std::bind(&client::recv, this));
                                    } else {
                                        std::cout << "Handshake failed: " << error.message() << "\n";
                                    }
                                });
    }

    void send_request(const std::string &msg) {
        boost::asio::async_write(
            socket_, boost::asio::buffer(msg),
            [this](const boost::system::error_code &error, std::size_t length) {
                if (!error) {
                    std::cout << "send data success, size : " << length << std::endl;
                } else {
                    std::cout << "Write failed: " << error.message() << std::endl;
                }
            });
    }

    void recv() {
        boost::asio::async_read(
            socket_, buffer_, boost::asio::transfer_exactly(9),
            boost::asio::bind_executor(
                strand_, [this](const boost::system::error_code &error, std::size_t length) {
                    if (!error) {
                        std::istream buffer(&buffer_);
                        std::vector<char> msg(length, 0);
                        buffer.readsome(msg.data(), length);
                        std::string recvMsg(msg.begin(), msg.end());
                        std::cout << recvMsg << std::endl;
                        send_request(recvMsg);
                        boost::asio::post(strand_, std::bind(&client::recv, this));
                    } else {
                        std::cout << "Read failed: " << error.message() << std::endl;
                    }
                }));
    }

    boost::asio::ssl::stream<tcp::socket> socket_;
    boost::asio::streambuf buffer_;
    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
};

int main(int argc, char *argv[]) {
    try {
        if (argc != 3) {
            std::cerr << "Usage: client <host> <port>\n";
            return 1;
        }

        boost::asio::io_context io_context;

        tcp::resolver resolver(io_context);
        auto endpoints = resolver.resolve(argv[1], argv[2]);

        boost::asio::ssl::context ctx(boost::asio::ssl::context::sslv23);
        ctx.load_verify_file("ca.pem");

        client c(io_context, ctx, endpoints);
        boost::thread_group threadPool;
        for (size_t i = 0; i < boost::thread::hardware_concurrency(); ++i) {
            threadPool.create_thread(boost::bind(&boost::asio::io_context::run, &io_context));
        }
        io_context.run();
    }
    catch (std::exception &e) {
        std::cerr << "Exception: " << e.what() << "\n";
    }

    return 0;
}

当数据没有发送到服务器时，客户端会这样打印

hello ssl
hello ssl
send data success, size : 9
send data success, size : 9

Answer 1

看看这个。如果你删除 thread_group （据我所知，它没有增加任何价值），一切正常。这是一个好兆头，表明您存在线程错误。

在看到问题之前，我没有心情阅读代码，所以让我们绕一圈。

添加 ASAN/UBSAN 立即显示没有任何问题，所以很好。

那我再看一下代码。

session 创建一个 m_strand - 从未使用过...
你忘了加入额外的线程

既然我注意到围绕 strand 的一些潜在混淆，我查看了 client strand 的使用。并且看到不一致：

插座本身不在绳子上
send_request 不会运行 on 也不会 将完成处理程序绑定到 strand 的执行程序
通信是全双工的（意思是 async_write 和 async_read 同时发生）。
这意味着 client::recv 被发布到 strand 的地方，它实际上并没有同步对 socket_ 的线程访问（因为 send_request 没有绑定到 strand第一名)

If the above is surprising, you're not the first to fall into that: . In your example connect() and handshake() can be considered safe because they form a logical strand (sequential flow of execution). The problem arises with the concurrent paths.

到目前为止，解决这种情况的最简单方法似乎是从 strand_ 构造 socket_。这意味着重新排序成员，因此首先初始化 strand_：

client(boost::asio::io_context& io_context, ssl::context& context,
       const tcp::resolver::results_type& endpoints)
    : strand_(io_context.get_executor())
    , socket_(strand_, context)

接下来，链中的所有 post 都可以删除，因为它们总是发生在该链上的完成处理程序中。

send_request("hello ssl");
recv(); // already on the strand in this completion handler

具有讽刺意味的是，send_request 是在隐含的假设下被处决的。

到目前为止清理的程序是

文件client.cpp

 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 #include <boost/thread.hpp>
 #include <cstdlib>
 #include <cstring>
 #include <functional>
 #include <iostream>

 using boost::asio::ip::tcp;
 using boost::system::error_code;
 using std::placeholders::_1;
 using std::placeholders::_2;
 namespace ssl = boost::asio::ssl;

 class client {
   public:
     client(boost::asio::io_context& io_context, ssl::context& context,
            const tcp::resolver::results_type& endpoints)
         : strand_(io_context.get_executor())
         , socket_(strand_, context)
     {
         socket_.set_verify_mode(ssl::verify_peer);
         socket_.set_verify_callback(
             std::bind(&client::verify_certificate, this, _1, _2));

         connect(endpoints);
     }

   private:
     bool verify_certificate(bool preverified, ssl::verify_context& ctx)
     {
         char  subject_name[256];
         X509* cert = X509_STORE_CTX_get_current_cert(ctx.native_handle());
         X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
         std::cout << "Verifying " << subject_name << "\n";

         return true;
     }

     void connect(const tcp::resolver::results_type& endpoints)
     {
         async_connect( //
             socket_.lowest_layer(), endpoints,
             bind_executor(
                 strand_, [this](error_code error, const tcp::endpoint&) {
                     if (!error) {
                         handshake();
                     } else {
                         std::cout << "Connect failed: " << error.message() << "\n";
                     }
                 }));
     }

     void handshake()
     {
         socket_.async_handshake(
             ssl::stream_base::client,
             bind_executor(strand_, [this](error_code error) {
                 if (!error) {
                     send_request("hello ssl");
                     recv(); // already on the strand in this completion handler
                 } else {
                     std::cout << "Handshake failed: " << error.message()
                               << "\n";
                 }
             }));
     }

     void send_request(std::string const& msg)
     {
         msg_ = msg;
         async_write(
             socket_, boost::asio::buffer(msg_),
             bind_executor(
                 strand_, [/*this*/](error_code error, std::size_t length) {
                     if (!error) {
                         std::cout << "send data success, size : " << length << std::endl;
                     } else {
                         std::cout << "Write failed: " << error.message() << std::endl;
                     }
                 }));
     }

     void recv()
     {
         async_read(
             socket_, buffer_, boost::asio::transfer_exactly(9),
             boost::asio::bind_executor(
                 strand_, [this](error_code error, std::size_t length) {
                     if (!error) {
                         std::istream      buffer(&buffer_);
                         std::vector<char> msg(length, 0);
                         buffer.readsome(msg.data(), length);

                         msg_.assign(msg.begin(), msg.end());
                         std::cout << msg_ << std::endl;

                         send_request(msg_);
                         recv(); // already on the strand in this completion handler
                     } else {
                         std::cout << "Read failed: " << error.message() << std::endl;
                     }
                 }));
     }

     boost::asio::strand<boost::asio::io_context::executor_type> strand_;
     ssl::stream<tcp::socket>                                    socket_;
     boost::asio::streambuf                                      buffer_;
     std::string                                                 msg_;
 };

 int main(int argc, char* argv[])
 {
     try {
         if (argc != 3) {
             std::cerr << "Usage: client <host> <port>\n";
             return 1;
         }

         boost::asio::io_context io_context;

         tcp::resolver resolver(io_context);
         auto          endpoints = resolver.resolve(argv[1], argv[2]);

         ssl::context ctx(ssl::context::sslv23);
         ctx.load_verify_file("ca.pem");

         client              c(io_context, ctx, endpoints);
         boost::thread_group threadPool;
         for (size_t i = 0; i < boost::thread::hardware_concurrency(); ++i) {
             threadPool.create_thread(
                 boost::bind(&boost::asio::io_context::run, &io_context));
         }
         threadPool.join_all();
         //io_context.run();

         return 0;
     } catch (std::exception const& e) {
         std::cerr << "Exception: " << e.what() << "\n";
         return 1;
     }
 }

文件server.cpp

 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 #include <boost/bind/bind.hpp>
 #include <cstdlib>
 #include <functional>
 #include <iostream>

 namespace ssl = boost::asio::ssl;
 using boost::asio::ip::tcp;
 using boost::system::error_code;

 class session : public std::enable_shared_from_this<session> {
   public:
     session(tcp::socket socket, ssl::context& context)
         : socket_(std::move(socket), context)
         , m_strand(socket.get_executor())
     {
     }

     void start()
     {
         do_handshake();
     }

   private:
     void do_handshake()
     {
         auto self(shared_from_this());
         socket_.async_handshake(ssl::stream_base::server,
                                 [this, self](error_code error) {
                                     if (!error) {
                                         do_read();
                                     }
                                 });
     }

     void do_read()
     {
         auto self(shared_from_this());

         socket_.async_read_some(
             boost::asio::buffer(data_),
             [this, self](error_code ec, std::size_t length) {
                 if (!ec) {
                     std::cout << "get <";
                     std::cout.write(data_, length);
                     std::cout << std::endl;
                     do_write(length);
                 }
             });
     }

     void do_write(std::size_t length)
     {
         auto self(shared_from_this());
         std::cout << "send <";
         std::cout.write(data_, length);
         std::cout << std::endl;
         boost::asio::async_write(
             socket_, boost::asio::buffer(data_, length),
             [this, self](error_code ec, std::size_t /*length*/) {
                 if (!ec) {
                     do_read();
                 }
             });
     }

     ssl::stream<tcp::socket>                        socket_;
     boost::asio::strand<tcp::socket::executor_type> m_strand;
     char                                            data_[1024];
 };

 class server {
   public:
     server(boost::asio::io_context& io_context, unsigned short port)
         : acceptor_(io_context, tcp::endpoint(tcp::v4(), port))
         , context_(ssl::context::sslv23)
     {
         context_.set_options(ssl::context::default_workarounds |
                              ssl::context::no_sslv2 |
                              ssl::context::single_dh_use);
         context_.set_password_callback(std::bind(&server::get_password, this));
         context_.use_certificate_chain_file("server.pem");
         context_.use_private_key_file("server.pem", ssl::context::pem);
         context_.use_tmp_dh_file("dh2048.pem");

         do_accept();
     }

   private:
     std::string get_password() const
     {
         return "test";
     }

     void do_accept()
     {
         acceptor_.async_accept([this](error_code error, tcp::socket socket) {
             if (!error) {
                 std::make_shared<session>(std::move(socket), context_)->start();
             }

             do_accept();
         });
     }

     tcp::acceptor acceptor_;
     ssl::context  context_;
 };

 int main(int argc, char* argv[])
 {
     try {
         if (argc != 2) {
             std::cerr << "Usage: server <port>\n";
             return 1;
         }

         boost::asio::io_context io_context;

         server s(io_context, std::atoi(argv[1]));

         io_context.run();
         return 0;
     } catch (std::exception const& e) {
         std::cerr << "Exception: " << e.what() << "\n";
         return 1;
     }
 }

其他问题

终身错误

UBSAN/ASAN没看出来，但是错了：

void send_request(const std::string& msg)
{
    async_write(
        socket_, boost::asio::buffer(msg),
        ...

问题是 msg 的生命周期，它在异步操作有机会运行之前就消失了，更不用说完成了。因此，移动缓冲区以使生命周期足够（例如成员msg_）。

并发写入

客户端锁死时，显示

send data success, size : 9
hello ssl
hello ssl
send data success, size : 9
send data success, size : 9

这表示在启动发送之前收到了第二个 hello ssl。这意味着启动了第二个发送。这实际上取消了 ssl 流上下文中的双工同步对象。您可以使用 -DBOOST_ASIO_ENABLE_HANDLER_TRACKING:

来查看

@asio|1630155694.209267|51139|deadline_timer@0x7ffc6fa61e48.cancel

使用 handlerviz.pl 脚本可视化：

问题违反了the requirements here:

The program must ensure that the stream performs no other write operations (such as async_write, the stream's async_write_some function, or any other composed operations that perform writes) until this operation completes.

两种简单的修复方法：

将 IO 从全双工更改为顺序 read/write/read/write 就像服务器一样
制作一个包含仍要按顺序写入的消息的输出队列

固定解

这使用发件箱作为上面重叠写入的第二种解决方案。我也冒昧

删除不必要的中间缓冲区streambuf buffer_而不是直接读入字符串。
将io_context + thread_group替换为更优雅的thread_pool
许多小改进（上面提到的一些）

文件client.cpp

 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 #include <boost/thread.hpp>
 #include <cstdlib>
 #include <cstring>
 #include <functional>
 #include <iostream>

 using boost::asio::ip::tcp;
 using boost::system::error_code;
 using std::placeholders::_1;
 using std::placeholders::_2;
 namespace ssl = boost::asio::ssl;

 using Executor = boost::asio::thread_pool::executor_type;

 class client {
   public:
     client(Executor ex, ssl::context& context,
            const tcp::resolver::results_type& endpoints)
         : strand_(ex)
         , socket_(strand_, context)
     {
         socket_.set_verify_mode(ssl::verify_peer);
         socket_.set_verify_callback(
             std::bind(&client::verify_certificate, this, _1, _2));

         connect(endpoints);
     }

   private:
     bool verify_certificate(bool preverified, ssl::verify_context& ctx)
     {
         char  subject_name[256];
         X509* cert = X509_STORE_CTX_get_current_cert(ctx.native_handle());
         X509_NAME_oneline(X509_get_subject_name(cert), subject_name, 256);
         std::cout << "Verifying " << subject_name << "\n";

         return true;
     }

     void connect(const tcp::resolver::results_type& endpoints)
     {
         async_connect( //
             socket_.lowest_layer(), endpoints,
             bind_executor(
                 strand_, [this](error_code error, const tcp::endpoint&) {
                     if (!error) {
                         handshake();
                     } else {
                         std::cout << "Connect failed: " << error.message() << "\n";
                     }
                 }));
     }

     void handshake()
     {
         socket_.async_handshake(
             ssl::stream_base::client,
             bind_executor(strand_, [this](error_code error) {
                 if (!error) {
                     send_request("hello ssl");
                     recv(); // already on the strand in this completion handler
                 } else {
                     std::cout << "Handshake failed: " << error.message()
                               << "\n";
                 }
             }));
     }

     void send_request(std::string msg)
     {
         outbox_.push_back(std::move(msg));
         if (outbox_.size() == 1)
         {
             send_loop();
         }
     }

     void send_loop()
     {
         async_write( //
             socket_, boost::asio::buffer(outbox_.back()),
             bind_executor(
                 strand_, [this](error_code error, std::size_t length) {
                     if (!error) {
                         std::cout << "send data success, size : " << length << std::endl;
                         outbox_.pop_back();
                     } else {
                         std::cout << "Write failed: " << error.message() << std::endl;
                     }
                     if (!outbox_.empty())
                         send_loop();
                 }));
     }

     void recv()
     {
         async_read(
             socket_, boost::asio::dynamic_buffer(buffer_), boost::asio::transfer_exactly(9),
             boost::asio::bind_executor(
                 strand_, [this](error_code error, std::size_t length) {
                     if (!error) {
                         std::cout << buffer_ << std::endl;

                         send_request(std::move(buffer_));
                         recv(); // already on the strand in this completion handler
                     } else {
                         std::cout << "Read failed: " << error.message() << std::endl;
                     }
                 }));
     }

     boost::asio::strand<Executor> strand_;
     ssl::stream<tcp::socket>      socket_;
     std::string                   buffer_;
     std::deque<std::string>       outbox_;
 };

 int main(int argc, char* argv[])
 {
     try {
         if (argc != 3) {
             std::cerr << "Usage: client <host> <port>\n";
             return 1;
         }

         ssl::context ctx(ssl::context::sslv23);
         ctx.load_verify_file("ca.pem");

         boost::asio::thread_pool io;
         tcp::resolver resolver(io);

         client c(io.get_executor(), ctx, resolver.resolve(argv[1], argv[2]));
         io.join();

         return 0;
     } catch (std::exception const& e) {
         std::cerr << "Exception: " << e.what() << "\n";
         return 1;
     }
 }

文件server.cpp

 #include <boost/asio.hpp>
 #include <boost/asio/ssl.hpp>
 #include <boost/bind/bind.hpp>
 #include <cstdlib>
 #include <functional>
 #include <iostream>

 namespace ssl = boost::asio::ssl;
 using boost::asio::ip::tcp;
 using boost::system::error_code;

 class session : public std::enable_shared_from_this<session> {
   public:
     session(tcp::socket socket, ssl::context& context)
         : socket_(std::move(socket), context)
     {
     }

     void start()
     {
         do_handshake();
     }

   private:
     void do_handshake()
     {
         auto self(shared_from_this());
         socket_.async_handshake(ssl::stream_base::server,
                                 [this, self](error_code error) {
                                     if (!error) {
                                         do_read();
                                     }
                                 });
     }

     void do_read()
     {
         auto self(shared_from_this());

         socket_.async_read_some(
             boost::asio::buffer(data_),
             [this, self](error_code ec, std::size_t length) {
                 if (!ec) {
                     std::cout << "get <";
                     std::cout.write(data_.data(), length);
                     std::cout << std::endl;
                     do_write(length);
                 }
             });
     }

     void do_write(std::size_t length)
     {
         auto self(shared_from_this());
         std::cout << "send <";
         std::cout.write(data_.data(), length);
         std::cout << std::endl;
         boost::asio::async_write(
             socket_, boost::asio::buffer(data_.data(), length),
             [this, self](error_code ec, std::size_t /*length*/) {
                 if (!ec) {
                     do_read();
                 }
             });
     }

     ssl::stream<tcp::socket> socket_;
     std::array<char, 1024>   data_;
 };

 class server {
   public:
     server(boost::asio::io_context& io_context, unsigned short port)
         : acceptor_(io_context, tcp::endpoint(tcp::v4(), port))
         , context_(ssl::context::sslv23)
     {
         context_.set_options(ssl::context::default_workarounds |
                              ssl::context::no_sslv2 |
                              ssl::context::single_dh_use);
         context_.set_password_callback(std::bind(&server::get_password, this));
         context_.use_certificate_chain_file("server.pem");
         context_.use_private_key_file("server.pem", ssl::context::pem);
         context_.use_tmp_dh_file("dh2048.pem");

         do_accept();
     }

   private:
     std::string get_password() const
     {
         return "test";
     }

     void do_accept()
     {
         acceptor_.async_accept([this](error_code error, tcp::socket socket) {
             if (!error) {
                 std::make_shared<session>(std::move(socket), context_)->start();
             }

             do_accept();
         });
     }

     tcp::acceptor acceptor_;
     ssl::context  context_;
 };

 int main(int argc, char* argv[])
 {
     try {
         if (argc != 2) {
             std::cerr << "Usage: server <port>\n";
             return 1;
         }

         boost::asio::io_context io_context;

         server s(io_context, std::atoi(argv[1]));

         io_context.run();
         return 0;
     } catch (std::exception const& e) {
         std::cerr << "Exception: " << e.what() << "\n";
         return 1;
     }
 }

现场演示：

如您所见（使用 uniq -dc 技巧来抑制所有非重复行）现在它可以在启动发送之前收到多个接收的情况下愉快地继续。

boost::asio::async_write 写入 ssl::stream 成功但服务器未获取

boost::asio::async_write write ssl::stream succuess but server not get

c++

ssl

asynchronous

boost-asio

其他问题

终身错误

并发写入

固定解