uhd: Update ASIO usage to modern style, compatible with Boost 1.87

Starting with Boost 1.66 and the corresponding ASIO version, there were
some changes introduced based on the C++ Networking TS. This includes
changes like replacing io_service with io_context, deprecating some
functions, etc. Starting with Boost 1.87, the old style is no longer
supported.

This commit updates all usage of ASIO in a way that makes UHD compatible
with future versions of ASIO. However, this makes UHD no longer
compatible with Boost 1.65 and below.

Summary of changes:
- Replace asio::io_service with asio::io_context
- Replace asio::io_service::strand with
  asio::strand<asio::io_context::executor_type>
  - This implies using asio::post() instead of asio::strand::post()
- Replace asio::buffer_cast<T>(buf) with static_cast<T>(buf.data())
- Update resolve(query) with new API
- Replace references to endpoint_iterator with resolver::results_type
- Replace ip::address::from_string() with ip::make_address()

Co-authored-by: Jörg Hofrichter <[email protected]>

Author: mbr0wn

host/examples/network_relay.cpp

@@ -78,21 +78,21 @@ class udp_relay_type
         : _port(port)
     {
         {
-            asio::ip::udp::resolver resolver(_io_service);
-            asio::ip::udp::resolver::query query(asio::ip::udp::v4(), server_addr, port);
-            asio::ip::udp::endpoint endpoint = *resolver.resolve(query);
+            asio::ip::udp::resolver resolver(_io_context);
+            asio::ip::udp::endpoint endpoint =
+                *resolver.resolve(asio::ip::udp::v4(), server_addr, port).begin();
 
             _server_socket = std::shared_ptr<asio::ip::udp::socket>(
-                new asio::ip::udp::socket(_io_service, endpoint));
+                new asio::ip::udp::socket(_io_context, endpoint));
             resize_buffs(_server_socket, server_rx_size, server_tx_size);
         }
         {
-            asio::ip::udp::resolver resolver(_io_service);
-            asio::ip::udp::resolver::query query(asio::ip::udp::v4(), client_addr, port);
-            asio::ip::udp::endpoint endpoint = *resolver.resolve(query);
+            asio::ip::udp::resolver resolver(_io_context);
+            asio::ip::udp::endpoint endpoint =
+                *resolver.resolve(asio::ip::udp::v4(), client_addr, port).begin();
 
             _client_socket = std::shared_ptr<asio::ip::udp::socket>(
-                new asio::ip::udp::socket(_io_service));
+                new asio::ip::udp::socket(_io_context));
             _client_socket->open(asio::ip::udp::v4());
             _client_socket->connect(endpoint);
             resize_buffs(_client_socket, client_rx_size, client_tx_size);
@@ -173,7 +173,7 @@ class udp_relay_type
 
     const std::string _port;
     boost::thread_group _thread_group;
-    asio::io_service _io_service;
+    asio::io_context _io_context;
     asio::ip::udp::endpoint _endpoint;
     std::mutex _endpoint_mutex;
     socket_type _server_socket, _client_socket;

host/include/uhd/transport/nirio/rpc/rpc_client.hpp

@@ -56,20 +56,20 @@ class rpc_client : private uhd::noncopyable
         const boost::system::error_code& err, size_t transferred, size_t expected);
     void _wait_for_next_response_header();
 
