BUG#37661257 assertion in jit-executor at execute

This patch improves the exception handling, specially in the boundaries
between polyglot and C++, where some exception types were not being
handled correctly, leading to crashes on specific events.

Also introduces a better handling not only of memory errors but any
critical resource that may lead to malfunction of the context.

Also removes the on demand garbage collection logic which now is
unnecessary since establishing the RAM limit gives automatic GC
as needed.

Additionally updates the default RAM size to be used to 1GB.

Change-Id: Id5c685cf923357d9b9dad9f17cd3374896e030a5

Author: rennox

router/src/harness/include/mysql/harness/waiting_queue_adaptor.h

@@ -26,6 +26,7 @@
 #ifndef MYSQL_HARNESS_WAITING_QUEUE_ADAPTOR_INCLUDED
 #define MYSQL_HARNESS_WAITING_QUEUE_ADAPTOR_INCLUDED
 
+#include <chrono>
 #include <condition_variable>
 #include <mutex>
 
@@ -78,6 +79,31 @@ class WaitingQueueAdaptor {
     return true;
   }
 
+  /**
+   * Attempt to dequeue an item from a queue for a given period of time.
+   *
+   * @param item dequeued item if queue was not empty within the specified time.
+   * @param timeout max time to wait for the item to be available.
+   *
+   * @returns item
+   * @retval true item dequeued
+   * @retval false queue was empty
+   */
+  template <class Rep, class Period>
+  bool try_pop(value_type &item, std::chrono::duration<Rep, Period> timeout) {
+    {
+      std::unique_lock<std::mutex> lk(dequeueable_cond_mutex_);
+      if (!dequeueable_cond_.wait_for(
+              lk, timeout, [this, &item] { return q_.dequeue(item); })) {
+        return false;
+      }
+    }
+
+    notify_enqueueable();
+
+    return true;
+  }
+
   /**
    * enqueue item into queue.
    *

router/src/jit_executor/include/mysqlrouter/jit_executor_component.h

@@ -77,7 +77,7 @@ class IServiceHandlers {
 
   virtual void release_debug_context() = 0;
 
-  virtual void init() = 0;
+  virtual bool init() = 0;
   virtual void teardown() = 0;
 
   virtual std::chrono::seconds idle_time() const = 0;

router/src/jit_executor/include/mysqlrouter/jit_executor_context.h

@@ -46,7 +46,7 @@ class IContext {
                               const std::vector<shcore::Value> &parameters,
                               int timeout, ResultType result_type,
                               const GlobalCallbacks &global_callbacks) = 0;
-  virtual bool got_memory_error() const = 0;
+  virtual bool got_resources_error() const = 0;
 };
 
 }  // namespace jit_executor

router/src/jit_executor/src/jit_executor_common_context.cc

@@ -27,6 +27,7 @@
 
 #include <iostream>
 
+#include "include/my_thread.h"
 #include "jit_executor_javascript.h"
 #include "mysql/harness/logging/logging.h"
 #include "mysqlrouter/polyglot_file_system.h"
@@ -37,8 +38,7 @@ IMPORT_LOG_FUNCTIONS()
 
 namespace jit_executor {
 
-bool CommonContext::m_fatal_error = false;
-std::string CommonContext::m_fatal_error_description;
+bool CommonContext::m_global_fatal_error = false;
 
 CommonContext::CommonContext(
     const std::shared_ptr<shcore::polyglot::IFile_system> &fs,
@@ -52,9 +52,16 @@ CommonContext::CommonContext(
 
 CommonContext::~CommonContext() {
   // We are done
+  {
+    std::scoped_lock lock(m_mutex);
+    m_terminated = true;
+  }
+
   m_finish_condition.notify_one();
-  m_life_cycle_thread->join();
-  m_life_cycle_thread.reset();
+  if (m_life_cycle_thread) {
+    m_life_cycle_thread->join();
+    m_life_cycle_thread.reset();
+  }
 }
 
 void CommonContext::fatal_error() {
@@ -65,7 +72,7 @@ void CommonContext::fatal_error() {
 
   // In our case, we simply set a flag indicating to log further attempts to use
   // JavaScript contexts
-  m_fatal_error = true;
+  m_global_fatal_error = true;
 }
 
 void CommonContext::flush() {}
@@ -139,8 +146,9 @@ bool CommonContext::start() {
 
   // Now waits for the destruction indication
   {
-    std::unique_lock<std::mutex> lock(m_init_mutex);
-    m_init_condition.wait(lock, [this]() { return m_initialized; });
+    std::unique_lock<std::mutex> lock(m_mutex);
+    m_init_condition.wait(lock,
+                          [this]() { return m_initialized || m_fatal_error; });
   }
 
   return !m_fatal_error;
@@ -155,23 +163,33 @@ bool CommonContext::start() {
  * above condition.
  */
 void CommonContext::life_cycle_thread() {
+  my_thread_self_setname("Jit-Common");
+  std::optional<std::string> init_error;
+
   try {
     initialize(m_isolate_args);
   } catch (const shcore::polyglot::Polyglot_generic_error &error) {
     // This error is not reported by graal but it is a fatal error!
-    m_fatal_error = true;
-    m_fatal_error_description = error.message();
+    init_error = error.message();
   }
 
-  m_initialized = true;
+  {
+    std::scoped_lock lock(m_mutex);
+    if (init_error.has_value()) {
+      m_fatal_error = true;
+      m_fatal_error_description = *init_error;
+    } else {
+      m_initialized = true;
+    }
+  }
 
   // Tell the constructor we are done
   m_init_condition.notify_one();
 
   // Now waits for the finalization indication
   {
-    std::unique_lock<std::mutex> lock(m_finish_mutex);
-    m_finish_condition.wait(lock);
+    std::unique_lock<std::mutex> lock(m_mutex);
+    m_finish_condition.wait(lock, [this]() { return m_terminated; });
   }
 
   finalize();

router/src/jit_executor/src/jit_executor_common_context.h

@@ -92,17 +92,19 @@ class CommonContext : public shcore::polyglot::Polyglot_common_context {
   shcore::Dictionary_t m_globals;
 
   std::unique_ptr<std::thread> m_life_cycle_thread;
-  std::mutex m_init_mutex;
+  std::mutex m_mutex;
   std::condition_variable m_init_condition;
 
   bool m_initialized = false;
+  bool m_terminated = false;
 
   std::mutex m_finish_mutex;
   std::condition_variable m_finish_condition;
 
   // Global fatal error flag to indicate when the VM was ended
-  static bool m_fatal_error;
-  static std::string m_fatal_error_description;
+  static bool m_global_fatal_error;
+  bool m_fatal_error;
+  std::string m_fatal_error_description;
   std::vector<std::string> m_isolate_args;
 };
 

router/src/jit_executor/src/jit_executor_component.cc

@@ -110,6 +110,9 @@ void JitExecutorComponent::update_active_contexts(
   std::unordered_map<std::string, std::shared_ptr<IServiceHandlers>>
       all_context_handlers = std::move(m_service_context_handlers);
 
+  std::unordered_map<std::string, std::shared_ptr<IServiceHandlers>>
+      candidate_context_handlers;
+
   uint64_t total_pool = 0;
   for (const auto &it : all_context_handlers) {
     // Adds the existing context handler to be discarded
@@ -127,7 +130,7 @@ void JitExecutorComponent::update_active_contexts(
       // Creates a new handler from the existing one
       auto source_handler =
           std::dynamic_pointer_cast<ServiceHandlers>(it.second);
-      m_service_context_handlers.emplace(
+      candidate_context_handlers.emplace(
           it.first, std::make_shared<ServiceHandlers>(*source_handler.get()));
     }
   }
@@ -139,20 +142,22 @@ void JitExecutorComponent::update_active_contexts(
 
     total_pool += replacement.second->pool_size();
 
-    m_service_context_handlers.emplace(std::move(replacement));
+    candidate_context_handlers.emplace(std::move(replacement));
   }
 
   // Now updates the memory limit for each active handler and starts it
   if (m_global_config.maximum_ram_size.has_value()) {
     uint64_t mem_per_pool_item = *m_global_config.maximum_ram_size / total_pool;
 
-    for (const auto &it : m_service_context_handlers) {
+    for (const auto &it : candidate_context_handlers) {
       it.second->set_max_heap_size(mem_per_pool_item * it.second->pool_size());
     }
   }
 
-  for (const auto &it : m_service_context_handlers) {
-    it.second->init();
+  for (const auto &it : candidate_context_handlers) {
+    if (it.second->init()) {
+      m_service_context_handlers.emplace(it.first, it.second);
+    }
   }
 }
 
@@ -168,11 +173,14 @@ std::shared_ptr<IContextHandle> JitExecutorComponent::get_context(
         update_active_contexts(
             {service_id, std::make_shared<ServiceHandlers>(config)});
 
-        return m_service_context_handlers.at(service_id)
-            ->get_context(debug_port);
+        it = m_service_context_handlers.find(service_id);
       }
 
-      return it->second->get_context(debug_port);
+      if (it != m_service_context_handlers.end()) {
+        return it->second->get_context(debug_port);
+      } else {
+        throw std::runtime_error("error to go below..., needed?");
+      }
     } catch (const std::runtime_error &) {
       // If failed to create a context, then let's try re-creating the whole
       // pool, if this failed on a brand new pool, then there's nothing else to

router/src/jit_executor/src/jit_executor_context_pool.cc

@@ -41,7 +41,12 @@ ContextPool::ContextPool(size_t size, CommonContext *common_context)
   m_pool = std::make_unique<Pool<IContext *>>(
       size,
       [this]() -> JavaScriptContext * {
-        return std::make_unique<JavaScriptContext>(m_common_context).release();
+        auto context = std::make_unique<JavaScriptContext>(m_common_context);
+        if (context->got_initialization_error()) {
+          return nullptr;
+        }
+
+        return context.release();
       },
       [](IContext *ctx) { delete ctx; });
 }
@@ -59,8 +64,8 @@ std::shared_ptr<PooledContextHandle> ContextPool::get_context() {
 }
 
 void ContextPool::release(IContext *ctx) {
-  if (ctx->got_memory_error()) {
-    m_pool->on_memory_error(ctx);
+  if (ctx->got_resources_error()) {
+    m_pool->on_resources_error(ctx);
   } else {
     m_pool->release(ctx);
   }

router/src/jit_executor/src/jit_executor_context_pool.h

@@ -83,8 +83,14 @@ class Pool {
       }
     }
 
-    m_active_items++;
-    return m_item_factory();
+    try {
+      T item = m_item_factory();
+      increase_active_items();
+      return item;
+    } catch (const std::runtime_error &err) {
+      // An initialization failure would raise this exception
+      return {};
+    }
   }
 
   void release(T ctx) {
@@ -93,17 +99,15 @@ class Pool {
 
       if (!m_teardown && m_items.size() < m_pool_size) {
         m_items.push_back(ctx);
-        m_item_availability.notify_one();
+        m_item_availability.notify_all();
         return;
       }
     }
 
-    m_active_items--;
-    m_teared_down.notify_one();
-    m_item_availability.notify_one();
     if (m_item_destructor) {
       m_item_destructor(ctx);
     }
+    decrease_active_items();
   }
 
   void teardown() {
@@ -116,15 +120,15 @@ class Pool {
       auto item = m_items.front();
       m_items.pop_front();
 
-      m_active_items--;
       if (m_item_destructor) {
         m_item_destructor(item);
       }
+      decrease_active_items();
     }
 
     // Waits until all the contexts created by the pool get released
     std::unique_lock<std::mutex> lock(m_mutex);
-    m_teared_down.wait(lock, [this]() { return m_active_items == 0; });
+    m_item_availability.wait(lock, [this]() { return m_active_items == 0; });
   }
 
   size_t active_items() const {
@@ -135,21 +139,34 @@ class Pool {
   /**
    * Discards the affected context and turns ON contention mode for the pool
    */
-  void on_memory_error(T ctx) {
-    std::scoped_lock lock(m_mutex);
-    m_contention_mode = true;
-
-    m_active_items--;
-    m_teared_down.notify_one();
-    m_item_availability.notify_one();
+  void on_resources_error(T ctx) {
     if (m_item_destructor) {
       m_item_destructor(ctx);
     }
+    decrease_active_items(true);
   }
 
  private:
+  void increase_active_items() {
+    {
+      std::scoped_lock lock(m_mutex);
+      m_active_items++;
+    }
+  }
+
+  void decrease_active_items(bool set_contention_mode = false) {
+    {
+      std::scoped_lock lock(m_mutex);
+      m_active_items--;
+
+      if (set_contention_mode) {
+        m_contention_mode = true;
+      }
+    }
+    m_item_availability.notify_all();
+  }
+
   std::mutex m_mutex;
-  std::condition_variable m_teared_down;
   std::condition_variable m_item_availability;
   bool m_teardown = false;
   size_t m_pool_size;