Merge pull request #182 from Bottinator22/scriptablethreads

Scriptable threads

6 files changed, 341 insertions, 2 deletions

diff --git a/doc/lua/openstarbound/threads.md b/doc/lua/openstarbound/threads.md
new file mode 100644
index 0000000..4560b89
--- /dev/null
+++ b/doc/lua/openstarbound/threads.md
@@ -0,0 +1,66 @@
+
+# Threads
+
+The new threads table is accessible from every script and allows creating, communicating with, and destroying scriptable threads.
+Scriptable threads are also automatically destroyed when their parent script component is destroyed.
+
+---
+
+#### `String` threads.create(`Json` parameters)
+
+Creates a thread using the given parameters.
+Here's an example that uses all available parameters:
+```
+threads.create({
+    name="example", -- this is the thread name you'll use to index the thread
+    scripts={
+        main={"/scripts/examplethread.lua"}, -- a list of scripts for each context, similarly to how other scripts work
+        other={"/scripts/examplesecondthreadscript.lua"}, -- threads can have multiple contexts
+    },
+    instructionLimit=100000000, -- optional, threads are allowed to change their own instruction limit (as they have nothing else to block if stuck)
+    tickRate=60, -- optional, how many ticks per second the thread runs at, defaults to 60 but can be any number
+    updateMeasureWindow=0.5, -- optional, defaults to 0.5, changing this is unnecessary unless you really care about an accurate tickrate for some reason
+    someParameter="scrungus" -- parameters for the scripts, all parameters are accessible using config.getParameter in the scripts
+}),
+```
+Returns the thread's name.
+
+---
+
+#### `void` threads.setPause(`String` name, `bool` paused)
+
+Pauses or unpauses a thread.
+
+---
+
+#### `void` threads.stop(`String` name)
+
+Stops and destroys a thread.
+
+---
+
+#### `RpcPromise<Json>` threads.sendMessage(`String` threadName, `String` messageName, [`LuaValue` args...])
+
+Sends a message to the given thread. Note that the return value from this is currently the only way to get data from the thread.
+
+---
+
+Threads have simple updateable scripts with access to only a few tables.
+They include:
+ - the basic tables all scripts have access to (including `threads`)
+ - `updateablescript` bindings
+ - `message`
+ - `config`
+ - `thread`
+ 
+---
+The `thread` table has only a single function.
+
+#### `void` thread.stop()
+
+Stops the thread.
+This does not destroy the thread; the parent script still has to stop the thread itself to destroy it, so avoid using this too much as it can cause memory leaks.
+
+---
+
+
diff --git a/source/game/CMakeLists.txt b/source/game/CMakeLists.txt
index 4825d31..11e0024 100644
--- a/source/game/CMakeLists.txt
+++ b/source/game/CMakeLists.txt
@@ -254,6 +254,7 @@ SET (star_game_HEADERS
     scripting/StarTeamClientLuaBindings.hpp
     scripting/StarWorldLuaBindings.hpp
     scripting/StarUniverseServerLuaBindings.hpp
+    scripting/StarScriptableThread.hpp
 
     terrain/StarCacheSelector.hpp
     terrain/StarConstantSelector.hpp
@@ -494,6 +495,7 @@ SET (star_game_SOURCES
     scripting/StarTeamClientLuaBindings.cpp
     scripting/StarWorldLuaBindings.cpp
     scripting/StarUniverseServerLuaBindings.cpp
+    scripting/StarScriptableThread.cpp
 
     terrain/StarCacheSelector.cpp
     terrain/StarConstantSelector.cpp
@@ -514,4 +516,4 @@ ADD_LIBRARY (star_game OBJECT ${star_game_SOURCES} ${star_game_HEADERS})
 
 IF(STAR_PRECOMPILED_HEADERS)
   TARGET_PRECOMPILE_HEADERS (star_game REUSE_FROM star_core)
-ENDIF()
\ No newline at end of file
+ENDIF()
diff --git a/source/game/scripting/StarLuaComponents.cpp b/source/game/scripting/StarLuaComponents.cpp
index 9a1b1c4..7a6db8b 100644
--- a/source/game/scripting/StarLuaComponents.cpp
+++ b/source/game/scripting/StarLuaComponents.cpp
@@ -1,16 +1,22 @@
 #include "StarLuaComponents.hpp"
 #include "StarUtilityLuaBindings.hpp"
 #include "StarRootLuaBindings.hpp"
+#include "StarScriptableThread.hpp"
+#include "StarRpcThreadPromise.hpp"
+#include "StarLuaGameConverters.hpp"
 
 namespace Star {
 
 LuaBaseComponent::LuaBaseComponent() {
   addCallbacks("sb", LuaBindings::makeUtilityCallbacks());
   addCallbacks("root", LuaBindings::makeRootCallbacks());
+  addCallbacks("threads", makeThreadsCallbacks());
   setAutoReInit(true);
 }
 
-LuaBaseComponent::~LuaBaseComponent() {}
+LuaBaseComponent::~LuaBaseComponent() {
+  m_threads.clear();
+}
 
 StringList const& LuaBaseComponent::scripts() const {
   return m_scripts;
@@ -109,6 +115,9 @@ void LuaBaseComponent::uninit() {
     contextShutdown();
     m_context.reset();
   }
+  for (auto p : m_threads) {
+    p.second->stop();
+  }
 
   m_error.reset();
 }
@@ -152,4 +161,36 @@ bool LuaBaseComponent::checkInitialization() {
   return initialized();
 }
 
+LuaCallbacks LuaBaseComponent::makeThreadsCallbacks() {
+  LuaCallbacks callbacks;
+  
+  callbacks.registerCallback("create", [this](Json parameters) {
+    auto name = parameters.getString("name");
+    if (m_threads.contains(name)) {
+      m_threads.get(name)->stop();
+      m_threads.remove(name);
+    }
+    auto thread = make_shared<ScriptableThread>(parameters);
+    thread->setPause(false);
+    thread->start();
+    m_threads.set(name,thread);
+    return name;
+  });
+  callbacks.registerCallback("setPause", [this](String const& threadName, bool paused) {
+      m_threads.get(threadName)->setPause(paused);
+  });
+  callbacks.registerCallback("stop", [this](String const& threadName) {
+      m_threads.get(threadName)->stop();
+      m_threads.remove(threadName);
+  });
+  callbacks.registerCallback("sendMessage", [this](String const& threadName, String const& message, LuaVariadic<Json> args) {
+    auto pair = RpcThreadPromise<Json>::createPair();
+    RecursiveMutexLocker locker(m_threadLock);
+    m_threads.get(threadName)->passMessage({ message, args, pair.second });
+    return pair.first;
+  });
+  
+  return callbacks;
+}
+
 }
diff --git a/source/game/scripting/StarLuaComponents.hpp b/source/game/scripting/StarLuaComponents.hpp
index e3d1d20..b6c19ac 100644
--- a/source/game/scripting/StarLuaComponents.hpp
+++ b/source/game/scripting/StarLuaComponents.hpp
@@ -10,6 +10,8 @@ namespace Star {
 
 STAR_EXCEPTION(LuaComponentException, LuaException);
 
+STAR_CLASS(ScriptableThread);
+
 // Basic lua component that can be initialized (takes and then owns a script
 // context, calls the script context's init function) and uninitialized
 // (releases the context, calls the context 'uninit' function).
@@ -94,12 +96,17 @@ protected:
   bool checkInitialization();
 
 private:
+  LuaCallbacks makeThreadsCallbacks();
+  
   StringList m_scripts;
   StringMap<LuaCallbacks> m_callbacks;
   LuaRootPtr m_luaRoot;
   TrackerListenerPtr m_reloadTracker;
   Maybe<LuaContext> m_context;
   Maybe<String> m_error;
+  
+  StringMap<shared_ptr<ScriptableThread>> m_threads;
+  mutable RecursiveMutex m_threadLock;
 };
 
 // Wraps a basic lua component to add a persistent storage table translated
diff --git a/source/game/scripting/StarScriptableThread.cpp b/source/game/scripting/StarScriptableThread.cpp
new file mode 100644
index 0000000..cbd32ad
--- /dev/null
+++ b/source/game/scripting/StarScriptableThread.cpp
@@ -0,0 +1,152 @@
+#include "StarScriptableThread.hpp"
+#include "StarLuaRoot.hpp"
+#include "StarLuaComponents.hpp"
+#include "StarConfigLuaBindings.hpp"
+#include "StarTickRateMonitor.hpp"
+#include "StarNpc.hpp"
+#include "StarRoot.hpp"
+#include "StarJsonExtra.hpp"
+#include "StarLogging.hpp"
+#include "StarAssets.hpp"
+
+namespace Star {
+
+ScriptableThread::ScriptableThread(Json parameters)
+  : Thread("ScriptableThread: " + parameters.getString("name")), // TODO
+    m_stop(false),
+    m_errorOccurred(false),
+    m_shouldExpire(true),
+    m_parameters(std::move(parameters)) {
+      m_luaRoot = make_shared<LuaRoot>();
+      m_name = m_parameters.getString("name");
+      
+      m_timestep = 1.0f / m_parameters.getFloat("tickRate",60.0f);
+      
+      // since thread's not blocking anything important, allow modifying the instruction limit
+      if (auto instructionLimit = m_parameters.optUInt("instructionLimit"))
+        m_luaRoot->luaEngine().setInstructionLimit(instructionLimit.value());
+        
+      m_luaRoot->addCallbacks("thread", makeThreadCallbacks());
+      m_luaRoot->addCallbacks(
+          "config", LuaBindings::makeConfigCallbacks(bind(&ScriptableThread::configValue, this, _1, _2)));
+      
+      for (auto& p : m_parameters.getObject("scripts")) {
+        auto scriptComponent = make_shared<ScriptComponent>();
+        scriptComponent->setLuaRoot(m_luaRoot);
+        scriptComponent->setScripts(jsonToStringList(p.second.toArray()));
+
+        m_scriptContexts.set(p.first, scriptComponent);
+        scriptComponent->init();
+      }
+}
+
+ScriptableThread::~ScriptableThread() {
+  m_stop = true;
+
+  m_scriptContexts.clear();
+  
+  join();
+}
+
+void ScriptableThread::start() {
+  m_stop = false;
+  m_errorOccurred = false;
+  Thread::start();
+}
+
+void ScriptableThread::stop() {
+  m_stop = true;
+  Thread::join();
+}
+
+void ScriptableThread::setPause(bool pause) {
+  m_pause = pause;
+}
+
+bool ScriptableThread::errorOccurred() {
+  return m_errorOccurred;
+}
+
+bool ScriptableThread::shouldExpire() {
+  return m_shouldExpire;
+}
+
+void ScriptableThread::passMessage(Message&& message) {
+  RecursiveMutexLocker locker(m_messageMutex);
+  m_messages.append(std::move(message));
+}
+
+void ScriptableThread::run() {
+  try {
+    auto& root = Root::singleton();
+    
+    double updateMeasureWindow = m_parameters.getDouble("updateMeasureWindow",0.5);
+    TickRateApproacher tickApproacher(1.0f / m_timestep, updateMeasureWindow);
+
+    while (!m_stop && !m_errorOccurred) {
+      LogMap::set(strf("lua_{}_update", m_name), strf("{:4.2f}Hz", tickApproacher.rate()));
+
+      update();
+      tickApproacher.setTargetTickRate(1.0f / m_timestep);
+      tickApproacher.tick();
+
+      double spareTime = tickApproacher.spareTime();
+
+      int64_t spareMilliseconds = floor(spareTime * 1000);
+      if (spareMilliseconds > 0)
+        Thread::sleepPrecise(spareMilliseconds);
+    }
+  } catch (std::exception const& e) {
+    Logger::error("ScriptableThread exception caught: {}", outputException(e, true));
+    m_errorOccurred = true;
+  }
+  for (auto& p : m_scriptContexts)
+    p.second->uninit();
+}
+
+Maybe<Json> ScriptableThread::receiveMessage(String const& message, JsonArray const& args) {
+  Maybe<Json> result;
+  for (auto& p : m_scriptContexts) {
+    result = p.second->handleMessage(message, true, args);
+    if (result)
+      break;
+  }
+  return result;
+}
+
+void ScriptableThread::update() {
+  float dt = m_timestep;
+  
+  if (dt > 0.0f && !m_pause) {
+    for (auto& p : m_scriptContexts) {
+      p.second->update(p.second->updateDt(dt));
+    }
+  }
+  
+  List<Message> messages;
+  {
+    RecursiveMutexLocker locker(m_messageMutex);
+    messages = std::move(m_messages);
+  }
+  for (auto& message : messages) {
+    if (auto resp = receiveMessage(message.message, message.args))
+      message.promise.fulfill(*resp);
+    else
+      message.promise.fail("Message not handled by thread");
+  }
+}
+
+LuaCallbacks ScriptableThread::makeThreadCallbacks() {
+  LuaCallbacks callbacks;
+
+  callbacks.registerCallback("stop", [this]() {
+      m_stop = true;
+    });
+
+  return callbacks;
+}
+
+Json ScriptableThread::configValue(String const& name, Json def) const {
+  return m_parameters.get(name, std::move(def));
+}
+}
diff --git a/source/game/scripting/StarScriptableThread.hpp b/source/game/scripting/StarScriptableThread.hpp
new file mode 100644
index 0000000..65f007e
--- /dev/null
+++ b/source/game/scripting/StarScriptableThread.hpp
@@ -0,0 +1,71 @@
+#pragma once
+
+#include "StarThread.hpp"
+#include "StarLuaRoot.hpp"
+#include "StarLuaComponents.hpp"
+#include "StarRpcThreadPromise.hpp"
+
+namespace Star {
+
+STAR_CLASS(ScriptableThread);
+
+// Runs a Lua in a separate thread and guards exceptions that occur in
+// it.  All methods are designed to not throw exceptions, but will instead log
+// the error and trigger the ScriptableThread error state.
+class ScriptableThread : public Thread {
+public:
+  struct Message {
+    String message;
+    JsonArray args;
+    RpcThreadPromiseKeeper<Json> promise;
+  };
+
+  typedef LuaMessageHandlingComponent<LuaUpdatableComponent<LuaBaseComponent>> ScriptComponent;
+  typedef shared_ptr<ScriptComponent> ScriptComponentPtr;
+
+  ScriptableThread(Json parameters);
+  ~ScriptableThread();
+
+  void start();
+  // Signals the ScriptableThread to stop and then joins it
+  void stop();
+  void setPause(bool pause);
+
+  // An exception occurred and the
+  // ScriptableThread has stopped running.
+  bool errorOccurred();
+  bool shouldExpire();
+
+  // 
+  void passMessage(Message&& message);
+
+protected:
+  virtual void run();
+
+private:
+  void update();
+  Maybe<Json> receiveMessage(String const& message, JsonArray const& args);
+
+  mutable RecursiveMutex m_mutex;
+  
+  LuaRootPtr m_luaRoot;
+  StringMap<ScriptComponentPtr> m_scriptContexts;
+  
+  Json m_parameters;
+  String m_name;
+  
+  float m_timestep;
+
+  mutable RecursiveMutex m_messageMutex;
+  List<Message> m_messages;
+
+  atomic<bool> m_stop;
+  atomic<bool> m_pause;
+  mutable atomic<bool> m_errorOccurred;
+  mutable atomic<bool> m_shouldExpire;
+  
+  LuaCallbacks makeThreadCallbacks();
+  Json configValue(String const& name, Json def) const;
+};
+
+}