From 022e40cc71beda874d0bad2cec227e472d5dd4af Mon Sep 17 00:00:00 2001 From: Benoit Germain Date: Fri, 18 Apr 2025 10:26:19 +0200 Subject: New feature: Linda periodical cancellation checks * lanes.linda() api change: takes all settings in a single table argument * new linda creation argument wake_period * new lanes.configure setting linda_wake_period * adjusted all unit tests (one TODO test fails on purpose) --- src/lanes.lua | 14 +++++ src/linda.cpp | 152 +++++++++++++++++++++++++++------------------------ src/linda.hpp | 10 +++- src/lindafactory.cpp | 9 ++- src/universe.cpp | 8 +++ src/universe.hpp | 2 + 6 files changed, 118 insertions(+), 77 deletions(-) (limited to 'src') diff --git a/src/lanes.lua b/src/lanes.lua index bd94a14..3ee959c 100644 --- a/src/lanes.lua +++ b/src/lanes.lua @@ -96,6 +96,7 @@ local default_params = -- it looks also like LuaJIT allocator may not appreciate direct use of its allocator for other purposes than the VM operation internal_allocator = isLuaJIT and "libc" or "allocator", keepers_gc_threshold = -1, + linda_wake_period = 'never', nb_user_keepers = 0, on_state_create = nil, shutdown_timeout = 0.25, @@ -141,6 +142,19 @@ local param_checkers = end return true end, + linda_wake_period = function(val_) + -- linda_wake_period should be a number > 0, or the string 'never' + if val_ == 'never' then + return true + end + if type(val_) ~= "number" then + return nil, "not a number" + end + if val_ <= 0 then + return nil, "value out of range" + end + return true + end, nb_user_keepers = function(val_) -- nb_user_keepers should be a number in [0,100] (so that nobody tries to run OOM by specifying a huge amount) if type(val_) ~= "number" then diff --git a/src/linda.cpp b/src/linda.cpp index 0cdacfa..a9ae61c 100644 --- a/src/linda.cpp +++ b/src/linda.cpp @@ -237,11 +237,23 @@ namespace { _lane->waiting_on = &_linda->writeHappened; _lane->status.store(Lane::Waiting, std::memory_order_release); } + + // wait until the final target date by small increments, interrupting regularly so that we can check for cancel requests, + // in case some timing issue caused a cancel request to be issued, and the condvar signalled, before we actually wait for it + auto const [_forceTryAgain, _until_check_cancel] = std::invoke([_until, wakePeriod = _linda->getWakePeriod()] { + auto _until_check_cancel{ std::chrono::time_point::max() }; + if (wakePeriod.count() > 0.0f) { + _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast(wakePeriod); + } + bool const _forceTryAgain{ _until_check_cancel < _until }; + return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : _until); + }); + // not enough data to read: wakeup when data was sent, or when timeout is reached std::unique_lock _guard{ _keeper->mutex, std::adopt_lock }; - std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until) }; + std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until_check_cancel) }; _guard.release(); // we don't want to unlock the mutex on exit! - _try_again = (_status == std::cv_status::no_timeout); // detect spurious wakeups + _try_again = _forceTryAgain || (_status == std::cv_status::no_timeout); // detect spurious wakeups if (_lane != nullptr) { _lane->waiting_on = nullptr; _lane->status.store(_prev_status, std::memory_order_release); @@ -296,9 +308,10 @@ LUAG_FUNC(linda); // ################################################################################################# // ################################################################################################# -Linda::Linda(Universe* const U_, LindaGroup const group_, std::string_view const& name_) +Linda::Linda(Universe* const U_, std::string_view const& name_, lua_Duration const wake_period_, LindaGroup const group_) : DeepPrelude{ LindaFactory::Instance } , U{ U_ } +, wakePeriod{ wake_period_ } , keeperIndex{ group_ % U_->keepers.getNbKeepers() } { setName(name_); @@ -330,9 +343,13 @@ Linda* Linda::CreateTimerLinda(lua_State* const L_) STACK_CHECK_START_REL(L_, 0); // L_: // Initialize 'timerLinda'; a common Linda object shared by all states lua_pushcfunction(L_, LG_linda); // L_: lanes.linda - luaG_pushstring(L_, "lanes-timer"); // L_: lanes.linda "lanes-timer" - lua_pushinteger(L_, 0); // L_: lanes.linda "lanes-timer" 0 - lua_call(L_, 2, 1); // L_: linda + lua_createtable(L_, 0, 3); // L_: lanes.linda {} + luaG_pushstring(L_, "lanes-timer"); // L_: lanes.linda {} "lanes-timer" + luaG_setfield(L_, StackIndex{ -2 }, std::string_view{ "name" }); // L_: lanes.linda { .name="lanes-timer" } + lua_pushinteger(L_, 0); // L_: lanes.linda { .name="lanes-timer" } 0 + luaG_setfield(L_, StackIndex{ -2 }, std::string_view{ "group" }); // L_: lanes.linda { .name="lanes-timer" .group = 0 } + // note that wake_period is not set (will default to the value in the universe) + lua_call(L_, 1, 1); // L_: linda STACK_CHECK(L_, 1); // Proxy userdata contents is only a 'DeepPrelude*' pointer @@ -941,11 +958,23 @@ LUAG_FUNC(linda_send) _lane->waiting_on = &_linda->readHappened; _lane->status.store(Lane::Waiting, std::memory_order_release); } + + // wait until the final target date by small increments, interrupting regularly so that we can check for cancel requests, + // in case some timing issue caused a cancel request to be issued, and the condvar signalled, before we actually wait for it + auto const [_forceTryAgain, _until_check_cancel] = std::invoke([_until, wakePeriod = _linda->getWakePeriod()] { + auto _until_check_cancel{ std::chrono::time_point::max() }; + if (wakePeriod.count() > 0.0f) { + _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast(wakePeriod); + } + bool const _forceTryAgain{ _until_check_cancel < _until }; + return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : _until); + }); + // could not send because no room: wait until some data was read before trying again, or until timeout is reached std::unique_lock _guard{ _keeper->mutex, std::adopt_lock }; - std::cv_status const status{ _linda->readHappened.wait_until(_guard, _until) }; + std::cv_status const status{ _linda->readHappened.wait_until(_guard, _until_check_cancel) }; _guard.release(); // we don't want to unlock the mutex on exit! - _try_again = (status == std::cv_status::no_timeout); // detect spurious wakeups + _try_again = _forceTryAgain || (status == std::cv_status::no_timeout); // detect spurious wakeups if (_lane != nullptr) { _lane->waiting_on = nullptr; _lane->status.store(_prev_status, std::memory_order_release); @@ -1129,88 +1158,69 @@ namespace { // ################################################################################################# /* - * ud = lanes.linda( [name[,group[,close_handler]]]) + * ud = lanes.linda{.name = , .group = , .close_handler = , .wake_period = } * * returns a linda object, or raises an error if creation failed */ LUAG_FUNC(linda) { - static constexpr StackIndex kLastArg{ LUA_VERSION_NUM >= 504 ? 3 : 2 }; + // unpack the received table on the stack, putting name wake_period group close_handler in that order StackIndex const _top{ lua_gettop(L_) }; - luaL_argcheck(L_, _top <= kLastArg, _top, "too many arguments"); - StackIndex _closeHandlerIdx{}; - StackIndex _nameIdx{}; - StackIndex _groupIdx{}; - for (StackIndex const _i : std::ranges::iota_view{ StackIndex{ 1 }, StackIndex{ _top + 1 }}) { - switch (luaG_type(L_, _i)) { + luaL_argcheck(L_, _top <= 1, _top, "too many arguments"); + if (_top == 0) { + lua_settop(L_, 3); // L_: nil nil nil + } + else if (!lua_istable(L_, kIdxTop)) { + luaL_argerror(L_, 1, "expecting a table"); + } else { + auto* const _U{ Universe::Get(L_) }; + lua_getfield(L_, 1, "wake_period"); // L_: {} wake_period + if (lua_isnil(L_, kIdxTop)) { + lua_pop(L_, 1); + lua_pushnumber(L_, _U->lindaWakePeriod.count()); + } else { + luaL_argcheck(L_, luaL_optnumber(L_, 2, 0) > 0, 1, "wake_period must be > 0"); + } + + lua_getfield(L_, 1, "group"); // L_: {} wake_period group + int const _nbKeepers{ _U->keepers.getNbKeepers() }; + if (lua_isnil(L_, kIdxTop)) { + luaL_argcheck(L_, _nbKeepers < 2, 0, "Group is mandatory in multiple Keeper scenarios"); + } else { + int const _group{ static_cast(lua_tointeger(L_, kIdxTop)) }; + luaL_argcheck(L_, _group >= 0 && _group < _nbKeepers, 1, "group out of range"); + } + #if LUA_VERSION_NUM >= 504 // to-be-closed support starts with Lua 5.4 - case LuaType::FUNCTION: - luaL_argcheck(L_, _closeHandlerIdx == 0, _i, "More than one __close handler"); - _closeHandlerIdx = _i; - break; - - case LuaType::USERDATA: - case LuaType::TABLE: - luaL_argcheck(L_, _closeHandlerIdx == 0, _i, "More than one __close handler"); - luaL_argcheck(L_, luaL_getmetafield(L_, _i, "__call") != 0, _i, "__close handler is not callable"); + lua_getfield(L_, 1, "close_handler"); // L_: {} wake_period group close_handler + LuaType const _handlerType{ luaG_type(L_, kIdxTop) }; + if (_handlerType == LuaType::NIL) { + lua_pop(L_, 1); // L_: {} wake_period group + } else if (_handlerType == LuaType::USERDATA || _handlerType == LuaType::TABLE) { + luaL_argcheck(L_, luaL_getmetafield(L_, kIdxTop, "__call") != 0, 1, "__close handler is not callable"); lua_pop(L_, 1); // luaL_getmetafield() pushed the field, we need to pop it - _closeHandlerIdx = _i; - break; + } else { + luaL_argcheck(L_, _handlerType == LuaType::FUNCTION, 1, "__close handler is not a function"); + } #endif // LUA_VERSION_NUM >= 504 - case LuaType::STRING: - luaL_argcheck(L_, _nameIdx == 0, _i, "More than one name"); - _nameIdx = _i; - break; - - case LuaType::NUMBER: - luaL_argcheck(L_, _groupIdx == 0, _i, "More than one group"); - _groupIdx = _i; - break; - - default: - luaL_argcheck(L_, false, _i, "Bad argument type (should be a string, a number, or a callable type)"); - } - } - - int const _nbKeepers{ Universe::Get(L_)->keepers.getNbKeepers() }; - if (!_groupIdx) { - luaL_argcheck(L_, _nbKeepers < 2, 0, "Group is mandatory in multiple Keeper scenarios"); - } else { - int const _group{ static_cast(lua_tointeger(L_, _groupIdx)) }; - luaL_argcheck(L_, _group >= 0 && _group < _nbKeepers, _groupIdx, "Group out of range"); + auto const _nameType{ luaG_getfield(L_, StackIndex{ 1 }, "name") }; // L_: {} wake_period group [close_handler] name + luaL_argcheck(L_, _nameType == LuaType::NIL || _nameType == LuaType::STRING, 1, "name is not a string"); + lua_replace(L_, 1); // L_: name wake_period group [close_handler] } // done with argument checking, let's proceed - if constexpr (LUA_VERSION_NUM >= 504) { - // make sure we have kMaxArgs arguments on the stack for processing, with name, group, and handler, in that order - lua_settop(L_, kLastArg); // L_: a b c - // If either index is 0, lua_settop() adjusted the stack with a nil in slot kLastArg - lua_pushvalue(L_, _closeHandlerIdx ? _closeHandlerIdx : kLastArg); // L_: a b c close_handler - lua_pushvalue(L_, _groupIdx ? _groupIdx : kLastArg); // L_: a b c close_handler group - lua_pushvalue(L_, _nameIdx ? _nameIdx : kLastArg); // L_: a b c close_handler group name - lua_replace(L_, 1); // L_: name b c close_handler group - lua_replace(L_, 2); // L_: name group c close_handler - lua_replace(L_, 3); // L_: name group close_handler - + if (lua_gettop(L_) == 4) { // if we have a __close handler, we need a uservalue slot to store it - UserValueCount const _nuv{ _closeHandlerIdx ? 1 : 0 }; - LindaFactory::Instance.pushDeepUserdata(DestState{ L_ }, _nuv); // L_: name group close_handler linda - if (_closeHandlerIdx != 0) { - lua_replace(L_, 2); // L_: name linda close_handler - lua_setiuservalue(L_, StackIndex{ 2 }, UserValueIndex{ 1 }); // L_: name linda - } + LindaFactory::Instance.pushDeepUserdata(DestState{ L_ }, UserValueCount{ 1 }); // L_: name wake_period group [close_handler] linda + lua_replace(L_, 3); // L_: name wake_period linda close_handler + lua_setiuservalue(L_, StackIndex{ 3 }, UserValueIndex{ 1 }); // L_: name wake_period linda // depending on whether we have a handler or not, the stack is not in the same state at this point // just make sure we have our Linda at the top LUA_ASSERT(L_, ToLinda(L_, kIdxTop)); return 1; } else { // no to-be-closed support - // ensure we have name, group in that order on the stack - if (_nameIdx > _groupIdx) { - lua_insert(L_, 1); // L_: name group - } - LindaFactory::Instance.pushDeepUserdata(DestState{ L_ }, UserValueCount{ 0 }); // L_: name group linda + LindaFactory::Instance.pushDeepUserdata(DestState{ L_ }, UserValueCount{ 0 }); // L_: name wake_period group linda return 1; } - } diff --git a/src/linda.hpp b/src/linda.hpp index 2d5c9dc..7874db3 100644 --- a/src/linda.hpp +++ b/src/linda.hpp @@ -42,19 +42,21 @@ class Linda final }; using enum Status; - private: + public: + Universe* const U{ nullptr }; // the universe this linda belongs to + private: static constexpr size_t kEmbeddedNameLength = 24; using EmbeddedName = std::array; // depending on the name length, it is either embedded inside the Linda, or allocated separately std::variant nameVariant{}; // counts the keeper operations in progress std::atomic keeperOperationCount{}; + lua_Duration wakePeriod{}; public: std::condition_variable readHappened{}; std::condition_variable writeHappened{}; - Universe* const U{ nullptr }; // the universe this linda belongs to KeeperIndex const keeperIndex{ -1 }; // the keeper associated to this linda Status cancelStatus{ Status::Active }; @@ -68,7 +70,7 @@ class Linda final static void operator delete(void* p_) { static_cast(p_)->U->internalAllocator.free(p_, sizeof(Linda)); } ~Linda(); - Linda(Universe* U_, LindaGroup group_, std::string_view const& name_); + Linda(Universe* U_, std::string_view const& name_, lua_Duration wake_period_, LindaGroup group_); Linda() = delete; // non-copyable, non-movable Linda(Linda const&) = delete; @@ -92,6 +94,8 @@ class Linda final [[nodiscard]] std::string_view getName() const; [[nodiscard]] + auto getWakePeriod() const { return wakePeriod; } + [[nodiscard]] bool inKeeperOperation() const { return keeperOperationCount.load(std::memory_order_seq_cst) != 0; } template [[nodiscard]] diff --git a/src/lindafactory.cpp b/src/lindafactory.cpp index 42d0984..4eab0c1 100644 --- a/src/lindafactory.cpp +++ b/src/lindafactory.cpp @@ -108,9 +108,11 @@ std::string_view LindaFactory::moduleName() const DeepPrelude* LindaFactory::newDeepObjectInternal(lua_State* const L_) const { - // we always expect name and group at the bottom of the stack (either can be nil). any extra stuff we ignore and keep unmodified + STACK_CHECK_START_REL(L_, 0); + // we always expect name, wake_period, group at the bottom of the stack (either can be nil). any extra stuff we ignore and keep unmodified std::string_view _linda_name{ luaG_tostring(L_, StackIndex{ 1 }) }; - LindaGroup _linda_group{ static_cast(lua_tointeger(L_, 2)) }; + auto const _wake_period{ static_cast(lua_tonumber(L_, 2)) }; + LindaGroup const _linda_group{ static_cast(lua_tointeger(L_, 3)) }; // store in the linda the location of the script that created it if (_linda_name == "auto") { @@ -129,6 +131,7 @@ DeepPrelude* LindaFactory::newDeepObjectInternal(lua_State* const L_) const // The deep data is allocated separately of Lua stack; we might no longer be around when last reference to it is being released. // One can use any memory allocation scheme. Just don't use L's allocF because we don't know which state will get the honor of GCing the linda Universe* const _U{ Universe::Get(L_) }; - Linda* const _linda{ new (_U) Linda{ _U, _linda_group, _linda_name } }; + Linda* const _linda{ new (_U) Linda{ _U, _linda_name, _wake_period, _linda_group } }; + STACK_CHECK(L_, 0); return _linda; } diff --git a/src/universe.cpp b/src/universe.cpp index 89ad02a..335f056 100644 --- a/src/universe.cpp +++ b/src/universe.cpp @@ -153,6 +153,14 @@ Universe* Universe::Create(lua_State* const L_) lua_setmetatable(L_, -2); // L_: settings universe lua_pop(L_, 1); // L_: settings + std::ignore = luaG_getfield(L_, kIdxSettings, "linda_wake_period"); // L_: settings linda_wake_period + if (luaG_type(L_, kIdxTop) == LuaType::NUMBER) { + _U->lindaWakePeriod = lua_Duration{ lua_tonumber(L_, kIdxTop) }; + } else { + LUA_ASSERT(L_, luaG_tostring(L_, kIdxTop) == "never"); + } + lua_pop(L_, 1); // L_: settings + std::ignore = luaG_getfield(L_, kIdxSettings, "strip_functions"); // L_: settings strip_functions _U->stripFunctions = lua_toboolean(L_, -1) ? true : false; lua_pop(L_, 1); // L_: settings diff --git a/src/universe.hpp b/src/universe.hpp index 42a3d83..0c5e659 100644 --- a/src/universe.hpp +++ b/src/universe.hpp @@ -99,6 +99,8 @@ class Universe final Keepers keepers; + lua_Duration lindaWakePeriod{}; + // Initialized by 'init_once_LOCKED()': the deep userdata Linda object // used for timers (each lane will get a proxy to this) Linda* timerLinda{ nullptr }; -- cgit v1.2.3-55-g6feb