Merge branch 'main_loop with single-threaded websocket'

* Branch commit log:
  ase/websocket.cc: use single-threaded POSIX socket + main_loop for WebSockets
	* ase/jsonapi.cc: adjust message() handling for single-threaded WebSockets
  ase/server.cc: fix websocket tear-down crash in TelemetryPlan, keep it alive via shared_ptr

Author: tim-janik

ase/jsonapi.cc

@@ -81,34 +81,16 @@ class JsonapiConnection : public WebSocketConnection, public CustomDataContainer
   {
     JsonapiConnectionP conp = std::dynamic_pointer_cast<JsonapiConnection> (shared_from_this());
     assert_return (conp);
-    struct MessageData { // keep arguments alive, even after __func__ returns
-      ScopedSemaphore sem;
-      String message, reply;
-      JsonapiConnectionP conp;
-      MessageData (JsonapiConnectionP c, const String &m) : message (m), conp (c) {}
-    };
-    auto data = std::make_shared<MessageData> (conp, message);
-    main_jobs += [data, this] () {
-      if (main_loop->has_quit())
-        return;
-      current_message_conection = data->conp;
-      data->reply = this->handle_jsonipc (data->message);
+    String reply;
+    if (!main_loop->has_quit()) {
+      current_message_conection = conp;
+      reply = this->handle_jsonipc (message);
       current_message_conection = nullptr;
-      data->sem.post();
-    };
-    nickname(); // cache socket nickname for use during errors
-    // wait with timeout, checking if main loop has quit to avoid deadlock on shutdown
-    const uint64_t timeout_us = 50 * 1000;
-    do {
-      const int ret = data->sem.wait_for (timeout_us);
-      if (ret == 0)
-        break; // semaphore was posted, job completed successfully
-      if (data->reply.empty() && main_loop->has_quit())
-        data->reply = "{id:0,error:{code:-32601,message:\"Method not found: endpoint shutting down\"}}\n";
-    } while (data->reply.empty());
-    // when queueing asynchronously, we have to use WebSocketConnectionP
-    if (!data->reply.empty())
-      send_text (data->reply);
+    }
+    else
+      reply = "{id:0,error:{code:-32601,message:\"Method not found: endpoint shutting down\"}}\n";
+    if (!reply.empty())
+      send_text (reply);
   }
   String handle_jsonipc (const std::string &message);
   std::vector<JsTrigger> triggers_; // HINT: use unordered_map if this becomes slow

ase/server.cc

@@ -456,7 +456,7 @@ validate_telemetry_segments (const TelemetrySegmentS &segments, size_t *payloadl
 }
 
 ASE_CLASS_DECLS (TelemetryPlan);
-class TelemetryPlan {
+class TelemetryPlan : public std::enable_shared_from_this<TelemetryPlan> {
 public:
   int32               interval_ms_ = -1;
   LoopID              timerid_ = LoopID::INVALID;
@@ -503,7 +503,8 @@ TelemetryPlan::setup (const char *start, size_t payloadlength, const TelemetrySe
     {
       if (timerid_ != LoopID::INVALID)
         main_loop->cancel (timerid_);
-      auto send_telemetry = [this] () { this->send_telemetry(); return true; };
+      auto tplan = shared_from_this();
+      auto send_telemetry = [tplan] () { tplan->send_telemetry(); return true; };
       interval_ms_ = interval_ms;
       timerid_ = interval_ms <= 0 || segments.empty() ? LoopID::INVALID : main_loop->add (send_telemetry, std::chrono::milliseconds (interval_ms));
     }
@@ -524,21 +525,27 @@ TelemetryPlan::setup (const char *start, size_t payloadlength, const TelemetrySe
 void
 TelemetryPlan::send_telemetry ()
 {
+  // Safety check: don't send if telemetry memory or segments are invalid
+  if (!telemem_ || segments_.empty() || payload_.empty())
+    return;
   char *data = &payload_[0];
   size_t datapos = 0;
   for (const auto &seg : segments_)     // offsets and lengths were validated earlier
     {
       memcpy (data + datapos, telemem_ + seg.offset, seg.length);
       datapos += seg.length;
     }
-  send_blob_ (payload_);
+  // send_blob_ handles the case where the connection is closed (returns false)
+  (void) send_blob_ (payload_);
 }
 
 TelemetryPlan::~TelemetryPlan()
 {
   if (timerid_ != LoopID::INVALID)
     {
-      main_loop->cancel (timerid_);
+      // Only cancel if the loop is still running
+      if (!main_loop->has_quit())
+        main_loop->cancel (timerid_);
       timerid_ = LoopID::INVALID;
     }
 }