refactor: Remove the internal thread from the macro debugger

language_server/src/blocking_queue.h

@@ -30,42 +30,55 @@ enum class blocking_queue_termination_policy : bool
 };
 
 template<typename T,
-    blocking_queue_termination_policy termination_policy = blocking_queue_termination_policy::drop_elements>
+    blocking_queue_termination_policy termination_policy = blocking_queue_termination_policy::drop_elements,
+    bool one_reader = true>
 class blocking_queue
 {
+    static constexpr unsigned char terminated_flag = 0x01;
+    static constexpr unsigned char has_elements_flag = 0x02;
+
     std::mutex mutex;
     std::condition_variable cond_var;
     std::deque<T> queue;
-    bool terminated = false;
+    std::atomic<unsigned char> state = 0;
 
 public:
-    void push(T&& t)
+    bool push(T&& t)
     {
         std::unique_lock g(mutex);
-        if (terminated)
-            return;
+
+        if (terminated())
+            return false;
 
         const bool notify = queue.size() == 0;
         queue.push_back(std::move(t));
+        state.fetch_or(has_elements_flag, std::memory_order_relaxed);
 
         g.unlock();
 
         if (notify)
             cond_var.notify_one();
+
+        return true;
     }
-    void push(const T& t)
+
+    bool push(const T& t)
     {
         std::unique_lock g(mutex);
-        if (terminated)
-            return;
+
+        if (terminated())
+            return false;
 
         const bool notify = queue.size() == 0;
         queue.push_back(t);
+        state.fetch_or(has_elements_flag, std::memory_order_relaxed);
 
         g.unlock();
 
         if (notify)
             cond_var.notify_one();
+
+        return true;
     }
 
     std::optional<T> pop()
@@ -74,25 +87,32 @@ class blocking_queue
         constexpr const auto process = blocking_queue_termination_policy::process_elements;
 
         std::unique_lock g(mutex);
-        cond_var.wait(g, [this] { return queue.size() || terminated; });
+        cond_var.wait(g, [this] { return queue.size() || terminated(); });
 
-        if ((termination_policy == drop && terminated) || (termination_policy == process && queue.size() == 0))
+        if ((termination_policy == drop && terminated()) || (termination_policy == process && queue.size() == 0))
             return std::nullopt;
 
         std::optional<T> result = std::move(queue.front());
         queue.pop_front();
+        if (queue.empty())
+            state.fetch_and(static_cast<unsigned char>(~has_elements_flag), std::memory_order_relaxed);
 
         return result;
     }
 
     void terminate()
     {
         std::unique_lock g(mutex);
-        terminated = true;
+        state.fetch_or(terminated_flag, std::memory_order_relaxed);
         g.unlock();
 
         cond_var.notify_one();
     }
+
+    bool terminated() const { return state.load(std::memory_order_relaxed) & terminated_flag; }
+    bool empty() const { return !(state.load(std::memory_order_relaxed) & has_elements_flag); }
+
+    bool will_block() const requires one_reader { return state.load(std::memory_order_relaxed) == 0; }
 };
 } // namespace hlasm_plugin::language_server
 

language_server/src/dap/dap_feature.cpp

@@ -356,4 +356,6 @@ void dap_feature::on_pause(const nlohmann::json& request_seq, const nlohmann::js
     response_->respond(request_seq, "pause", nlohmann::json());
 }
 
+bool dap_feature::idle_handler() { return debugger && debugger->analysis_step(); }
+
 } // namespace hlasm_plugin::language_server::dap

language_server/src/dap/dap_feature.h

@@ -65,6 +65,8 @@ class dap_feature : public feature, public hlasm_plugin::parser_library::debuggi
     void on_continue(const nlohmann::json& request_seq, const nlohmann::json& args);
     void on_pause(const nlohmann::json& request_seq, const nlohmann::json& args);
 
+    bool idle_handler();
+
 private:
     // Inherited via feature
     void register_methods(std::map<std::string, method>& methods) override;

language_server/src/dap/dap_server.cpp

