Async environment for mods to do concurrent tasks (#11131)

1 files changed, 49 insertions, 5 deletions

diff --git a/src/script/cpp_api/s_async.h b/src/script/cpp_api/s_async.h
index 697cb0221..1e34e40ea 100644
--- a/src/script/cpp_api/s_async.h
+++ b/src/script/cpp_api/s_async.h
@@ -21,11 +21,15 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 
 #include <vector>
 #include <deque>
+#include <unordered_set>
+#include <memory>
 
+#include <lua.h>
 #include "threading/semaphore.h"
 #include "threading/thread.h"
-#include "lua.h"
+#include "common/c_packer.h"
 #include "cpp_api/s_base.h"
+#include "cpp_api/s_security.h"
 
 // Forward declarations
 class AsyncEngine;
@@ -42,8 +46,12 @@ struct LuaJobInfo
 	std::string function;
 	// Parameter to be passed to function (serialized)
 	std::string params;
+	// Alternative parameters
+	std::unique_ptr<PackedValue> params_ext;
 	// Result of function call (serialized)
 	std::string result;
+	// Alternative result
+	std::unique_ptr<PackedValue> result_ext;
 	// Name of the mod who invoked this call
 	std::string mod_origin;
 	// JobID used to identify a job and match it to callback
@@ -51,7 +59,8 @@ struct LuaJobInfo
 };
 
 // Asynchronous working environment
-class AsyncWorkerThread : public Thread, virtual public ScriptApiBase {
+class AsyncWorkerThread : public Thread,
+	virtual public ScriptApiBase, public ScriptApiSecurity {
 	friend class AsyncEngine;
 public:
 	virtual ~AsyncWorkerThread();
@@ -63,6 +72,7 @@ protected:
 
 private:
 	AsyncEngine *jobDispatcher = nullptr;
+	bool isErrored = false;
 };
 
 // Asynchornous thread and job management
@@ -71,6 +81,7 @@ class AsyncEngine {
 	typedef void (*StateInitializer)(lua_State *L, int top);
 public:
 	AsyncEngine() = default;
+	AsyncEngine(Server *server) : server(server) {};
 	~AsyncEngine();
 
 	/**
@@ -81,7 +92,7 @@ public:
 
 	/**
 	 * Create async engine tasks and lock function registration
-	 * @param numEngines Number of async threads to be started
+	 * @param numEngines Number of worker threads, 0 for automatic scaling
 	 */
 	void initialize(unsigned int numEngines);
 
@@ -95,8 +106,16 @@ public:
 			const std::string &mod_origin = "");
 
 	/**
+	 * Queue an async job
+	 * @param func Serialized lua function
+	 * @param params Serialized parameters (takes ownership!)
+	 * @return ID of queued job
+	 */
+	u32 queueAsyncJob(std::string &&func, PackedValue *params,
+			const std::string &mod_origin = "");
+
+	/**
 	 * Engine step to process finished jobs
-	 *   the engine step is one way to pass events back, PushFinishedJobs another
 	 * @param L The Lua stack
 	 */
 	void step(lua_State *L);
@@ -117,18 +136,43 @@ protected:
 	void putJobResult(LuaJobInfo &&result);
 
 	/**
+	 * Start an additional worker thread
+	 */
+	void addWorkerThread();
+
+	/**
+	 * Process finished jobs callbacks
+	 */
+	void stepJobResults(lua_State *L);
+
+	/**
+	 * Handle automatic scaling of worker threads
+	 */
+	void stepAutoscale();
+
+	/**
 	 * Initialize environment with current registred functions
 	 *  this function adds all functions registred by registerFunction to the
 	 *  passed lua stack
 	 * @param L Lua stack to initialize
 	 * @param top Stack position
+	 * @return false if a mod error ocurred
 	 */
-	void prepareEnvironment(lua_State* L, int top);
+	bool prepareEnvironment(lua_State* L, int top);
 
 private:
 	// Variable locking the engine against further modification
 	bool initDone = false;
 
+	// Maximum number of worker threads for automatic scaling
+	// 0 if disabled
+	unsigned int autoscaleMaxWorkers = 0;
+	u64 autoscaleTimer = 0;
+	std::unordered_set<u32> autoscaleSeenJobs;
+
+	// Only set for the server async environment (duh)
+	Server *server = nullptr;
+
 	// Internal store for registred state initializers
 	std::vector<StateInitializer> stateInitializers;