Tôi có thể đọc từ ổ cắm đồng bộ bằng Boost.Asio với thời gian chờ trên dịch vụ I/O đa luồng không?

Question

Tôi có một ứng dụng sử dụng Boost.Asio cho giao tiếp TCP và UDP socket. Tôi hiểu rằng 'A' trong "Asio" là viết tắt của Không đồng bộ, do đó thư viện được uốn cong về việc khuyến khích bạn sử dụng I/O không đồng bộ khi có thể. Tôi có một vài trường hợp đọc đồng bộ socket thích hợp hơn. Đồng thời, tuy nhiên, tôi muốn thiết lập một thời gian chờ trên cho biết nhận cuộc gọi, do đó, không có khả năng chặn đọc vô thời hạn.

này dường như là một vấn đề khá phổ biến giữa các người dùng Boost.Asio, với những điều sau đây câu hỏi Stack Overflow về chủ đề này:

• C++ Boost ASIO: how to read/write with a timeout?
• asio::read with timeout
• boost asio timeout
• How to set a timeout on blocking sockets in boost asio?

Có thể ev vi được nhiều hơn. Thậm chí còn có examples in the documentation để biết cách triển khai các hoạt động đồng bộ với thời gian chờ. Chúng đun sôi xuống để chuyển đổi hoạt động đồng bộ thành một hoạt động không đồng bộ, sau đó khởi động nó song song với asio::deadline_timer. Trình xử lý hết hạn của bộ hẹn giờ sau đó có thể hủy đọc không đồng bộ trong trường hợp hết thời gian chờ. Đây trông giống như thế này (đoạn lấy từ ví dụ liên kết ở trên):

std::size_t receive(const boost::asio::mutable_buffer& buffer, 
     boost::posix_time::time_duration timeout, boost::system::error_code& ec) 
    { 
    // Set a deadline for the asynchronous operation. 
    deadline_.expires_from_now(timeout); 

    // Set up the variables that receive the result of the asynchronous 
    // operation. The error code is set to would_block to signal that the 
    // operation is incomplete. Asio guarantees that its asynchronous 
    // operations will never fail with would_block, so any other value in 
    // ec indicates completion. 
    ec = boost::asio::error::would_block; 
    std::size_t length = 0; 

    // Start the asynchronous operation itself. The handle_receive function 
    // used as a callback will update the ec and length variables. 
    socket_.async_receive(boost::asio::buffer(buffer), 
     boost::bind(&client::handle_receive, _1, _2, &ec, &length)); 

    // Block until the asynchronous operation has completed. 
    do io_service_.run_one(); while (ec == boost::asio::error::would_block); 

    return length; 
    } 

Đây thực sự là một giải pháp tương đối sạch: bắt đầu các hoạt động không đồng bộ, sau đó tự thăm dò ý kiến ​​các asio::io_service để thực hiện xử lý không đồng bộ cùng một lúc cho đến khi một trong hai các async_receive() hoàn thành hoặc bộ đếm thời gian hết hạn.

Tuy nhiên, điều gì xảy ra với trường hợp dịch vụ I/O cơ bản của socket đang được chạy trong một hoặc nhiều chủ đề nền? Trong trường hợp đó, không có sự đảm bảo rằng các trình xử lý cho các hoạt động không đồng bộ sẽ được chạy bởi luồng tiền cảnh trong đoạn mã trên, vì vậy run_one() sẽ không trả lại cho đến khi một số sau đó, có thể không liên quan, trình xử lý thực thi. Điều này sẽ làm cho ổ cắm đọc khá không phản hồi.

asio::io_service có chức năng poll_one() sẽ kiểm tra hàng đợi của dịch vụ mà không chặn, nhưng tôi không thấy cách tốt để chặn luồng tiền cảnh (mô phỏng hành vi cuộc gọi đồng bộ) cho đến khi trình xử lý thực hiện, ngoại trừ trường hợp không có chủ đề nền nào đang thực thi asio::io_service::run() rồi.

tôi thấy hai giải pháp tiềm năng, không ai trong số đó tôi thích:

1. Sử dụng một biến tình trạng hoặc xây dựng tương tự để làm cho khối chủ đề foreground sau khi bắt đầu hoạt động không đồng bộ. Trong trình xử lý cuộc gọi async_receive(), hãy báo hiệu biến điều kiện để bỏ chặn chuỗi. Điều này gây ra một số khóa cho mỗi đọc, mà tôi muốn tránh, vì tôi muốn đạt được thông lượng tối đa có thể trên ổ cắm UDP đọc. Nếu không, nó là khả thi, và có lẽ là những gì tôi sẽ làm trừ khi một phương pháp cao cấp trình bày chính nó.

2. Đảm bảo rằng ổ cắm có riêng asio::io_service không được chạy bởi bất kỳ chủ đề nền nào. Điều này làm cho việc sử dụng I/O không đồng bộ với ổ cắm trở nên khó khăn hơn trong trường hợp mong muốn.

Bất kỳ ý tưởng nào về cách khác để thực hiện điều này một cách an toàn?

---