-    inline void _stop_io_service()
+    inline void _stop_io_context()
     {
-        if (_io_service_thread.get()) {
+        if (_io_context_thread.get()) {
             UHD_LOG_DEBUG("NIRIO", "rpc_client stopping...");
-            _io_service.stop();
-            _io_service_thread->join();
-            _io_service_thread.reset();
+            _io_context.stop();
+            _io_context_thread->join();
+            _io_context_thread.reset();
             UHD_LOG_DEBUG("NIRIO", "rpc_client stopped.");
         }
     }
 
     // Services
-    boost::asio::io_service _io_service;
-    std::unique_ptr<boost::thread> _io_service_thread;
+    boost::asio::io_context _io_context;
+    std::unique_ptr<boost::thread> _io_context_thread;
     boost::asio::ip::tcp::socket _socket;
     // Handshake info
     hshake_args_t _hshake_args_client;

host/lib/deps/rpclib/include/rpc/detail/async_writer.h

@@ -19,9 +19,9 @@ namespace detail {
 //! \brief Common logic for classes that have a write queue with async writing.
 class async_writer : public std::enable_shared_from_this<async_writer> {
 public:
-    async_writer(boost::asio::io_service *io,
+    async_writer(boost::asio::io_context* io,
                  boost::asio::ip::tcp::socket socket)
-        : socket_(std::move(socket)), write_strand_(*io), exit_(false) {}
+        : socket_(std::move(socket)), write_strand_(io->get_executor()), exit_(false) {}
 
     void do_write() {
         if (exit_) {
@@ -31,9 +31,9 @@ class async_writer : public std::enable_shared_from_this<async_writer> {
         auto &item = write_queue_.front();
         // the data in item remains valid until the handler is called
         // since it will still be in the queue physically until then.
-        boost::asio::async_write(
-            socket_, boost::asio::buffer(item.data(), item.size()),
-            write_strand_.wrap(
+        boost::asio::async_write(socket_,
+            boost::asio::buffer(item.data(), item.size()),
+            boost::asio::bind_executor(write_strand_,
                 [this, self](boost::system::error_code ec, std::size_t transferred) {
                     (void)transferred;
                     if (!ec) {
@@ -69,7 +69,7 @@ class async_writer : public std::enable_shared_from_this<async_writer> {
 
 protected:
     boost::asio::ip::tcp::socket socket_;
-    boost::asio::io_service::strand write_strand_;
+    boost::asio::strand<boost::asio::io_context::executor_type> write_strand_;
     std::atomic_bool exit_{false};
     bool exited_ = false;
     std::mutex m_exit_;

host/lib/deps/rpclib/include/rpc/detail/server_session.h

@@ -22,7 +22,7 @@ namespace detail {
 
 class server_session : public async_writer {
 public:
-    server_session(server *srv, boost::asio::io_service *io,
+    server_session(server* srv, boost::asio::io_context* io,
                    boost::asio::ip::tcp::socket socket,
                    std::shared_ptr<dispatcher> disp, bool suppress_exceptions);
     void start();
@@ -34,8 +34,8 @@ class server_session : public async_writer {
 
 private:
     server* parent_;
-    boost::asio::io_service *io_;
-    boost::asio::io_service::strand read_strand_;
+    boost::asio::io_context* io_;
+    boost::asio::strand<boost::asio::io_context::executor_type> read_strand_;
     std::shared_ptr<dispatcher> disp_;
     RPCLIB_MSGPACK::unpacker pac_;
     RPCLIB_MSGPACK::sbuffer output_buf_;

host/lib/deps/rpclib/lib/rpc/client.cc

@@ -38,7 +38,7 @@ struct client::impl {
     impl(client *parent, std::string const &addr, uint16_t port)
         : parent_(parent),
           io_(),
-          strand_(io_),
+          strand_(io_.get_executor()),
           call_idx_(0),
           addr_(addr),
           port_(port),
@@ -50,11 +50,11 @@ struct client::impl {
         pac_.reserve_buffer(default_buffer_size);
     }
 
-    void do_connect(tcp::resolver::iterator endpoint_iterator) {
+    void do_connect(const tcp::resolver::results_type& endpoints) {
         LOG_INFO("Initiating connection.");
         boost::asio::async_connect(
-            writer_->socket_, endpoint_iterator,
-            [this](boost::system::error_code ec, tcp::resolver::iterator) {
+            writer_->socket_, endpoints,
+            [this](boost::system::error_code ec, tcp::endpoint) {
                 std::unique_lock<std::mutex> lock(mut_connection_finished_);
                 if (!ec) {
                     LOG_INFO("Client connected to {}:{}", addr_, port_);
@@ -101,7 +101,7 @@ struct client::impl {
                             std::get<1>(current_call)
                                 .set_exception(std::current_exception());
                         }
-                        strand_.post(
+                        boost::asio::post(strand_,
                             [this, id]() { ongoing_calls_.erase(id); });
                     }
 
@@ -144,8 +144,8 @@ struct client::impl {
         std::pair<std::string, std::promise<RPCLIB_MSGPACK::object_handle>>;
 
     client *parent_;
-    boost::asio::io_service io_;
-    boost::asio::io_service::strand strand_;
+    boost::asio::io_context io_;
+    boost::asio::strand<boost::asio::io_context::executor_type> strand_;
     std::atomic<int> call_idx_; /// The index of the last call made
     std::unordered_map<uint32_t, call_t> ongoing_calls_;
     std::string addr_;
@@ -164,9 +164,9 @@ struct client::impl {
 client::client(std::string const &addr, uint16_t port)
     : pimpl(new client::impl(this, addr, port)) {
     tcp::resolver resolver(pimpl->io_);
-    auto endpoint_it =
-        resolver.resolve({pimpl->addr_, std::to_string(pimpl->port_)});
-    pimpl->do_connect(endpoint_it);
+    auto endpoints =
+        resolver.resolve(pimpl->addr_, std::to_string(pimpl->port_));
+    pimpl->do_connect(endpoints);
     std::thread io_thread([this]() {
         RPCLIB_CREATE_LOG_CHANNEL(client)
         name_thread("client");
@@ -190,15 +190,15 @@ int client::get_next_call_idx() {
 void client::post(std::shared_ptr<RPCLIB_MSGPACK::sbuffer> buffer, int idx,
                   std::string const &func_name,
                   std::shared_ptr<rsp_promise> p) {
-    pimpl->strand_.post([buffer, idx, func_name, p, this]() {
+    boost::asio::post(pimpl->strand_, [buffer, idx, func_name, p, this]() {
         pimpl->ongoing_calls_.insert(
             std::make_pair(idx, std::make_pair(func_name, std::move(*p))));
         pimpl->write(std::move(*buffer));
     });
 }
 
 void client::post(RPCLIB_MSGPACK::sbuffer *buffer) {
-    pimpl->strand_.post([buffer, this]() {
+    boost::asio::post(pimpl->strand_, [buffer, this]() {
         pimpl->write(std::move(*buffer));
         delete buffer;
     });

host/lib/deps/rpclib/lib/rpc/detail/server_session.cc

@@ -22,14 +22,14 @@ namespace detail {
 
 static constexpr std::size_t default_buffer_size = rpc::constants::DEFAULT_BUFFER_SIZE;
 
-server_session::server_session(server *srv, boost::asio::io_service *io,
+server_session::server_session(server *srv, boost::asio::io_context *io,
                                boost::asio::ip::tcp::socket socket,
                                std::shared_ptr<dispatcher> disp,
                                bool suppress_exceptions)
     : async_writer(io, std::move(socket)),
       parent_(srv),
       io_(io),
-      read_strand_(*io),
+      read_strand_(io->get_executor()),
       disp_(disp),
       pac_(),
       suppress_exceptions_(suppress_exceptions) {
@@ -42,7 +42,7 @@ void server_session::start() { do_read(); }
 void server_session::close() {
     LOG_INFO("Closing session.");
     exit_ = true;
-    write_strand_.post([this]() {
+    boost::asio::post(write_strand_, [this]() {
         try {
             socket_.close();
         } catch (const boost::system::system_error&) {
@@ -54,12 +54,11 @@ void server_session::close() {
 void server_session::do_read() {
     auto self(shared_from_this());
     constexpr std::size_t max_read_bytes = default_buffer_size;
-    socket_.async_read_some(
-        boost::asio::buffer(pac_.buffer(), default_buffer_size),
+    socket_.async_read_some(boost::asio::buffer(pac_.buffer(), default_buffer_size),
         // I don't think max_read_bytes needs to be captured explicitly
         // (since it's constexpr), but MSVC insists.
-        read_strand_.wrap([this, self, max_read_bytes](boost::system::error_code ec,
-                                                       std::size_t length) {
+        boost::asio::bind_executor(read_strand_, [this, self, max_read_bytes](boost::system::error_code ec,
+                                                                              std::size_t length) {
             if (!ec) {
                 pac_.buffer_consumed(length);
                 RPCLIB_MSGPACK::unpacked result;
@@ -69,7 +68,7 @@ void server_session::do_read() {
 
                     // any worker thread can take this call
                     auto z = std::shared_ptr<RPCLIB_MSGPACK::zone>(result.zone().release());
-                    io_->post([
+                    boost::asio::post(io_->get_executor(), [
                         this, msg, z
                     ]() {
                         this_handler().clear();
@@ -102,10 +101,10 @@ void server_session::do_read() {
                         if (!resp.is_empty()) {
 #ifdef _MSC_VER
                             // doesn't compile otherwise.
-                            write_strand_.post(
+                            boost::asio::post(write_strand_,
                                 [=]() { write(resp.get_data()); });
 #else
-                            write_strand_.post(
+                            boost::asio::post(write_strand_,
                                 [this, resp, z]() { write(resp.get_data()); });
 #endif
                         }
@@ -114,14 +113,14 @@ void server_session::do_read() {
                             LOG_WARN("Session exit requested from a handler.");
                             // posting through the strand so this comes after
                             // the previous write
-                            write_strand_.post([this]() { exit_ = true; });
+                            boost::asio::post(write_strand_, [this]() { exit_ = true; });
                         }
 
                         if (this_server().stopping_) {
                             LOG_WARN("Server exit requested from a handler.");
                             // posting through the strand so this comes after
                             // the previous write
-                            write_strand_.post(
+                            boost::asio::post(write_strand_,
                                 [this]() { parent_->close_sessions(); });
                         }
                     });

host/lib/deps/rpclib/lib/rpc/server.cc

@@ -25,7 +25,7 @@ struct server::impl {
         : parent_(parent),
           io_(),
           acceptor_(io_,
-                    tcp::endpoint(ip::address::from_string(address), port)),
+                    tcp::endpoint(ip::make_address(address), port)),
           socket_(io_),
           suppress_exceptions_(false) {}
 
@@ -66,7 +66,7 @@ struct server::impl {
     }
 
     server *parent_;
-    io_service io_;
+    io_context io_;
     ip::tcp::acceptor acceptor_;
     ip::tcp::socket socket_;
     rpc::detail::thread_group loop_workers_;

host/lib/include/uhdlib/transport/udp_boost_asio_link.hpp

@@ -146,7 +146,7 @@ class udp_boost_asio_link : public recv_link_base<udp_boost_asio_link>,
     std::vector<udp_boost_asio_frame_buff> _recv_buffs;
     std::vector<udp_boost_asio_frame_buff> _send_buffs;
 
-    boost::asio::io_service _io_service;
+    boost::asio::io_context _io_context;
     std::shared_ptr<boost::asio::ip::udp::socket> _socket;
     int _sock_fd;
     adapter_id_t _adapter_id;

host/lib/include/uhdlib/transport/udp_common.hpp

@@ -81,17 +81,16 @@ UHD_INLINE bool wait_for_recv_ready(int sock_fd, int32_t timeout_ms)
 }
 
 UHD_INLINE socket_sptr open_udp_socket(
-    const std::string& addr, const std::string& port, boost::asio::io_service& io_service)
+    const std::string& addr, const std::string& port, boost::asio::io_context& io_context)
 {
     using udp = boost::asio::ip::udp;
 
     // resolve the address
-    udp::resolver resolver(io_service);
-    udp::resolver::query query(udp::v4(), addr, port);
-    udp::endpoint receiver_endpoint = *resolver.resolve(query);
+    udp::resolver resolver(io_context);
+    udp::endpoint receiver_endpoint = *resolver.resolve(udp::v4(), addr, port).begin();
 
     // create, open, and connect the socket
-    socket_sptr socket = socket_sptr(new udp::socket(io_service));
+    socket_sptr socket = socket_sptr(new udp::socket(io_context));
     socket->open(udp::v4());
     socket->connect(receiver_endpoint);
 

host/lib/include/uhdlib/utils/eeprom_utils.hpp

@@ -10,7 +10,6 @@
 #include <uhd/types/dict.hpp>
 #include <uhd/types/mac_addr.hpp>
 #include <uhd/utils/log.hpp>
-#include <boost/asio/ip/address_v4.hpp>
 #include <string>
 #include <vector>
 
@@ -29,7 +28,7 @@ std::string uint16_bytes_to_string(const uhd::byte_vector_t& bytes);
  * see if the resulting contents will contain duplicates.  Useful error
  * messages are logged describing any duplicates.
  *
- * <field_type> must provide to_string() and from_string() functions
+ * <field_type> must provide from_string() functions
  *
  * \param error_label Label to put on error messages
  * \param new_eeprom New EEPROM contents
@@ -38,12 +37,13 @@ std::string uint16_bytes_to_string(const uhd::byte_vector_t& bytes);
  * \param keys Keys to examine for duplicate values
  * \return true if duplicates are found, false if not
  */
-template <typename field_type>
+template <typename str_normalizer_type>
 bool check_for_duplicates(const std::string& error_label,
     const uhd::dict<std::string, std::string>& new_eeprom,
     const uhd::dict<std::string, std::string>& curr_eeprom,
     const std::string& category,
-    const std::vector<std::string>& keys)
+    const std::vector<std::string>& keys,
+    str_normalizer_type&& str_normalizer)
 {
     bool has_duplicates = false;
     for (size_t i = 0; i < keys.size(); i++) {
@@ -54,7 +54,7 @@ bool check_for_duplicates(const std::string& error_label,
             continue;
         }
 
-        auto value = field_type::from_string(new_eeprom[key]).to_string();
+        auto value = str_normalizer(new_eeprom[key]);
 
         // Check other values in new_eeprom for duplicate
         // Starting at key index i+1 so the same duplicate is not found twice
@@ -63,7 +63,7 @@ bool check_for_duplicates(const std::string& error_label,
             if (not new_eeprom.has_key(other_key)) {
                 continue;
             }
-            auto other_value = field_type::from_string(new_eeprom[other_key]).to_string();
+            auto other_value = str_normalizer(new_eeprom[other_key]);
             if (value == other_value) {
                 // Value is a duplicate of another supplied value
                 UHD_LOG_ERROR(error_label,

host/lib/transport/dpdk_simple.cpp

@@ -96,7 +96,7 @@ class dpdk_simple_impl : public dpdk_simple
         // Extract buff and sanity check
         const size_t nbytes = boost::asio::buffer_size(user_buff);
         UHD_ASSERT_THROW(nbytes <= _link->get_send_frame_size())
-        const uint8_t* user_data = boost::asio::buffer_cast<const uint8_t*>(user_buff);
+        const uint8_t* user_data = static_cast<const uint8_t*>(user_buff.data());
 
         // Get send buff
         auto buff = _send_io->get_send_buff(SEND_TIMEOUT_MS);
@@ -120,7 +120,7 @@ class dpdk_simple_impl : public dpdk_simple
     size_t recv(const boost::asio::mutable_buffer& user_buff, double timeout) override
     {
         size_t user_buff_size = boost::asio::buffer_size(user_buff);
-        uint8_t* user_data    = boost::asio::buffer_cast<uint8_t*>(user_buff);
+        uint8_t* user_data    = static_cast<uint8_t*>(user_buff.data());
 
         auto buff = _recv_io->get_recv_buff(static_cast<int32_t>(timeout * 1e3));
         if (!buff) {