aboutsummaryrefslogtreecommitdiff
path: root/src/script/cpp_api/s_async.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/script/cpp_api/s_async.cpp')
-rw-r--r--src/script/cpp_api/s_async.cpp183
1 files changed, 154 insertions, 29 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));
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-10 16:45:07 +0000