Revert "wamp_transport, session, and boost::asio::io_context"

Reverting in an attempt to fix the mingw compilation/login crash issues.

This reverts commit 52f2b2e.

Author: jgriffiths

src/CMakeLists.txt

@@ -24,7 +24,6 @@ add_library(greenaddress-objects OBJECT
     ga_tx.cpp ga_tx.hpp
     ga_wally.cpp ga_wally.hpp
     http_client.cpp http_client.hpp
-    io_runner.cpp io_runner.hpp
     json_utils.cpp json_utils.hpp
     network_parameters.cpp network_parameters.hpp
     session.cpp session.hpp

src/ga_session.cpp

@@ -270,11 +270,11 @@ namespace sdk {
         , m_multi_call_category(0)
         , m_cache(std::make_shared<cache>(m_net_params, m_net_params.network()))
         , m_user_agent(std::string(GDK_COMMIT) + " " + m_net_params.user_agent())
+        , m_wamp(new wamp_transport(m_net_params,
+              std::bind(&ga_session::emit_notification, this, std::placeholders::_1, std::placeholders::_2)))
         , m_spv_thread_done(false)
         , m_spv_thread_stop(false)
     {
-        m_wamp = std::make_unique<wamp_transport>(m_net_params, *session_impl::m_io, *session_impl::m_strand,
-            std::bind(&ga_session::emit_notification, this, std::placeholders::_1, std::placeholders::_2));
         m_fee_estimates.assign(NUM_FEE_ESTIMATES, m_min_fee_rate);
     }
 
@@ -287,12 +287,6 @@ namespace sdk {
             constexpr bool do_start = false;
             download_headers_ctl(locker, do_start);
         });
-        no_std_exception_escape(
-            [this] {
-                m_wamp->disconnect();
-                m_wamp.reset();
-            },
-            "ga_session wamp_transport");
     }
 
     void ga_session::connect()

src/ga_session.hpp

