如何在 C++ 中关闭/结束阻塞的 boost::asio::read 从串行端口操作，以便 std::thread 可以加入？

Question

由于是同步读取，所以是阻塞操作，t1永远不会加入主线程。

主要问题：

0. 如何转换此代码以使用简单的截止日期计时器？
1. 如何以尽可能简单的方式使用async？
2. 是否可以在codeblock m2中模拟串行端口的输入以解锁读取？如果是的话，怎么办？

// #include headers

bool flag = true; 

void f1()
{
   while(flag)
   {
      // BLOCKING READ OPERATION
      boost::asio::read( serial_port, boost::asio::buffer( byte ,1 )) ;
      // codeblock f1 -> do stuff with byte
   }
}

void main()
{
    std::thread t1 (f1); 
    
    // codeblock m1 -> do stuff

    flag = false ; 

    // codeblock m2 -> do more stuff
   
    t1.join(); 

    // codeblock m3 -> do eve more stuff

}

我研究了需要使用异步读取的不同资源，但这并不是非常简单。 所以我写这篇文章是为了准确讨论以最简单和最少的代码解决这个问题的策略。

Answer 1

问题顺序错误:)

我将从最简单的开始，最后我们将解决您描述的选项。

1. 确实，阻塞无法取消完全停止。代码被修复为独立的，没有全局变量和数据竞争，如下所示：

   #include <boost/asio.hpp>
   #include <iomanip>
   #include <iostream>
   namespace asio = boost::asio;
   using std::this_thread::sleep_for;
   using namespace std::chrono_literals;
   
   struct Program {
       std::atomic_bool     flag{true};
       std::array<char, 10> buf;
   
       asio::serial_port serial_port;
   
       Program(std::string device) : serial_port{asio::system_executor{}, device} {}
   
       void read_loop() {
           while (flag) {
               auto n = serial_port.read_some(asio::buffer(buf));
               f1(n);
           }
       }
   
       void f1(size_t n) {
           std::cout << "f1 do stuff with bytes: " << quoted(std::string_view(buf.data(), n)) << std::endl;
       }
   };
   
   void m1() {
       std::cout << "m1 do stuff" << std::endl;
       sleep_for(6s);
   }
   
   void m2() {
       std::cout << "m2 do more stuff" << std::endl;
       sleep_for(4s);
   }
   
   void m3() { std::cout << "m3 do even more stuff (bye)" << std::endl; }
   
   int main(int argc, char** argv) {
       Program p(argc > 1 ? argv[1] : "/dev/pts/5");
   
       std::thread t1([&p] { p.read_loop(); });
   
       m1();
       p.flag = false;
       m2();
       t1.join(); 
       m3();
   }
   

   我可以使用 socat 作为串口在本地进行演示：

   

2. 最简单的异步：

   using Buffer = std::array<char, 10>;
   
   void read_loop(asio::serial_port& stream, Buffer& buf) {
       stream.async_read_some(asio::buffer(buf), [&](error_code ec, size_t n) {
           std::cout << "f1 do stuff with bytes: " << quoted(std::string_view(buf.data(), n)) << " (" << ec.message() << ")" << std::endl;
           if (!ec.failed())
               read_loop(stream, buf);
       });
   }
   
   int main(int argc, char** argv) {
       asio::io_context io(1);
   
       asio::serial_port sp{make_strand(io),  argc > 1 ? argv[1] : "/dev/pts/5"};
   
       Buffer buf;
       read_loop(sp, buf);
   
       std::cout << "m1 do stuff" << std::endl;
       io.run_for(6s);
   
       std::cout << "m2 do more stuff" << std::endl;
       sleep_for(4s);
   
       std::cout << "m3 do even more stuff (bye)" << std::endl;
   }
   

   要获得“真正的”取消和加入，请将

   sleep_for(4s)

   替换为例如

   sp.cancel();
   io.run(); // "join"
   

   或者确实

   sp.cancel();
   io.run_for(4s);
   

   如果您不相信取消总是有效

3. 前面已经给出了最简单形式的截止日期：只需停止 io 上下文即可。

4. 作为中间步骤，让我们再次将端口和操作封装到一个类中，同时将 IO 放在加入的线程上，这样我们就可以在 main 中实际执行工作了：

   struct Program {
       Program(asio::any_io_executor ex, std::string device) : serial_port{ex, device} { read_loop(); }
   
       void cancel() {
           asio::post(serial_port.get_executor(), [this] { serial_port.cancel(); });
       }
   
     private:
       asio::serial_port serial_port;
       std::array<char, 10> buf;
   
       void read_loop() {
           serial_port.async_read_some(asio::buffer(buf), [this](error_code ec, size_t n) {
               std::cout << "f1 do stuff with bytes: " << quoted(std::string_view(buf.data(), n)) << "(" << ec.message() << ")" << std::endl;
               if (!ec.failed())
                   read_loop();
           });
       }
   };
   
   void m1() {
       std::cout << "m1 do stuff" << std::endl;
       sleep_for(6s);
   }
   
   void m2() {
       std::cout << "m2 do more stuff" << std::endl;
       sleep_for(4s);
   }
   
   void m3() { std::cout << "m3 do even more stuff (bye)" << std::endl; }
   
   int main(int argc, char** argv) {
       asio::thread_pool io(1);
       Program p(make_strand(io), argc > 1 ? argv[1] : "/dev/pts/5");
   
       m1();
       p.cancel();
       m2();
       io.join(); 
       m3();
   }
   

   仍然如广告所示工作：

5. 为具体操作添加明确的期限：

   using duration = std::chrono::steady_clock::duration;
   
   struct Program {
       Program(asio::any_io_executor ex, duration dur, std::string device)
           : serial_port{ex, device}
           , deadline{ex, dur} //
       {
           deadline.async_wait([this](error_code ec) {
               if (!ec) {
                   std::cerr << "Deadline" << std::endl;
                   serial_port.cancel();
               }
           });
           read_loop();
       }
   
       void cancel() {
           asio::post(serial_port.get_executor(), [this] { serial_port.cancel(); });
       }
   
     private:
       asio::serial_port    serial_port;
       asio::steady_timer   deadline;
       std::array<char, 10> buf;
   
       void read_loop() {
           serial_port.async_read_some(asio::buffer(buf), [this](error_code ec, size_t n) {
               std::cout << "f1 do stuff with bytes: " << quoted(std::string_view(buf.data(), n)) << "(" << ec.message() << ")" << std::endl;
               if (!ec.failed())
                   read_loop();
           });
       }
   };
   
   int main(int argc, char** argv) {
       asio::thread_pool io(1);
       Program p(make_strand(io), 4s, argc > 1 ? argv[1] : "/dev/pts/5");
   
       std::cout << "m1 do stuff" << std::endl;
       sleep_for(6s);
       std::cout << "m2 do more stuff" << std::endl;
       sleep_for(4s);
   
       io.join();
       std::cout << "m3 do even more stuff (bye)" << std::endl;
   }
   

   注意 4s 的截止日期现在是如何独立于 main 中的工作而发生的。另请注意，如果您愿意，您仍然可以通过

   cancel()

   main

   进行操作。