src/script/cpp_api/s_async.cpp

   1 /*
   2 Minetest
   3 Copyright (C) 2013 sapier, <sapier AT gmx DOT net>
   4
   5 This program is free software; you can redistribute it and/or modify
   6 it under the terms of the GNU Lesser General Public License as published by
   7 the Free Software Foundation; either version 2.1 of the License, or
   8 (at your option) any later version.
   9
  10 This program is distributed in the hope that it will be useful,
  11 but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 GNU Lesser General Public License for more details.
  14
  15 You should have received a copy of the GNU Lesser General Public License along
  16 with this program; if not, write to the Free Software Foundation, Inc.,
  17 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18 */
  19
  20 #include <stdio.h>
  21 #include <stdlib.h>
  22
  23 extern "C" {
  24 #include "lua.h"
  25 #include "lauxlib.h"
  26 #include "lualib.h"
  27 }
  28
  29 #include "server.h"
  30 #include "s_async.h"
  31 #include "log.h"
  32 #include "filesys.h"
  33 #include "porting.h"
  34 #include "common/c_internal.h"
  35
  36 /******************************************************************************/
  37 AsyncEngine::AsyncEngine() :
  38         initDone(false),
  39         jobIdCounter(0)
  40 {
  41 }
  42
  43 /******************************************************************************/
  44 AsyncEngine::~AsyncEngine()
  45 {
  46
  47         // Request all threads to stop
  48         for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
  49                         it != workerThreads.end(); it++) {
  50                 (*it)->stop();
  51         }
  52
  53
  54         // Wake up all threads
  55         for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
  56                         it != workerThreads.end(); it++) {
  57                 jobQueueCounter.post();
  58         }
  59
  60         // Wait for threads to finish
  61         for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
  62                         it != workerThreads.end(); it++) {
  63                 (*it)->wait();
  64         }
  65
  66         // Force kill all threads
  67         for (std::vector<AsyncWorkerThread *>::iterator it = workerThreads.begin();
  68                         it != workerThreads.end(); it++) {
  69                 delete *it;
  70         }
  71
  72         jobQueueMutex.lock();
  73         jobQueue.clear();
  74         jobQueueMutex.unlock();
  75         workerThreads.clear();
  76 }
  77
  78 /******************************************************************************/
  79 bool AsyncEngine::registerFunction(const char* name, lua_CFunction func)
  80 {
  81         if (initDone) {
  82                 return false;
  83         }
  84
  85         functionList[name] = func;
  86         return true;
  87 }
  88
  89 /******************************************************************************/
  90 void AsyncEngine::initialize(unsigned int numEngines)
  91 {
  92         initDone = true;
  93
  94         for (unsigned int i = 0; i < numEngines; i++) {
  95                 AsyncWorkerThread *toAdd = new AsyncWorkerThread(this,
  96                         std::string("AsyncWorker-") + itos(i));
  97                 workerThreads.push_back(toAdd);
  98                 toAdd->start();
  99         }
 100 }
 101
 102 /******************************************************************************/
 103 unsigned int AsyncEngine::queueAsyncJob(const std::string &func,
 104                 const std::string &params)
 105 {
 106         jobQueueMutex.lock();
 107         LuaJobInfo toAdd;
 108         toAdd.id = jobIdCounter++;
 109         toAdd.serializedFunction = func;
 110         toAdd.serializedParams = params;
 111
 112         jobQueue.push_back(toAdd);
 113
 114         jobQueueCounter.post();
 115
 116         jobQueueMutex.unlock();
 117
 118         return toAdd.id;
 119 }
 120
 121 /******************************************************************************/
 122 LuaJobInfo AsyncEngine::getJob()
 123 {
 124         jobQueueCounter.wait();
 125         jobQueueMutex.lock();
 126
 127         LuaJobInfo retval;
 128
 129         if (!jobQueue.empty()) {
 130                 retval = jobQueue.front();
 131                 jobQueue.pop_front();
 132                 retval.valid = true;
 133         }
 134         jobQueueMutex.unlock();
 135
 136         return retval;
 137 }
 138
 139 /******************************************************************************/
 140 void AsyncEngine::putJobResult(const LuaJobInfo &result)
 141 {
 142         resultQueueMutex.lock();
 143         resultQueue.push_back(result);
 144         resultQueueMutex.unlock();
 145 }
 146
 147 /******************************************************************************/
 148 void AsyncEngine::step(lua_State *L)
 149 {
 150         int error_handler = PUSH_ERROR_HANDLER(L);
 151         lua_getglobal(L, "core");
 152         resultQueueMutex.lock();
 153         while (!resultQueue.empty()) {
 154                 LuaJobInfo jobDone = resultQueue.front();
 155                 resultQueue.pop_front();
 156
 157                 lua_getfield(L, -1, "async_event_handler");
 158
 159                 if (lua_isnil(L, -1)) {
 160                         FATAL_ERROR("Async event handler does not exist!");
 161                 }
 162
 163                 luaL_checktype(L, -1, LUA_TFUNCTION);
 164
 165                 lua_pushinteger(L, jobDone.id);
 166                 lua_pushlstring(L, jobDone.serializedResult.data(),
 167                                 jobDone.serializedResult.size());
 168
 169                 PCALL_RESL(L, lua_pcall(L, 2, 0, error_handler));
 170         }
 171         resultQueueMutex.unlock();
 172         lua_pop(L, 2); // Pop core and error handler
 173 }
 174
 175 /******************************************************************************/
 176 void AsyncEngine::pushFinishedJobs(lua_State* L) {
 177         // Result Table
 178         MutexAutoLock l(resultQueueMutex);
 179
 180         unsigned int index = 1;
 181         lua_createtable(L, resultQueue.size(), 0);
 182         int top = lua_gettop(L);
 183
 184         while (!resultQueue.empty()) {
 185                 LuaJobInfo jobDone = resultQueue.front();
 186                 resultQueue.pop_front();
 187
 188                 lua_createtable(L, 0, 2);  // Pre-allocate space for two map fields
 189                 int top_lvl2 = lua_gettop(L);
 190
 191                 lua_pushstring(L, "jobid");
 192                 lua_pushnumber(L, jobDone.id);
 193                 lua_settable(L, top_lvl2);
 194
 195                 lua_pushstring(L, "retval");
 196                 lua_pushlstring(L, jobDone.serializedResult.data(),
 197                         jobDone.serializedResult.size());
 198                 lua_settable(L, top_lvl2);
 199
 200                 lua_rawseti(L, top, index++);
 201         }
 202 }
 203
 204 /******************************************************************************/
 205 void AsyncEngine::prepareEnvironment(lua_State* L, int top)
 206 {
 207         for (UNORDERED_MAP<std::string, lua_CFunction>::iterator it = functionList.begin();
 208                         it != functionList.end(); it++) {
 209                 lua_pushstring(L, it->first.c_str());
 210                 lua_pushcfunction(L, it->second);
 211                 lua_settable(L, top);
 212         }
 213 }
 214
 215 /******************************************************************************/
 216 AsyncWorkerThread::AsyncWorkerThread(AsyncEngine* jobDispatcher,
 217                 const std::string &name) :
 218         Thread(name),
 219         ScriptApiBase(),
 220         jobDispatcher(jobDispatcher)
 221 {
 222         lua_State *L = getStack();
 223
 224         // Prepare job lua environment
 225         lua_getglobal(L, "core");
 226         int top = lua_gettop(L);
 227
 228         // Push builtin initialization type
 229         lua_pushstring(L, "async");
 230         lua_setglobal(L, "INIT");
 231
 232         jobDispatcher->prepareEnvironment(L, top);
 233 }
 234
 235 /******************************************************************************/
 236 AsyncWorkerThread::~AsyncWorkerThread()
 237 {
 238         sanity_check(!isRunning());
 239 }
 240
 241 /******************************************************************************/
 242 void* AsyncWorkerThread::run()
 243 {
 244         lua_State *L = getStack();
 245
 246         std::string script = getServer()->getBuiltinLuaPath() + DIR_DELIM + "init.lua";
 247         try {
 248                 loadScript(script);
 249         } catch (const ModError &e) {
 250                 errorstream << "Execution of async base environment failed: "
 251                         << e.what() << std::endl;
 252                 FATAL_ERROR("Execution of async base environment failed");
 253         }
 254
 255         int error_handler = PUSH_ERROR_HANDLER(L);
 256
 257         lua_getglobal(L, "core");
 258         if (lua_isnil(L, -1)) {
 259                 FATAL_ERROR("Unable to find core within async environment!");
 260         }
 261
 262         // Main loop
 263         while (!stopRequested()) {
 264                 // Wait for job
 265                 LuaJobInfo toProcess = jobDispatcher->getJob();
 266
 267                 if (!toProcess.valid || stopRequested()) {
 268                         continue;
 269                 }
 270
 271                 lua_getfield(L, -1, "job_processor");
 272                 if (lua_isnil(L, -1)) {
 273                         FATAL_ERROR("Unable to get async job processor!");
 274                 }
 275
 276                 luaL_checktype(L, -1, LUA_TFUNCTION);
 277
 278                 // Call it
 279                 lua_pushlstring(L,
 280                                 toProcess.serializedFunction.data(),
 281                                 toProcess.serializedFunction.size());
 282                 lua_pushlstring(L,
 283                                 toProcess.serializedParams.data(),
 284                                 toProcess.serializedParams.size());
 285
 286                 int result = lua_pcall(L, 2, 1, error_handler);
 287                 if (result) {
 288                         PCALL_RES(result);
 289                         toProcess.serializedResult = "";
 290                 } else {
 291                         // Fetch result
 292                         size_t length;
 293                         const char *retval = lua_tolstring(L, -1, &length);
 294                         toProcess.serializedResult = std::string(retval, length);
 295                 }
 296
 297                 lua_pop(L, 1);  // Pop retval
 298
 299                 // Put job result
 300                 jobDispatcher->putJobResult(toProcess);
 301         }
 302
 303         lua_pop(L, 2);  // Pop core and error handler
 304
 305         return 0;
 306 }
 307