@@ -24,7 +24,9 @@ namespace hlasm_plugin::language_server::dap {
 server::server(parser_library::workspace_manager& ws_mngr, telemetry_sink* telemetry_reporter)
     : language_server::server(ws_mngr, telemetry_reporter)
 {
-    features_.push_back(std::make_unique<dap_feature>(ws_mngr_, *this, this));
+    auto dap_f = std::make_unique<dap_feature>(ws_mngr_, *this, this);
+    m_dap_feature = dap_f.get();
+    features_.push_back(std::move(dap_f));
     register_feature_methods();
 }
 
@@ -105,4 +107,7 @@ void server::disconnected()
     exit_notification_received_ = true;
 }
 
+
+bool server::idle_handler() { return m_dap_feature->idle_handler(); }
+
 } // namespace hlasm_plugin::language_server::dap

language_server/src/dap/dap_server.h

@@ -46,8 +46,11 @@ class server final : public hlasm_plugin::language_server::server, public dap_di
 
     void message_received(const nlohmann::json& message) override;
 
+    bool idle_handler();
+
 private:
     std::atomic<uint64_t> last_seq_ = 0;
+    dap_feature* m_dap_feature = nullptr;
 
     void register_methods();
 

language_server/src/dap/dap_session.cpp

@@ -26,13 +26,31 @@ void session::thread_routine()
 {
     try
     {
-        std::atomic<bool> cancel = false;
+        json_channel_adapter channel(msg_unwrapper, msg_wrapper);
+        struct smp_t final : send_message_provider
+        {
+            json_channel_adapter& channel;
+            void reply(const nlohmann::json& result) override { channel.write(result); }
+            explicit smp_t(json_channel_adapter& channel)
+                : channel(channel)
+            {}
+        } smp(channel);
+
         scope_exit indicate_end([this]() { running = false; });
-        request_manager req_mgr(&cancel);
-        scope_exit end_request_manager([&req_mgr]() { req_mgr.end_worker(); });
+
         dap::server server(*ws_mngr, telemetry_reporter);
-        dispatcher dispatcher(json_channel_adapter(msg_unwrapper, msg_wrapper), server, req_mgr);
-        dispatcher.run_server_loop();
+        server.set_send_message_provider(&smp);
+
+        while (!server.is_exit_notification_received())
+        {
+            if (queue.will_read_block() && server.idle_handler())
+                continue;
+
+            auto msg = channel.read();
+            if (!msg.has_value())
+                break;
+            server.message_received(msg.value());
+        }
     }
     catch (const std::exception& ex)
     {

language_server/src/json_queue_channel.h

@@ -32,6 +32,7 @@ class json_queue_channel final : public json_channel
     void write(nlohmann::json&&) override;
 
     void terminate();
+    bool will_read_block() const { return queue.will_block(); }
 };
 } // namespace hlasm_plugin::language_server
 

language_server/test/dap/dap_feature_test.cpp

@@ -76,36 +76,8 @@ struct response_provider_mock : public response_provider
 
     std::vector<response_mock> responses;
     std::vector<notif_mock> notifs;
-    std::atomic<bool> stopped = false;
-    std::atomic<bool> exited = false;
-
-    void wait_for_stopped()
-    {
-        size_t i = 0;
-        while (!stopped)
-        {
-            if (i > 50)
-                throw std::runtime_error("Wait for stopped timeout.");
-            ++i;
-
-            std::this_thread::sleep_for(100ms);
-        }
-        stopped = false;
-    }
-
-    void wait_for_exited()
-    {
-        size_t i = 0;
-        while (!exited)
-        {
-            if (i > 50)
-                throw std::runtime_error("Wait for exited timeout.");
-            ++i;
-
-            std::this_thread::sleep_for(100ms);
-        }
-        exited = false;
-    }
+    bool stopped = false;
+    bool exited = false;
 
 
     void reset()
