From e7659883cc6fca343785da2a1af3890ae273abbf Mon Sep 17 00:00:00 2001 From: sfan5 Date: Mon, 2 May 2022 20:55:04 +0200 Subject: Async environment for mods to do concurrent tasks (#11131) --- src/script/cpp_api/s_async.cpp | 183 ++++++++++++++++++++++++++++++++++------- src/script/cpp_api/s_async.h | 54 ++++++++++-- 2 files changed, 203 insertions(+), 34 deletions(-) (limited to 'src/script/cpp_api') 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 extern "C" { -#include "lua.h" -#include "lauxlib.h" -#include "lualib.h" +#include +#include +#include } #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 &¶ms, 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 &¶ms, 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; } @@ -131,6 +163,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, ""); } 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 #include +#include +#include +#include #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 params_ext; // Result of function call (serialized) std::string result; + // Alternative result + std::unique_ptr 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); @@ -94,9 +105,17 @@ public: u32 queueAsyncJob(std::string &&func, std::string &¶ms, 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); @@ -116,19 +135,44 @@ 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 autoscaleSeenJobs; + + // Only set for the server async environment (duh) + Server *server = nullptr; + // Internal store for registred state initializers std::vector stateInitializers; -- cgit v1.2.3