Ngoài: Có một số câu trả lời cho câu hỏi trước SO rằng chủ trương sử dụng tùy chọn SO_RCVTIMEO ổ cắm để thực hiện thời gian chờ ổ cắm đọc. Điều này nghe có vẻ tuyệt vời trong lý thuyết, nhưng nó dường như không hoạt động trên nền tảng của tôi ít nhất (Ubuntu 12.04, Boost v1.55). Tôi có thể đặt thời gian chờ của ổ cắm, nhưng nó sẽ không cho hiệu ứng mong muốn với Asio. Các mã có liên quan là trong /boost/asio/detail/impl/socket_ops.ipp:

size_t sync_recvfrom(socket_type s, state_type state, buf* bufs, 
    size_t count, int flags, socket_addr_type* addr, 
    std::size_t* addrlen, boost::system::error_code& ec) 
{ 
    if (s == invalid_socket) 
    { 
    ec = boost::asio::error::bad_descriptor; 
    return 0; 
    } 

    // Read some data. 
    for (;;) 
    { 
    // Try to complete the operation without blocking. 
    signed_size_type bytes = socket_ops::recvfrom(
     s, bufs, count, flags, addr, addrlen, ec); 

    // Check if operation succeeded. 
    if (bytes >= 0) 
     return bytes; 

    // Operation failed. 
    if ((state & user_set_non_blocking) 
     || (ec != boost::asio::error::would_block 
      && ec != boost::asio::error::try_again)) 
     return 0; 

    // Wait for socket to become ready. 
    if (socket_ops::poll_read(s, 0, ec) < 0) 
     return 0; 
    } 
} 

Nếu một ổ cắm đọc lần ra ngoài, cuộc gọi đến recvfrom() trên sẽ trở EAGAIN hoặc EWOULDBLOCK, mà có được phiên dịch sang boost::asio::error::try_again hoặc boost::asio::error::would_block. Trong trường hợp này, các mã trên sẽ gọi poll_read() chức năng, mà cho nền tảng của tôi trông giống như:

int poll_read(socket_type s, state_type state, boost::system::error_code& ec) 
{ 
    if (s == invalid_socket) 
    { 
    ec = boost::asio::error::bad_descriptor; 
    return socket_error_retval; 
    } 

    pollfd fds; 
    fds.fd = s; 
    fds.events = POLLIN; 
    fds.revents = 0; 
    int timeout = (state & user_set_non_blocking) ? 0 : -1; 
    clear_last_error(); 
    int result = error_wrapper(::poll(&fds, 1, timeout), ec); 
    if (result == 0) 
    ec = (state & user_set_non_blocking) 
     ? boost::asio::error::would_block : boost::system::error_code(); 
    else if (result > 0) 
    ec = boost::system::error_code(); 
    return result; 
} 

tôi snipped ra mã có điều kiện biên soạn cho các nền tảng khác. Như bạn có thể thấy, nếu ổ cắm không phải là một ổ cắm không chặn, nó kết thúc lên gọi poll() với thời gian chờ vô hạn, do đó chặn cho đến khi ổ cắm có dữ liệu được đọc (và làm hỏng nỗ lực tại thời gian chờ). Do đó, tùy chọn SO_RCVTIMEO không hiệu quả.

Answer 1

Boost.Asio hỗ trợ cho futures có thể cung cấp giải pháp thanh lịch. Khi một phép toán không đồng bộ được cung cấp giá trị boost::asio::use_future làm bộ xử lý hoàn thành của nó, hàm khởi tạo sẽ trả về một đối tượng std::future sẽ nhận được kết quả của phép toán. Ngoài ra, nếu hoạt động hoàn tất với lỗi, error_code được chuyển đổi thành system_error và được chuyển cho người gọi qua future.

Trong Boost.Asio C++ 11 Futures datytime client example, a thread chuyên dụng chạy io_service, và các chủ đề chính khởi hoạt động không đồng bộ sau đó đồng bộ chờ đợi về việc hoàn thành các hoạt động, chẳng hạn như sau:

std::array<char, 128> recv_buf; 
udp::endpoint sender_endpoint; 
std::future<std::size_t> recv_length = 
    socket.async_receive_from(
     boost::asio::buffer(recv_buf), 
     sender_endpoint, 
     boost::asio::use_future); 

// Do other things here while the receive completes. 

std::cout.write(
    recv_buf.data(), 
    recv_length.get()); // Blocks until receive is complete. 

Khi sử dụng future s, cách tiếp cận tổng thể để thực hiện đọc đồng bộ với thời gian chờ cũng giống như trước đây. Thay vì sử dụng một đồng bộ đọc, người ta sẽ sử dụng một đọc không đồng bộ và không đồng bộ chờ đợi trên một bộ đếm thời gian. Thay đổi nhỏ duy nhất là thay vì thay vì chặn trên io_service hoặc định kỳ kiểm tra biến vị ngữ, người ta sẽ gọi future::get() để chặn cho đến khi thao tác hoàn tất thành công hay thất bại (chẳng hạn như thời gian chờ).

Nếu không có sẵn C++ 11, thì kiểu trả về cho các hoạt động không đồng bộ có thể được tùy chỉnh cho số future của Boost.Thread, như được minh họa trong câu trả lời this.