@@ -18,7 +18,6 @@ namespace ga {
 namespace sdk {
     struct cache;
     class ga_user_pubkeys;
-    class io_runner;
     class wamp_transport;
 
     class ga_session final : public session_impl {

src/io_runner.cpp

@@ -8,7 +8,6 @@
 #include "ga_tor.hpp"
 #include "ga_tx.hpp"
 #include "http_client.hpp"
-#include "io_runner.hpp"
 #include "logging.hpp"
 #include "signer.hpp"
 #include "transaction_utils.hpp"
@@ -31,6 +30,17 @@ namespace sdk {
 
     } // namespace
 
+    struct io_context_and_guard {
+        io_context_and_guard()
+            : m_io()
+            , m_work_guard(boost::asio::make_work_guard(m_io))
+        {
+        }
+
+        boost::asio::io_context m_io;
+        boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_work_guard;
+    };
+
     std::shared_ptr<session_impl> session_impl::create(const nlohmann::json& net_params)
     {
         auto defaults = network_parameters::get(net_params.value("name", std::string()));
@@ -44,8 +54,7 @@ namespace sdk {
 
     session_impl::session_impl(network_parameters&& net_params)
         : m_net_params(net_params)
-        , m_io(std::make_unique<io_runner>())
-        , m_strand(std::make_unique<boost::asio::io_context::strand>(m_io->get_io_context()))
+        , m_io(std::make_unique<io_context_and_guard>())
         , m_user_proxy(socksify(m_net_params.get_json().value("proxy", std::string())))
         , m_notification_handler(nullptr)
         , m_notification_context(nullptr)
@@ -55,12 +64,13 @@ namespace sdk {
             // Enable internal tor controller
             m_tor_ctrl = tor_controller::get_shared_ref();
         }
+        m_run_thread = std::thread([this] { m_io->m_io.run(); });
     }
 
     session_impl::~session_impl()
     {
-        no_std_exception_escape([this] { m_io.reset(); }, "session_impl m_io");
-        no_std_exception_escape([this] { m_strand.reset(); }, "session_impl m_strand");
+        no_std_exception_escape([this] { m_io->m_work_guard.reset(); }, "session_impl dtor(1)");
+        no_std_exception_escape([this] { m_run_thread.join(); }, "session_impl dtor(2)");
     }
 
     void session_impl::set_notification_handler(GA_notification_handler handler, void* context)
@@ -123,7 +133,7 @@ namespace sdk {
 
             std::shared_ptr<http_client> client;
             auto&& get = [&] {
-                client = make_http_client(m_io->get_io_context(), ssl_ctx.get());
+                client = make_http_client(m_io->m_io, ssl_ctx.get());
                 GDK_RUNTIME_ASSERT(client != nullptr);
 
                 const auto verb = boost::beast::http::string_to_verb(params["method"]);

src/io_runner.hpp

@@ -1,9 +1,8 @@
 #ifndef GDK_SESSION_IMPL_HPP
 #define GDK_SESSION_IMPL_HPP
-#pragma once
 
+#pragma once
 #include <atomic>
-#include <boost/asio/io_context.hpp>
 #include <mutex>
 #include <set>
 #include <thread>
@@ -18,11 +17,11 @@ namespace sdk {
     using unique_pubkeys_and_scripts_t = std::set<pubkey_and_script_t>;
 
     class ga_pubkeys;
-    class io_runner;
     class user_pubkeys;
     class signer;
     class Tx;
     struct tor_controller;
+    struct io_context_and_guard;
 
     class session_impl {
     public:
@@ -259,9 +258,9 @@ namespace sdk {
 
         // Immutable upon construction
         const network_parameters m_net_params;
-        std::unique_ptr<io_runner> m_io;
-        std::unique_ptr<boost::asio::io_context::strand> m_strand;
+        std::unique_ptr<io_context_and_guard> m_io;
 
+        std::thread m_run_thread; // Runs the asio context
         const std::string m_user_proxy;
         std::shared_ptr<tor_controller> m_tor_ctrl;
 

src/session_impl.cpp

@@ -1,5 +1,4 @@
 #include <array>
-#include <boost/asio/io_context.hpp>
 #include <cstdio>
 #include <fstream>
 #include <map>
@@ -18,7 +17,6 @@
 #include "boost_wrapper.hpp"
 #include "exception.hpp"
 #include "http_client.hpp"
-#include "io_runner.hpp"
 #include "logging.hpp"
 #include "network_parameters.hpp"
 #include "utils.hpp"
@@ -313,11 +311,11 @@ namespace sdk {
 
     nlohmann::json wamp_cast_json(const autobahn::wamp_call_result& result) { return wamp_cast_json_impl(result); }
 
-    wamp_transport::wamp_transport(const network_parameters& net_params, io_runner& runner,
-        boost::asio::io_context::strand& strand, wamp_transport::notify_fn_t fn, std::string server_prefix)
+    wamp_transport::wamp_transport(
+        const network_parameters& net_params, wamp_transport::notify_fn_t fn, std::string server_prefix)
         : m_net_params(net_params)
-        , m_io(runner)
-        , m_strand(strand)
+        , m_io()
+        , m_work_guard(asio::make_work_guard(m_io))
         , m_server_prefix(std::move(server_prefix))
         , m_server(m_net_params.get_connection_string(m_server_prefix))
         , m_wamp_host_name(websocketpp::uri(m_net_params.gait_wamp_url(m_server_prefix)).get_host())
@@ -333,12 +331,13 @@ namespace sdk {
 
         m_wamp_call_options.set_timeout(std::chrono::seconds(WAMP_CALL_TIMEOUT_SECS));
 
+        m_run_thread = std::thread([this] { m_io.run(); });
         m_reconnect_thread = std::thread([this] { reconnect_handler(); });
 
         if (!m_net_params.is_tls_connection(m_server_prefix)) {
             m_client = std::make_unique<client>();
             m_client->set_pong_timeout_handler(std::bind(&wamp_transport::heartbeat_timeout_cb, this, _1, _2));
-            m_client->init_asio(&m_io.get_io_context());
+            m_client->init_asio(&m_io);
             return;
         }
 
@@ -348,13 +347,15 @@ namespace sdk {
             return tls_init(m_wamp_host_name, m_net_params.gait_wamp_cert_roots(), m_net_params.gait_wamp_cert_pins(),
                 m_net_params.cert_expiry_threshold());
         });
-        m_client_tls->init_asio(&m_io.get_io_context());
+        m_client_tls->init_asio(&m_io);
     }
 
     wamp_transport::~wamp_transport()
     {
         no_std_exception_escape([this] { change_state_to(state_t::exited, std::string(), false); }, "wamp dtor(1)");
         no_std_exception_escape([this] { m_reconnect_thread.join(); }, "wamp dtor(2)");
+        no_std_exception_escape([this] { m_work_guard.reset(); }, "wamp dtor(2)");
+        no_std_exception_escape([this] { m_run_thread.join(); }, "wamp dtor(3)");
     }
 
     void wamp_transport::connect(const std::string& proxy) { change_state_to(state_t::connected, proxy, true); }
@@ -492,7 +493,7 @@ namespace sdk {
     {
         const bool is_tls = m_net_params.is_tls_connection(m_server_prefix);
         const bool is_debug = gdk_config()["log_level"] == "debug";
-        const auto& executor = m_io.get_io_context().get_executor();
+        const auto& executor = m_io.get_executor();
 
         // The last failure number that we handled
         auto last_handled_failure_count = m_failure_count.load();
@@ -580,7 +581,7 @@ namespace sdk {
                 } else {
                     t = std::make_shared<transport>(*m_client, m_server, proxy, is_debug);
                 }
-                s = std::make_shared<autobahn::wamp_session>(m_io.get_io_context(), is_debug);
+                s = std::make_shared<autobahn::wamp_session>(m_io, is_debug);
                 t->attach(std::static_pointer_cast<autobahn::wamp_transport_handler>(s));
                 bool failed = false;
                 if (no_std_exception_escape(

src/session_impl.hpp

@@ -2,7 +2,6 @@
 #define GDK_WAMP_TRANSPORT_HPP
 #pragma once
 
-#include <boost/asio/io_context.hpp>
 #include <nlohmann/json_fwd.hpp>
 #include <optional>
 #include <string>
@@ -17,7 +16,6 @@ namespace sdk {
     class exponential_backoff;
     struct websocketpp_gdk_config;
     struct websocketpp_gdk_tls_config;
-    class io_runner;
 
     nlohmann::json wamp_cast_json(const autobahn::wamp_event& event);
     nlohmann::json wamp_cast_json(const autobahn::wamp_call_result& result);
@@ -46,8 +44,7 @@ namespace sdk {
         using notify_fn_t = std::function<void(nlohmann::json, bool)>;
         using subscribe_fn_t = std::function<void(nlohmann::json)>;
 
-        wamp_transport(const network_parameters& net_params, io_runner& runner, boost::asio::io_context::strand& strand,
-            notify_fn_t fn, std::string server_prefix = {});
+        wamp_transport(const network_parameters& net_params, notify_fn_t fn, std::string server_prefix = {});
         ~wamp_transport();
 
         // Connect the transport. The proxy to use is passed to us as it can
@@ -93,7 +90,7 @@ namespace sdk {
         }
 
         // Post a function to run on the asio executor thread
-        template <typename FN> void post(FN&& fn) { boost::asio::post(m_strand, fn); }
+        template <typename FN> void post(FN&& fn) { boost::asio::post(m_io.get_executor(), fn); }
 
     private:
         using session_ptr = std::shared_ptr<autobahn::wamp_session>;
@@ -125,8 +122,9 @@ namespace sdk {
 
         // These members are immutable after construction
         const network_parameters& m_net_params;
-        io_runner& m_io;
-        boost::asio::io_context::strand& m_strand;
+        boost::asio::io_context m_io;
+        boost::asio::executor_work_guard<boost::asio::io_context::executor_type> m_work_guard;
+        std::thread m_run_thread; // Runs the asio context
         std::thread m_reconnect_thread; // Runs the reconnection logic
         const std::string m_server_prefix;
         const std::string m_server;