@@ -150,8 +122,27 @@ struct feature_launch_test : public testing::Test
 
     void wait_for_stopped()
     {
-        resp_provider.wait_for_stopped();
+        for (int i = 0; !resp_provider.stopped; ++i)
+        {
+            if (i >= 1000000)
+                throw std::runtime_error("Wait for stopped timeout.");
+
+            feature.idle_handler();
+        }
         EXPECT_EQ(resp_provider.notifs.size(), 1U);
+        resp_provider.stopped = false;
+    }
+
+    void wait_for_exited()
+    {
+        for (int i = 0; !resp_provider.exited; ++i)
+        {
+            if (i >= 1000000)
+                throw std::runtime_error("Wait for exited timeout.");
+
+            feature.idle_handler();
+        }
+        resp_provider.exited = false;
     }
 
     std::map<std::string, method> methods;
@@ -271,7 +262,7 @@ TEST_F(feature_launch_test, step)
     ASSERT_EQ(r2.args["stackFrames"].size(), 1U);
 
     feature.on_continue("47"_json, nlohmann::json());
-    resp_provider.wait_for_exited();
+    wait_for_exited();
     feature.on_disconnect("48"_json, {});
 }
 
@@ -337,11 +328,13 @@ TEST_F(feature_launch_test, variables)
     check_simple_stack_trace("2"_json, 0);
 
     feature.on_step_in("3"_json, nlohmann::json());
-    resp_provider.wait_for_stopped();
+    wait_for_stopped();
+    resp_provider.reset();
     feature.on_step_in("3"_json, nlohmann::json());
-    resp_provider.wait_for_stopped();
+    wait_for_stopped();
+    resp_provider.reset();
     feature.on_step_in("3"_json, nlohmann::json());
-    resp_provider.wait_for_stopped();
+    wait_for_stopped();
     resp_provider.reset();
 
     feature.on_stack_trace("1"_json, nlohmann::json());
@@ -429,7 +422,7 @@ TEST_F(feature_launch_test, pause)
 
     feature.on_pause("1"_json, {});
 
-    resp_provider.wait_for_stopped();
+    wait_for_stopped();
     std::vector<response_mock> expected_resp = { { "0"_json, "launch", nlohmann::json() },
         { "1"_json, "pause", nlohmann::json() } };
     EXPECT_EQ(resp_provider.responses, expected_resp);

parser_library/include/debugger.h

@@ -95,6 +95,8 @@ class debugger
     stack_frames_t stack_frames() const;
     scopes_t scopes(frame_id_t frame_id) const;
     variables_t variables(var_reference_t var_ref) const;
+
+    bool analysis_step();
 };
 
 } // namespace hlasm_plugin::parser_library::debugging

parser_library/src/analyzer.cpp

@@ -18,6 +18,7 @@
 #include "lsp/lsp_context.h"
 #include "processing/opencode_provider.h"
 #include "processing/preprocessor.h"
+#include "utils/task.h"
 
 using namespace hlasm_plugin::parser_library;
 using namespace hlasm_plugin::parser_library::lexing;
@@ -148,6 +149,14 @@ void analyzer::analyze()
 
 bool analyzer::analyze_step() { return mngr_.step() || (src_proc_.finish(), false); }
 
+
+hlasm_plugin::utils::task analyzer::co_analyze() &
+{
+    co_await mngr_.co_step();
+
+    src_proc_.finish();
+}
+
 void analyzer::collect_diags() const
 {
     collect_diags_from_child(mngr_);

parser_library/src/analyzer.h

@@ -39,6 +39,11 @@
 #include "virtual_file_monitor.h"
 #include "workspaces/parse_lib_provider.h"
 
+
+namespace hlasm_plugin::utils {
+struct task;
+} // namespace hlasm_plugin::utils
+
 namespace hlasm_plugin::parser_library::parsing {
 class hlasmparser_multiline;
 } // namespace hlasm_plugin::parser_library::parsing
@@ -154,6 +159,7 @@ class analyzer : public diagnosable_ctx
 
     void analyze();
     bool analyze_step();
+    utils::task co_analyze() &;
 
     void collect_diags() const override;
     const performance_metrics& get_metrics() const;