BUG#37676790 Scripting REST end point returning stalled results

There was a major issue with stalled results on a released context.
i.e. in the case processing ended with a time-out, even the request
was considered complete, because the context remains alive for the
execution of additional requests, the responses associated to the
first request were still generated and consumed by the next request
handled by the context.

To address this, logic was needed to ensure stalled responses after
a context is released are discardedi before putting back the context
in the pool, this change made evident a second problem: the release
logic is synchronous, so the client did not get any response but until
the context was really returned to the pool.

To address this second problem, the a release queue and a release thread
were added to the pool, once a context is done processing a request, it
is added to the queue, and the pool release thread waits for the context
to be available to return it to the pool.

Change-Id: I8973b202265f862f824dc378b07d546235ad266b

Author: rennox

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

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

router/src/jit_executor/src/jit_executor_common_context.cc

@@ -187,8 +187,8 @@ void CommonContext::life_cycle_thread() {
   m_init_condition.notify_one();
 
   // Now waits for the finalization indication
-  {
-    std::unique_lock<std::mutex> lock(m_mutex);
+  if (m_initialized) {
+    std::unique_lock<std::mutex> lock(m_finish_mutex);
     m_finish_condition.wait(lock, [this]() { return m_terminated; });
   }
 

router/src/jit_executor/src/jit_executor_context_pool.cc

@@ -31,6 +31,7 @@
 #include <string>
 #include <vector>
 
+#include "include/my_thread.h"
 #include "jit_executor_common_context.h"
 #include "jit_executor_javascript_context.h"
 
@@ -40,10 +41,12 @@ ContextPool::ContextPool(size_t size, CommonContext *common_context)
     : m_common_context{common_context} {
   m_pool = std::make_unique<Pool<IContext *>>(
       size,
-      [this]() -> JavaScriptContext * {
-        auto context = std::make_unique<JavaScriptContext>(m_common_context);
-        if (context->got_initialization_error()) {
-          return nullptr;
+      [this](size_t id) -> JavaScriptContext * {
+        auto context =
+            std::make_unique<JavaScriptContext>(id, m_common_context);
+        if (!context->started()) {
+          // The factory function should throw runtime exception if fails
+          throw std::runtime_error("Failed initializing JavaScriptContext");
         }
 
         return context.release();
@@ -53,21 +56,49 @@ ContextPool::ContextPool(size_t size, CommonContext *common_context)
 
 ContextPool::~ContextPool() { teardown(); }
 
+void ContextPool::teardown() {
+  m_pool->teardown();
+
+  // Tear down the releaser thread
+  release(nullptr);
+
+  if (m_release_thread) {
+    m_release_thread->join();
+    m_release_thread.reset();
+  }
+}
+
 std::shared_ptr<PooledContextHandle> ContextPool::get_context() {
   auto ctx = m_pool->get();
 
   if (ctx) {
+    if (!m_release_thread) {
+      m_release_thread =
+          std::make_unique<std::thread>(&ContextPool::release_thread, this);
+    }
+
     return std::make_shared<PooledContextHandle>(this, ctx);
   }
 
   return {};
 }
 
-void ContextPool::release(IContext *ctx) {
-  if (ctx->got_resources_error()) {
-    m_pool->on_resources_error(ctx);
-  } else {
-    m_pool->release(ctx);
+void ContextPool::release(IContext *ctx) { m_release_queue.push(ctx); }
+
+void ContextPool::release_thread() {
+  my_thread_self_setname("Jit-CtxDispose");
+  while (true) {
+    auto ctx = m_release_queue.pop();
+    if (ctx) {
+      if (ctx->wait_for_idle()) {
+        m_pool->release(ctx);
+      } else {
+        m_pool->discard(ctx, true);
+      }
+    } else {
+      // nullptr arrived, meaning we are done
+      break;
+    }
   }
 }
 

router/src/jit_executor/src/jit_executor_context_pool.h

@@ -33,22 +33,27 @@
 #include <mutex>
 #include <stdexcept>
 #include <string>
+#include <thread>
 #include <vector>
 
+#include "mysql/harness/logging/logging.h"
+#include "mysql/harness/mpsc_queue.h"
 #include "mysqlrouter/jit_executor_context.h"
 #include "mysqlrouter/jit_executor_context_handle.h"
 #include "mysqlrouter/jit_executor_value.h"
 #include "mysqlrouter/polyglot_file_system.h"
 
 namespace jit_executor {
 
+IMPORT_LOG_FUNCTIONS()
+
 /**
  * Generic implementation of a pool
  */
 template <class T>
 class Pool {
  public:
-  explicit Pool(size_t size, const std::function<T()> &factory,
+  explicit Pool(size_t size, const std::function<T(size_t)> &factory,
                 const std::function<void(T)> &destructor = {})
       : m_pool_size{size},
         m_item_factory{factory},
@@ -84,7 +89,7 @@ class Pool {
     }
 
     try {
-      T item = m_item_factory();
+      T item = m_item_factory(m_created_items);
       increase_active_items();
       return item;
     } catch (const std::runtime_error &err) {
@@ -99,15 +104,12 @@ class Pool {
 
       if (!m_teardown && m_items.size() < m_pool_size) {
         m_items.push_back(ctx);
-        m_item_availability.notify_all();
+        m_item_availability.notify_one();
         return;
       }
     }
 
-    if (m_item_destructor) {
-      m_item_destructor(ctx);
-    }
-    decrease_active_items();
+    discard(ctx, false);
   }
 
   void teardown() {
@@ -120,10 +122,7 @@ class Pool {
       auto item = m_items.front();
       m_items.pop_front();
 
-      if (m_item_destructor) {
-        m_item_destructor(item);
-      }
-      decrease_active_items();
+      discard(item, false);
     }
 
     // Waits until all the contexts created by the pool get released
@@ -139,18 +138,24 @@ class Pool {
   /**
    * Discards the affected context and turns ON contention mode for the pool
    */
-  void on_resources_error(T ctx) {
+  void discard(T ctx, bool set_contention_mode) {
+    decrease_active_items(set_contention_mode);
+
     if (m_item_destructor) {
-      m_item_destructor(ctx);
+      try {
+        m_item_destructor(ctx);
+      } catch (const std::exception &e) {
+        log_error("%s", e.what());
+      }
     }
-    decrease_active_items(true);
   }
 
  private:
   void increase_active_items() {
     {
       std::scoped_lock lock(m_mutex);
       m_active_items++;
+      m_created_items++;
     }
   }
 
@@ -171,9 +176,10 @@ class Pool {
   bool m_teardown = false;
   size_t m_pool_size;
   std::deque<T> m_items;
-  std::function<T()> m_item_factory;
+  std::function<T(size_t id)> m_item_factory;
   std::function<void(T)> m_item_destructor;
   size_t m_active_items = 0;
+  size_t m_created_items = 0;
   bool m_contention_mode = false;
 };
 
@@ -187,11 +193,15 @@ class ContextPool final {
 
   std::shared_ptr<PooledContextHandle> get_context();
   void release(IContext *ctx);
-  void teardown() { m_pool->teardown(); }
+  void teardown();
 
  private:
+  void release_thread();
+
   CommonContext *m_common_context;
   std::unique_ptr<Pool<IContext *>> m_pool;
+  mysql_harness::WaitingMPSCQueue<IContext *> m_release_queue;
+  std::unique_ptr<std::thread> m_release_thread;
 };
 
 /**

router/src/jit_executor/src/jit_executor_debug_context_handle.cc

@@ -35,7 +35,7 @@ DebugContextHandle::DebugContextHandle(const std::string &debug_port,
                                        CommonContext *common_context) {
   assert(!debug_port.empty());
   assert(common_context);
-  m_ctx = std::make_unique<JavaScriptContext>(common_context, debug_port);
+  m_ctx = std::make_unique<JavaScriptContext>(0, common_context, debug_port);
 }
 
 }  // namespace jit_executor