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

2 files changed, 203 insertions, 34 deletions

diff --git a/src/script/cpp_api/s_async.cpp b/src/script/cpp_api/s_async.cpp
index dacdcd75a..42a794ceb 100644
--- a/src/script/cpp_api/s_async.cpp
+++ b/src/script/cpp_api/s_async.cpp
@@ -21,9 +21,9 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #include <cstdlib>
 
 extern "C" {
-#include "lua.h"
-#include "lauxlib.h"
-#include "lualib.h"
+#include <lua.h>
+#include <lauxlib.h>
+#include <lualib.h>
 }
 
 #include "server.h"
@@ -32,6 +32,7 @@ extern "C" {
 #include "filesys.h"
 #include "porting.h"
 #include "common/c_internal.h"
+#include "common/c_packer.h"
 #include "lua_api/l_base.h"
 
 /******************************************************************************/
@@ -76,19 +77,34 @@ void AsyncEngine::initialize(unsigned int numEngines)
 {
 	initDone = true;
 
-	for (unsigned int i = 0; i < numEngines; i++) {
-		AsyncWorkerThread *toAdd = new AsyncWorkerThread(this,
-			std::string("AsyncWorker-") + itos(i));
-		workerThreads.push_back(toAdd);
-		toAdd->start();
+	if (numEngines == 0) {
+		// Leave one core for the main thread and one for whatever else
+		autoscaleMaxWorkers = Thread::getNumberOfProcessors();
+		if (autoscaleMaxWorkers >= 2)
+			autoscaleMaxWorkers -= 2;
+		infostream << "AsyncEngine: using at most " << autoscaleMaxWorkers
+			<< " threads with automatic scaling" << std::endl;
+
+		addWorkerThread();
+	} else {
+		for (unsigned int i = 0; i < numEngines; i++)
+			addWorkerThread();
 	}
 }
 
+void AsyncEngine::addWorkerThread()
+{
+	AsyncWorkerThread *toAdd = new AsyncWorkerThread(this,
+		std::string("AsyncWorker-") + itos(workerThreads.size()));
+	workerThreads.push_back(toAdd);
+	toAdd->start();
+}
+
 /******************************************************************************/
 u32 AsyncEngine::queueAsyncJob(std::string &&func, std::string &&params,
 		const std::string &mod_origin)
 {
-	jobQueueMutex.lock();
+	MutexAutoLock autolock(jobQueueMutex);
 	u32 jobId = jobIdCounter++;
 
 	jobQueue.emplace_back();
@@ -99,7 +115,23 @@ u32 AsyncEngine::queueAsyncJob(std::string &&func, std::string &&params,
 	to_add.mod_origin = mod_origin;
 
 	jobQueueCounter.post();
-	jobQueueMutex.unlock();
+	return jobId;
+}
+
+u32 AsyncEngine::queueAsyncJob(std::string &&func, PackedValue *params,
+		const std::string &mod_origin)
+{
+	MutexAutoLock autolock(jobQueueMutex);
+	u32 jobId = jobIdCounter++;
+
+	jobQueue.emplace_back();
+	auto &to_add = jobQueue.back();
+	to_add.id = jobId;
+	to_add.function = std::move(func);
+	to_add.params_ext.reset(params);
+	to_add.mod_origin = mod_origin;
+
+	jobQueueCounter.post();
 	return jobId;
 }
 
@@ -132,6 +164,12 @@ void AsyncEngine::putJobResult(LuaJobInfo &&result)
 /******************************************************************************/
 void AsyncEngine::step(lua_State *L)
 {
+	stepJobResults(L);
+	stepAutoscale();
+}
+
+void AsyncEngine::stepJobResults(lua_State *L)
+{
 	int error_handler = PUSH_ERROR_HANDLER(L);
 	lua_getglobal(L, "core");
 
@@ -148,7 +186,10 @@ void AsyncEngine::step(lua_State *L)
 		luaL_checktype(L, -1, LUA_TFUNCTION);
 
 		lua_pushinteger(L, j.id);
-		lua_pushlstring(L, j.result.data(), j.result.size());
+		if (j.result_ext)
+			script_unpack(L, j.result_ext.get());
+		else
+			lua_pushlstring(L, j.result.data(), j.result.size());
 
 		// Call handler
 		const char *origin = j.mod_origin.empty() ? nullptr : j.mod_origin.c_str();
@@ -161,12 +202,71 @@ void AsyncEngine::step(lua_State *L)
 	lua_pop(L, 2); // Pop core and error handler
 }
 
+void AsyncEngine::stepAutoscale()
+{
+	if (workerThreads.size() >= autoscaleMaxWorkers)
+		return;
+
+	MutexAutoLock autolock(jobQueueMutex);
+
+	// 2) If the timer elapsed, check again
+	if (autoscaleTimer && porting::getTimeMs() >= autoscaleTimer) {
+		autoscaleTimer = 0;
+		// Determine overlap with previous snapshot
+		unsigned int n = 0;
+		for (const auto &it : jobQueue)
+			n += autoscaleSeenJobs.count(it.id);
+		autoscaleSeenJobs.clear();
+		infostream << "AsyncEngine: " << n << " jobs were still waiting after 1s" << std::endl;
+		// Start this many new threads
+		while (workerThreads.size() < autoscaleMaxWorkers && n > 0) {
+			addWorkerThread();
+			n--;
+		}
+		return;
+	}
+
+	// 1) Check if there's anything in the queue
+	if (!autoscaleTimer && !jobQueue.empty()) {
+		// Take a snapshot of all jobs we have seen
+		for (const auto &it : jobQueue)
+			autoscaleSeenJobs.emplace(it.id);
+		// and set a timer for 1 second
+		autoscaleTimer = porting::getTimeMs() + 1000;
+	}
+}
+
 /******************************************************************************/
-void AsyncEngine::prepareEnvironment(lua_State* L, int top)
+bool AsyncEngine::prepareEnvironment(lua_State* L, int top)
 {
 	for (StateInitializer &stateInitializer : stateInitializers) {
 		stateInitializer(L, top);
 	}
+
+	auto *script = ModApiBase::getScriptApiBase(L);
+	try {
+		script->loadMod(Server::getBuiltinLuaPath() + DIR_DELIM + "init.lua",
+			BUILTIN_MOD_NAME);
+	} catch (const ModError &e) {
+		errorstream << "Execution of async base environment failed: "
+			<< e.what() << std::endl;
+		FATAL_ERROR("Execution of async base environment failed");
+	}
+
+	// Load per mod stuff
+	if (server) {
+		const auto &list = server->m_async_init_files;
+		try {
+			for (auto &it : list)
+				script->loadMod(it.second, it.first);
+		} catch (const ModError &e) {
+			errorstream << "Failed to load mod script inside async environment." << std::endl;
+			server->setAsyncFatalError(e.what());
+			return false;
+		}
+	}
+
+	return true;
 }
 
 /******************************************************************************/
@@ -178,15 +278,25 @@ AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher,
 {
 	lua_State *L = getStack();
 
+	if (jobDispatcher->server) {
+		setGameDef(jobDispatcher->server);
+
+		if (g_settings->getBool("secure.enable_security"))
+			initializeSecurity();
+	}
+
 	// Prepare job lua environment
 	lua_getglobal(L, "core");
 	int top = lua_gettop(L);
 
 	// Push builtin initialization type
-	lua_pushstring(L, "async");
+	lua_pushstring(L, jobDispatcher->server ? "async_game" : "async");
 	lua_setglobal(L, "INIT");
 
-	jobDispatcher->prepareEnvironment(L, top);
+	if (!jobDispatcher->prepareEnvironment(L, top)) {
+		// can't throw from here so we're stuck with this
+		isErrored = true;
+	}
 }
 
 /******************************************************************************/
@@ -198,19 +308,20 @@ AsyncWorkerThread::~AsyncWorkerThread()
 /******************************************************************************/
 void* AsyncWorkerThread::run()
 {
-	lua_State *L = getStack();
+	if (isErrored)
+		return nullptr;
 
-	try {
-		loadMod(getServer()->getBuiltinLuaPath() + DIR_DELIM + "init.lua",
-			BUILTIN_MOD_NAME);
-	} catch (const ModError &e) {
-		errorstream << "Execution of async base environment failed: "
-			<< e.what() << std::endl;
-		FATAL_ERROR("Execution of async base environment failed");
-	}
+	lua_State *L = getStack();
 
 	int error_handler = PUSH_ERROR_HANDLER(L);
 
+	auto report_error = [this] (const ModError &e) {
+		if (jobDispatcher->server)
+			jobDispatcher->server->setAsyncFatalError(e.what());
+		else
+			errorstream << e.what() << std::endl;
+	};
+
 	lua_getglobal(L, "core");
 	if (lua_isnil(L, -1)) {
 		FATAL_ERROR("Unable to find core within async environment!");
@@ -223,6 +334,8 @@ void* AsyncWorkerThread::run()
 		if (!jobDispatcher->getJob(&j) || stopRequested())
 			continue;
 
+		const bool use_ext = !!j.params_ext;
+
 		lua_getfield(L, -1, "job_processor");
 		if (lua_isnil(L, -1))
 			FATAL_ERROR("Unable to get async job processor!");
@@ -232,7 +345,10 @@ void* AsyncWorkerThread::run()
 			errorstream << "ASYNC WORKER: Unable to deserialize function" << std::endl;
 			lua_pushnil(L);
 		}
-		lua_pushlstring(L, j.params.data(), j.params.size());
+		if (use_ext)
+			script_unpack(L, j.params_ext.get());
+		else
+			lua_pushlstring(L, j.params.data(), j.params.size());
 
 		// Call it
 		setOriginDirect(j.mod_origin.empty() ? nullptr : j.mod_origin.c_str());
@@ -241,19 +357,28 @@ void* AsyncWorkerThread::run()
 			try {
 				scriptError(result, "<async>");
 			} catch (const ModError &e) {
-				errorstream << e.what() << std::endl;
+				report_error(e);
 			}
 		} else {
 			// Fetch result
-			size_t length;
-			const char *retval = lua_tolstring(L, -1, &length);
-			j.result.assign(retval, length);
+			if (use_ext) {
+				try {
+					j.result_ext.reset(script_pack(L, -1));
+				} catch (const ModError &e) {
+					report_error(e);
+					result = LUA_ERRERR;
+				}
+			} else {
+				size_t length;
+				const char *retval = lua_tolstring(L, -1, &length);
+				j.result.assign(retval, length);
+			}
 		}
 
 		lua_pop(L, 1);  // Pop retval
 
 		// Put job result
-		if (!j.result.empty())
+		if (result == 0)
 			jobDispatcher->putJobResult(std::move(j));
 	}
 
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;