nf7/common/luajit_thread.cc

#include "common/luajit_thread.hh"
#include "common/luajit_thread_lambda.hh"
#include "common/luajit_thread_lock.hh"

#include <sstream>
#include <tuple>

#include "common/async_buffer.hh"
#include "common/luajit_obj.hh"


namespace nf7::luajit {

constexpr size_t kInstructionLimit = 10000000;
constexpr size_t kBufferSizeMax    = 64 * 1024 * 1024;


// Pushes a metatable for Thread object, available on global table as 'nf7'.
static void PushMeta(lua_State*) noexcept;

// Pushes Lua object built by a file who implements luajit::Obj interface.
static void GetLuaObjAndPush(
    lua_State* L, const std::shared_ptr<Thread>& th, File& f);


lua_State* Thread::Init(lua_State* L) noexcept {
  assert(state_ == kInitial);

  th_ = lua_newthread(L);
  PushImmEnv(L);
  lua_setfenv(L, -2);
  th_ref_.emplace(ctx_, ljq_, luaL_ref(L, LUA_REGISTRYINDEX));

  state_ = kPaused;
  return th_;
}
void Thread::Resume(lua_State* L, int narg) noexcept {
  std::unique_lock<std::mutex> k(mtx_);

  if (state_ == kAborted) return;
  assert(L      == th_);
  assert(state_ == kPaused);
  (void) L;

  static const auto kHook = [](auto L, auto) {
    luaL_error(L, "reached instruction limit (<=1e7)");
  };
  lua_sethook(th_, kHook, LUA_MASKCOUNT, kInstructionLimit);

  PushGlobalTable(th_);
  PushWeakPtr(th_, weak_from_this());
  PushMeta(th_);
  lua_setmetatable(th_, -2);
  lua_setfield(th_, -2, "nf7");
  lua_pop(th_, 1);

  state_ = kRunning;
  k.unlock();
  active_ = true;
  const auto ret = lua_resume(th_, narg);
  active_ = false;
  k.lock();
  if (state_ == kAborted) return;
  switch (ret) {
  case 0:
    state_ = kFinished;
    break;
  case LUA_YIELD:
    state_ = kPaused;
    break;
  default:
    state_ = kAborted;
  }
  if (!std::exchange(skip_handle_, false)) {
    handler_(*this, th_);
  }
}
void Thread::Abort() noexcept {
  std::unique_lock<std::mutex> k(mtx_);
  state_ = kAborted;
}


static void PushMeta(lua_State* L) noexcept {
  if (luaL_newmetatable(L, Thread::kTypeName)) {
    PushWeakPtrDeleter<Thread>(L);
    lua_setfield(L, -2, "__gc");

    lua_createtable(L, 0, 0);
    {
      // nf7:resolve(path)
      lua_pushcfunction(L, [](auto L) {
        auto th   = Thread::GetPtr(L, 1);
        auto base = th->ctx()->initiator();

        std::string path = luaL_checkstring(L, 2);
        th->env().ExecSub(th->ctx(), [th, L, base, path = std::move(path)]() {
          try {
            th->ExecResume(L, th->env().GetFileOrThrow(base).ResolveOrThrow(path).id());
          } catch (nf7::File::NotFoundException&) {
            th->ExecResume(L, 0);
          }
        });
        th->ExpectYield(L);
        return lua_yield(L, 0);
      });
      lua_setfield(L, -2, "resolve");

      // nf7:query(file_id, interface)
      lua_pushcfunction(L, [](auto L) {
        auto th = Thread::GetPtr(L, 1);

        const auto  id    = luaL_checkinteger(L, 2);
        std::string iface = luaL_checkstring(L, 3);
        th->env().ExecSub(th->ctx(), [th, L, id, iface = std::move(iface)]() {
          try {
            auto& f = th->env().GetFileOrThrow(static_cast<nf7::File::Id>(id));
            if (iface == "buffer") {
              Thread::Lock<nf7::AsyncBuffer>::AcquireAndPush(L, th, f, false);
            } else if (iface == "exbuffer") {
              Thread::Lock<nf7::AsyncBuffer>::AcquireAndPush(L, th, f, true);
            } else if (iface == "lua") {
              GetLuaObjAndPush(L, th, f);
            } else if (iface == "node") {
              Thread::Lambda::CreateAndPush(L, th, f);
            } else {
              throw nf7::Exception {"unknown interface: "+iface};
            }
          } catch (nf7::Exception& e) {
            th->ExecResume(L, nullptr, e.msg());
          }
        });
        th->ExpectYield(L);
        return lua_yield(L, 0);
      });
      lua_setfield(L, -2, "query");

      // nf7:yield(results...)
      lua_pushcfunction(L, [](auto L) {
        auto th = Thread::GetPtr(L, 1);
        th->ExecResume(L);
        th->ExpectYield(L);
        return lua_yield(L, lua_gettop(L)-1);
      });
      lua_setfield(L, -2, "yield");

      // logging functions
      static const auto log_write = [](lua_State* L, nf7::Logger::Level lv) {
        auto th     = Thread::GetPtr(L, 1);
        auto logger = th->logger();
        if (!logger) return luaL_error(L, "logger is not installed on current thread");

        const int n = lua_gettop(L);
        std::stringstream st;
        for (int i = 2; i <= n; ++i) {
          st << lua_tostring(L, i);
        }
        logger->Write({lv, st.str()});
        return 0;
      };
      lua_pushcfunction(L, [](auto L) { return log_write(L, nf7::Logger::kTrace); });
      lua_setfield(L, -2, "trace");
      lua_pushcfunction(L, [](auto L) { return log_write(L, nf7::Logger::kInfo); });
      lua_setfield(L, -2, "info");
      lua_pushcfunction(L, [](auto L) { return log_write(L, nf7::Logger::kWarn); });
      lua_setfield(L, -2, "warn");
      lua_pushcfunction(L, [](auto L) { return log_write(L, nf7::Logger::kError); });
      lua_setfield(L, -2, "error");
    }
    lua_setfield(L, -2, "__index");
  }
}

static void GetLuaObjAndPush(
    lua_State* L, const std::shared_ptr<Thread>& th, File& f) {
  f.interfaceOrThrow<nf7::luajit::Obj>().Build().
      Then([th, L](auto fu) {
        th->ljq()->Push(th->ctx(), [th, L, fu](auto) mutable {
          try {
            const auto& obj = fu.value();
            if (th->ljq() != obj->ljq()) {
              throw nf7::Exception {"the object is built on other LuaJIT context"};
            }
            lua_rawgeti(L, LUA_REGISTRYINDEX, obj->index());
            th->Resume(L, 1);
          } catch (nf7::Exception& e) {
            th->Resume(L, luajit::PushAll(L, nullptr, e.msg()));
          }
        });
      });
}

}  // namespace nf7::luajit