1 files changed, 258 insertions, 252 deletions

diff --git a/src/linda.cpp b/src/linda.cpp
index 1119d71..1f4b19d 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -47,7 +47,7 @@ namespace {
     {
         STACK_CHECK_START_REL(L_, 0);
         for (StackIndex const _i : std::ranges::iota_view{ start_, StackIndex{ end_ + 1 } }) {
-            LuaType const _t{ luaG_type(L_, _i) };
+            LuaType const _t{ luaW_type(L_, _i) };
             switch (_t) {
             case LuaType::BOOLEAN:
             case LuaType::NUMBER:
@@ -62,7 +62,7 @@ namespace {
 
             case LuaType::LIGHTUSERDATA:
                 {
-                    static constexpr std::array<std::reference_wrapper<UniqueKey const>, 3> kKeysToCheck{ kLindaBatched, kCancelError, kNilSentinel };
+                    static constexpr std::array<std::reference_wrapper<UniqueKey const>, 2> kKeysToCheck{ kCancelError, kNilSentinel };
                     for (UniqueKey const& _key : kKeysToCheck) {
                         if (_key.equals(L_, _i)) {
                             raise_luaL_error(L_, "argument #%d: can't use %s as a slot", _i, _key.debugName.data());
@@ -109,13 +109,13 @@ namespace {
     {
         Linda* const _linda{ ToLinda<OPT>(L_, idx_) };
         if (_linda != nullptr) {
-            luaG_pushstring(L_, "Linda: ");
+            luaW_pushstring(L_, "Linda: ");
             std::string_view const _lindaName{ _linda->getName() };
             if (!_lindaName.empty()) {
-                luaG_pushstring(L_, _lindaName);
+                luaW_pushstring(L_, _lindaName);
             } else {
                 // obfuscate the pointer so that we can't read the value with our eyes out of a script
-                luaG_pushstring(L_, "%p", _linda->obfuscated());
+                luaW_pushstring(L_, "%p", _linda->obfuscated());
             }
             lua_concat(L_, 2);
             return 1;
@@ -123,6 +123,8 @@ namespace {
         return 0;
     }
 
+    // #############################################################################################
+
     // a helper to process the timeout argument of linda:send() and linda:receive()
     [[nodiscard]]
     static auto ProcessTimeoutArg(lua_State* const L_)
@@ -130,7 +132,7 @@ namespace {
         StackIndex _key_i{ 2 }; // index of first slot, if timeout not there
 
         std::chrono::time_point<std::chrono::steady_clock> _until{ std::chrono::time_point<std::chrono::steady_clock>::max() };
-        if (luaG_type(L_, StackIndex{ 2 }) == LuaType::NUMBER) { // we don't want to use lua_isnumber() because of autocoercion
+        if (luaW_type(L_, StackIndex{ 2 }) == LuaType::NUMBER) { // we don't want to use lua_isnumber() because of autocoercion
             lua_Duration const _duration{ lua_tonumber(L_, 2) };
             if (_duration.count() >= 0.0) {
                 _until = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(_duration);
@@ -145,6 +147,160 @@ namespace {
     }
 
     // #############################################################################################
+    static bool WaitInternal([[maybe_unused]] lua_State* const L_, Lane* const lane_, Linda* const linda_, Keeper* const keeper_, std::condition_variable& waitingOn_, std::chrono::time_point<std::chrono::steady_clock> until_)
+    {
+        Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
+        if (lane_ != nullptr) {
+            // change status of lane to "waiting"
+            _prev_status = lane_->status.load(std::memory_order_acquire); // Running, most likely
+            LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
+            LUA_ASSERT(L_, lane_->waiting_on == nullptr);
+            lane_->waiting_on = &waitingOn_;
+            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<std::chrono::steady_clock>::max() };
+            if (wakePeriod.count() > 0.0f) {
+                _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(wakePeriod);
+            }
+            bool const _forceTryAgain{ _until_check_cancel < until_ };
+            return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : until_);
+        });
+
+        // operation can't complete: wake when it is signalled to be possible, or when timeout is reached
+        std::unique_lock<std::mutex> _guard{ keeper_->mutex, std::adopt_lock };
+        std::cv_status const _status{ waitingOn_.wait_until(_guard, _until_check_cancel) };
+        _guard.release(); // we don't want to unlock the mutex on exit!
+        bool const _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);
+        }
+        return _try_again;
+    }
+
+    // #############################################################################################
+
+    // the implementation for linda:receive() and linda:receive_batched()
+    static int ReceiveInternal(lua_State* const L_, bool const batched_)
+    {
+        Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
+
+        auto const [_key_i, _until] = ProcessTimeoutArg(L_);
+
+        keeper_api_t _selected_keeper_receive{ nullptr };
+        int _expected_pushed_min{ 0 }, _expected_pushed_max{ 0 };
+        // are we in batched mode?
+        if (batched_) {
+            // make sure the keys are of a valid type
+            CheckKeyTypes(L_, _key_i, _key_i);
+            // receive multiple values from a single slot
+            _selected_keeper_receive = KEEPER_API(receive_batched);
+            // we expect a user-defined amount of return value
+            _expected_pushed_min = (int) luaL_checkinteger(L_, _key_i + 1);
+            if (_expected_pushed_min < 1) {
+                raise_luaL_argerror(L_, StackIndex{ _key_i + 1 }, "bad min count");
+            }
+            _expected_pushed_max = (int) luaL_optinteger(L_, _key_i + 2, _expected_pushed_min);
+            // don't forget to count the slot in addition to the values
+            ++_expected_pushed_min;
+            ++_expected_pushed_max;
+            if (_expected_pushed_min > _expected_pushed_max) {
+                raise_luaL_argerror(L_, StackIndex{ _key_i + 2 }, "batched min/max error");
+            }
+        } else {
+            // make sure the keys are of a valid type
+            CheckKeyTypes(L_, _key_i, StackIndex{ lua_gettop(L_) });
+            // receive a single value, checking multiple slots
+            _selected_keeper_receive = KEEPER_API(receive);
+            // we expect a single (value, slot) pair of returned values
+            _expected_pushed_min = _expected_pushed_max = 2;
+        }
+
+        Lane* const _lane{ kLanePointerRegKey.readLightUserDataValue<Lane>(L_) };
+        Keeper* const _keeper{ _linda->whichKeeper() };
+        KeeperState const _K{ _keeper ? _keeper->K : KeeperState{ static_cast<lua_State*>(nullptr) } };
+        if (_K == nullptr)
+            return 0;
+
+        CancelRequest _cancel{ CancelRequest::None };
+        KeeperCallResult _pushed{};
+
+        STACK_CHECK_START_REL(_K, 0);
+        for (bool _try_again{ true };;) {
+            if (_lane != nullptr) {
+                _cancel = _lane->cancelRequest.load(std::memory_order_relaxed);
+            }
+            _cancel = (_cancel != CancelRequest::None)
+                ? _cancel
+                : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
+
+            // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
+            if (!_try_again || _cancel != CancelRequest::None) {
+                _pushed.emplace(0);
+                break;
+            }
+
+            // all arguments of receive() but the first are passed to the keeper's receive function
+            STACK_CHECK(_K, 0);
+            _pushed = keeper_call(_K, _selected_keeper_receive, L_, _linda, _key_i);
+            if (!_pushed.has_value()) {
+                break;
+            }
+            if (_pushed.value() > 0) {
+                LUA_ASSERT(L_, _pushed.value() >= _expected_pushed_min && _pushed.value() <= _expected_pushed_max);
+                if (kRestrictedChannel.equals(L_, StackIndex{ kIdxTop })) {
+                    raise_luaL_error(L_, "Key is restricted");
+                }
+                _linda->readHappened.notify_all();
+                break;
+            }
+
+            if (std::chrono::steady_clock::now() >= _until) {
+                break; /* instant timeout */
+            }
+
+            // nothing received, wait until timeout or signalled that we should try again
+            _try_again = WaitInternal(L_, _lane, _linda, _keeper, _linda->writeHappened, _until);
+        }
+        STACK_CHECK(_K, 0);
+
+        if (!_pushed.has_value()) {
+            raise_luaL_error(L_, "tried to copy unsupported types");
+        }
+
+        switch (_cancel) {
+        case CancelRequest::None:
+            {
+                int const _nbPushed{ _pushed.value() };
+                if (_nbPushed == 0) {
+                    // not enough data in the linda slot to fulfill the request, return nil, "timeout"
+                    lua_pushnil(L_);
+                    luaW_pushstring(L_, "timeout");
+                    return 2;
+                }
+                return _nbPushed;
+            }
+
+        case CancelRequest::Soft:
+            // if user wants to soft-cancel, the call returns nil, kCancelError
+            lua_pushnil(L_);
+            kCancelError.pushKey(L_);
+            return 2;
+
+        case CancelRequest::Hard:
+            // raise an error interrupting execution only in case of hard cancel
+            raise_cancel_error(L_); // raises an error and doesn't return
+
+        default:
+            raise_luaL_error(L_, "internal error: unknown cancel request");
+        }
+    }
+
+    // #############################################################################################
     // #############################################################################################
 } // namespace
 // #################################################################################################
@@ -158,9 +314,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_);
@@ -181,6 +338,7 @@ Keeper* Linda::acquireKeeper() const
     Keeper* const _keeper{ whichKeeper() };
     if (_keeper) {
         _keeper->mutex.lock();
+        keeperOperationCount.fetch_add(1, std::memory_order_seq_cst);
     }
     return _keeper;
 }
@@ -192,13 +350,17 @@ 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 {}
+    luaW_pushstring(L_, "lanes-timer");                                                            // L_: lanes.linda {} "lanes-timer"
+    luaW_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
+    luaW_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
-    auto const _timerLinda{ *luaG_tofulluserdata<Linda*>(L_, kIdxTop) };
+    auto const _timerLinda{ *luaW_tofulluserdata<Linda*>(L_, kIdxTop) };
     // increment refcount so that this linda remains alive as long as the universe exists.
     _timerLinda->refcount.fetch_add(1, std::memory_order_relaxed);
     lua_pop(L_, 1);                                                                                // L_:
@@ -261,7 +423,6 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
     // doing LindaFactory::deleteDeepObjectInternal -> keeper_call(clear)
     lua_gc(L_, LUA_GCSTOP, 0);
 
-    LUA_ASSERT_CODE(auto const _koip{ _linda->startKeeperOperation(L_) });
     // if we didn't do anything wrong, the keeper stack should be clean
     LUA_ASSERT(L_, lua_gettop(_K) == 0);
 
@@ -269,7 +430,7 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
     lua_pushcfunction(L_, f_);
     lua_insert(L_, 1);
     // do a protected call
-    LuaError const _rc{ lua_pcall(L_, lua_gettop(L_) - 1, LUA_MULTRET, 0) };
+    LuaError const _rc{ ToLuaError(lua_pcall(L_, lua_gettop(L_) - 1, LUA_MULTRET, 0)) };
     // whatever happens, the keeper state stack must be empty when we are done
     lua_settop(_K, 0);
 
@@ -291,7 +452,7 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
 
 void Linda::pushCancelString(lua_State* L_) const
 {
-    luaG_pushstring(L_, cancelStatus == Status::Cancelled ? "cancelled" : "active");
+    luaW_pushstring(L_, cancelStatus == Status::Cancelled ? "cancelled" : "active");
 }
 
 // #################################################################################################
@@ -300,6 +461,7 @@ void Linda::releaseKeeper(Keeper* const keeper_) const
 {
     if (keeper_) { // can be nullptr if we tried to acquire during shutdown
         assert(keeper_ == whichKeeper());
+        keeperOperationCount.fetch_sub(1, std::memory_order_seq_cst);
         keeper_->mutex.unlock();
     }
 }
@@ -343,7 +505,7 @@ void Linda::setName(std::string_view const& name_)
 LUAG_FUNC(linda_cancel)
 {
     Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
-    std::string_view const _who{ luaG_optstring(L_, StackIndex{ 2 }, "both") };
+    std::string_view const _who{ luaW_optstring(L_, StackIndex{ 2 }, "both") };
     // make sure we got 2 arguments: the linda and the cancellation mode
     luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments");
 
@@ -431,13 +593,13 @@ static int linda_index_string(lua_State* L_)
     Linda* const _linda{ ToLinda<false>(L_, kIdxSelf) };
     LUA_ASSERT(L_, lua_gettop(L_) == 2);                                                           // L_: linda "key"
 
-    std::string_view const _keystr{ luaG_tostring(L_, kIdxKey) };
+    std::string_view const _keystr{ luaW_tostring(L_, kIdxKey) };
     lua_settop(L_, 2); // keep only our original arguments on the stack
 
     // look in metatable first
     lua_getmetatable(L_, kIdxSelf);                                                                // L_: linda "key" mt
     lua_replace(L_, -3);                                                                           // L_: mt "key"
-    if (luaG_rawget(L_, StackIndex{ -2 }) != LuaType::NIL) { // found something?                   // L_: mt value
+    if (luaW_rawget(L_, StackIndex{ -2 }) != LuaType::NIL) { // found something?                   // L_: mt value
         return 1; // done
     }
 
@@ -457,12 +619,12 @@ static LUAG_FUNC(linda_index)
     static constexpr StackIndex kIdxKey{ 2 };
     LUA_ASSERT(L_, lua_gettop(L_) == 2);
 
-    switch (luaG_type(L_, kIdxKey)) {
+    switch (luaW_type(L_, kIdxKey)) {
     case LuaType::STRING:
         return linda_index_string(L_); // stack modification is undefined, returned value is at the top
 
     default: // unknown key
-        raise_luaL_error(L_, "Unsupported linda indexing key type %s", luaG_typename(L_, kIdxKey).data());
+        raise_luaL_error(L_, "Unsupported linda indexing key type %s", luaW_typename(L_, kIdxKey).data());
     }
 }
 
@@ -605,7 +767,7 @@ LUAG_FUNC(linda_limit)
             int const _nargs{ lua_gettop(L_) };
             luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments");
             // make sure we got a numeric limit, or "unlimited", (or nothing)
-            bool const _unlimited{ luaG_tostring(L_, StackIndex{ 3 }) == "unlimited" };
+            bool const _unlimited{ luaW_tostring(L_, StackIndex{ 3 }) == "unlimited" };
             LindaLimit const _val{ _unlimited ? std::numeric_limits<LindaLimit::type>::max() : static_cast<LindaLimit::type>(luaL_optinteger(L_, 3, 0)) };
             if (_val < 0) {
                 raise_luaL_argerror(L_, StackIndex{ 3 }, "limit must be >= 0");
@@ -616,23 +778,23 @@ LUAG_FUNC(linda_limit)
             KeeperCallResult _pushed;
             if (_linda->cancelStatus == Linda::Active) {
                 if (_unlimited) {
-                    LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaG_tostring(L_, StackIndex{ 3 }) == "unlimited");
+                    LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaW_tostring(L_, StackIndex{ 3 }) == "unlimited");
                     // inside the Keeper, unlimited is signified with a -1 limit (can't use nil because of nil kNilSentinel conversions!)
                     lua_pop(L_, 1);                                                                // L_: linda slot
                     lua_pushinteger(L_, -1);                                                       // L_: linda slot nil
                 }
                 Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(limit), L_, _linda, StackIndex{ 2 });
-                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 2) && luaG_type(L_, kIdxTop) == LuaType::STRING);
+                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 2) && luaW_type(L_, kIdxTop) == LuaType::STRING);
                 if (_nargs == 3) { // 3 args: setting the limit
                     // changing the limit: no error, boolean value saying if we should wake blocked writer threads
-                    LUA_ASSERT(L_, luaG_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN);           // L_: bool string
+                    LUA_ASSERT(L_, luaW_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN);           // L_: bool string
                     if (lua_toboolean(L_, -2)) {
                         _linda->readHappened.notify_all(); // To be done from within the 'K' locking area
                     }
                 } else { // 2 args: reading the limit
                     // reading the limit: a number >=0 or "unlimited"
-                    LUA_ASSERT(L_, luaG_type(L_, StackIndex{ -2 }) == LuaType::NUMBER || luaG_tostring(L_, StackIndex{ -2 }) == "unlimited");
+                    LUA_ASSERT(L_, luaW_type(L_, StackIndex{ -2 }) == LuaType::NUMBER || luaW_tostring(L_, StackIndex{ -2 }) == "unlimited");
                 }
             } else { // linda is cancelled
                 // do nothing and return nil,lanes.cancel_error
@@ -650,153 +812,25 @@ LUAG_FUNC(linda_limit)
 // #################################################################################################
 
 /*
- * 2 modes of operation
- * [val, slot]= linda:receive([timeout_secs_num=nil], key_num|str|bool|lightuserdata [, ...] )
+ * [val, slot] = linda:receive([timeout_secs_num=nil], key_num|str|bool|lightuserdata [, ...] )
  * Consumes a single value from the Linda, in any slot.
  * Returns: received value (which is consumed from the slot), and the slot which had it
-
- * [val1, ... valCOUNT]= linda_receive( linda_ud, [timeout_secs_num=-1], linda.batched, key_num|str|bool|lightuserdata, min_COUNT[, max_COUNT])
- * Consumes between min_COUNT and max_COUNT values from the linda, from a single slot.
- * returns the actual consumed values, or nil if there weren't enough values to consume
  */
 LUAG_FUNC(linda_receive)
 {
-    static constexpr lua_CFunction _receive{
-        +[](lua_State* const L_) {
-            Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
-
-            auto [_key_i, _until] = ProcessTimeoutArg(L_);
-
-            keeper_api_t _selected_keeper_receive{ nullptr };
-            int _expected_pushed_min{ 0 }, _expected_pushed_max{ 0 };
-            // are we in batched mode?
-            if (kLindaBatched.equals(L_, _key_i)) {
-                // no need to pass linda.batched in the keeper state
-                ++_key_i;
-                // make sure the keys are of a valid type
-                CheckKeyTypes(L_, _key_i, _key_i);
-                // receive multiple values from a single slot
-                _selected_keeper_receive = KEEPER_API(receive_batched);
-                // we expect a user-defined amount of return value
-                _expected_pushed_min = (int) luaL_checkinteger(L_, _key_i + 1);
-                if (_expected_pushed_min < 1) {
-                    raise_luaL_argerror(L_, StackIndex{ _key_i + 1 }, "bad min count");
-                }
-                _expected_pushed_max = (int) luaL_optinteger(L_, _key_i + 2, _expected_pushed_min);
-                // don't forget to count the slot in addition to the values
-                ++_expected_pushed_min;
-                ++_expected_pushed_max;
-                if (_expected_pushed_min > _expected_pushed_max) {
-                    raise_luaL_argerror(L_, StackIndex{ _key_i + 2 }, "batched min/max error");
-                }
-            } else {
-                // make sure the keys are of a valid type
-                CheckKeyTypes(L_, _key_i, StackIndex{ lua_gettop(L_) });
-                // receive a single value, checking multiple slots
-                _selected_keeper_receive = KEEPER_API(receive);
-                // we expect a single (value, slot) pair of returned values
-                _expected_pushed_min = _expected_pushed_max = 2;
-            }
-
-            Lane* const _lane{ kLanePointerRegKey.readLightUserDataValue<Lane>(L_) };
-            Keeper* const _keeper{ _linda->whichKeeper() };
-            KeeperState const _K{ _keeper ? _keeper->K : KeeperState{ static_cast<lua_State*>(nullptr) } };
-            if (_K == nullptr)
-                return 0;
-
-            CancelRequest _cancel{ CancelRequest::None };
-            KeeperCallResult _pushed{};
-
-            STACK_CHECK_START_REL(_K, 0);
-            for (bool _try_again{ true };;) {
-                if (_lane != nullptr) {
-                    _cancel = _lane->cancelRequest.load(std::memory_order_relaxed);
-                }
-                _cancel = (_cancel != CancelRequest::None)
-                    ? _cancel
-                    : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
-                // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
-                if (!_try_again || _cancel != CancelRequest::None) {
-                    _pushed.emplace(0);
-                    break;
-                }
-
-                // all arguments of receive() but the first are passed to the keeper's receive function
-                STACK_CHECK(_K, 0);
-                _pushed = keeper_call(_K, _selected_keeper_receive, L_, _linda, _key_i);
-                if (!_pushed.has_value()) {
-                    break;
-                }
-                if (_pushed.value() > 0) {
-                    LUA_ASSERT(L_, _pushed.value() >= _expected_pushed_min && _pushed.value() <= _expected_pushed_max);
-                    if (kRestrictedChannel.equals(L_, StackIndex{ kIdxTop })) {
-                        raise_luaL_error(L_, "Key is restricted");
-                    }
-                    _linda->readHappened.notify_all();
-                    break;
-                }
-
-                if (std::chrono::steady_clock::now() >= _until) {
-                    break; /* instant timeout */
-                }
-
-                // nothing received, wait until timeout or signalled that we should try again
-                {
-                    Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
-                    if (_lane != nullptr) {
-                        // change status of lane to "waiting"
-                        _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
-                        LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
-                        LUA_ASSERT(L_, _lane->waiting_on == nullptr);
-                        _lane->waiting_on = &_linda->writeHappened;
-                        _lane->status.store(Lane::Waiting, std::memory_order_release);
-                    }
-                    // not enough data to read: wakeup when data was sent, or when timeout is reached
-                    std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
-                    std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until) };
-                    _guard.release(); // we don't want to unlock the mutex on exit!
-                    _try_again = (_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);
-                    }
-                }
-            }
-            STACK_CHECK(_K, 0);
-
-            if (!_pushed.has_value()) {
-                raise_luaL_error(L_, "tried to copy unsupported types");
-            }
-
-            switch (_cancel) {
-            case CancelRequest::None:
-                {
-                    int const _nbPushed{ _pushed.value() };
-                    if (_nbPushed == 0) {
-                        // not enough data in the linda slot to fulfill the request, return nil, "timeout"
-                        lua_pushnil(L_);
-                        luaG_pushstring(L_, "timeout");
-                        return 2;
-                    }
-                    return _nbPushed;
-                }
-
-            case CancelRequest::Soft:
-                // if user wants to soft-cancel, the call returns nil, kCancelError
-                lua_pushnil(L_);
-                kCancelError.pushKey(L_);
-                return 2;
+    return Linda::ProtectedCall(L_, [](lua_State* const L_) { return ReceiveInternal(L_, false); });
+}
 
-            case CancelRequest::Hard:
-                // raise an error interrupting execution only in case of hard cancel
-                raise_cancel_error(L_); // raises an error and doesn't return
+// #################################################################################################
 
-            default:
-                raise_luaL_error(L_, "internal error: unknown cancel request");
-            }
-        }
-    };
-    return Linda::ProtectedCall(L_, _receive);
+/*
+ * [val1, ... valCOUNT] = linda_receive_batched( linda_ud, [timeout_secs_num=-1], key_num|str|bool|lightuserdata, min_COUNT[, max_COUNT])
+ * Consumes between min_COUNT and max_COUNT values from the linda, from a single slot.
+ * returns the actual consumed values, or nil if there weren't enough values to consume
+ */
+LUAG_FUNC(linda_receive_batched)
+{
+    return Linda::ProtectedCall(L_, [](lua_State* const L_) { return ReceiveInternal(L_, true); });
 }
 
 // #################################################################################################
@@ -816,7 +850,7 @@ LUAG_FUNC(linda_restrict)
             int const _nargs{ lua_gettop(L_) };
             luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments");
             // make sure we got a known restrict mode, (or nothing)
-            std::string_view const _mode{ luaG_tostring(L_, StackIndex{ 3 }) };
+            std::string_view const _mode{ luaW_tostring(L_, StackIndex{ 3 }) };
             if (!_mode.empty() && (_mode != "none" && _mode != "set/get" && _mode != "send/receive")) {
                 raise_luaL_argerror(L_, StackIndex{ 3 }, "unknown restrict mode");
             }
@@ -828,7 +862,7 @@ LUAG_FUNC(linda_restrict)
                 Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(restrict), L_, _linda, StackIndex{ 2 });
                 // we should get a single return value: the string describing the previous restrict mode
-                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 1) && luaG_type(L_, kIdxTop) == LuaType::STRING);
+                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 1) && luaW_type(L_, kIdxTop) == LuaType::STRING);
             } else { // linda is cancelled
                 // do nothing and return nil,lanes.cancel_error
                 lua_pushnil(L_);
@@ -889,6 +923,7 @@ LUAG_FUNC(linda_send)
                 _cancel = (_cancel != CancelRequest::None)
                     ? _cancel
                     : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
+
                 // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
                 if (!_try_again || _cancel != CancelRequest::None) {
                     _pushed.emplace(0);
@@ -921,26 +956,7 @@ LUAG_FUNC(linda_send)
                 }
 
                 // storage limit hit, wait until timeout or signalled that we should try again
-                {
-                    Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
-                    if (_lane != nullptr) {
-                        // change status of lane to "waiting"
-                        _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
-                        LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
-                        LUA_ASSERT(L_, _lane->waiting_on == nullptr);
-                        _lane->waiting_on = &_linda->readHappened;
-                        _lane->status.store(Lane::Waiting, std::memory_order_release);
-                    }
-                    // could not send because no room: wait until some data was read before trying again, or until timeout is reached
-                    std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
-                    std::cv_status const status{ _linda->readHappened.wait_until(_guard, _until) };
-                    _guard.release(); // we don't want to unlock the mutex on exit!
-                    _try_again = (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);
-                    }
-                }
+                _try_again = WaitInternal(L_, _lane, _linda, _keeper, _linda->readHappened, _until);
             }
             STACK_CHECK(_K, 0);
 
@@ -966,7 +982,7 @@ LUAG_FUNC(linda_send)
                 } else {
                     // not enough room in the Linda slot to fulfill the request, return nil, "timeout"
                     lua_pushnil(L_);
-                    luaG_pushstring(L_, "timeout");
+                    luaW_pushstring(L_, "timeout");
                     return 2;
                 }
             }
@@ -1001,7 +1017,7 @@ LUAG_FUNC(linda_set)
                     if (kRestrictedChannel.equals(L_, kIdxTop)) {
                         raise_luaL_error(L_, "Key is restricted");
                     }
-                    LUA_ASSERT(L_, _pushed.value() == 2 && luaG_type(L_, kIdxTop) == LuaType::STRING && luaG_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN);
+                    LUA_ASSERT(L_, _pushed.value() == 2 && luaW_type(L_, kIdxTop) == LuaType::STRING && luaW_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN);
 
                     if (_has_data) {
                         // we put some data in the slot, tell readers that they should wake
@@ -1063,7 +1079,7 @@ LUAG_FUNC(linda_towatch)
 LUAG_FUNC(linda_wake)
 {
     Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
-    std::string_view const _who{ luaG_optstring(L_, StackIndex{ 2 }, "both") };
+    std::string_view const _who{ luaW_optstring(L_, StackIndex{ 2 }, "both") };
     // make sure we got 2 arguments: the linda and the wake targets
     luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments");
 
@@ -1102,6 +1118,7 @@ namespace {
             { "get", LG_linda_get },
             { "limit", LG_linda_limit },
             { "receive", LG_linda_receive },
+            { "receive_batched", LG_linda_receive_batched },
             { "restrict", LG_linda_restrict },
             { "send", LG_linda_send },
             { "set", LG_linda_set },
@@ -1118,88 +1135,77 @@ namespace {
 // #################################################################################################
 
 /*
- * ud = lanes.linda( [name[,group[,close_handler]]])
+ * ud = lanes.linda{.name = <string>, .group = <number>, .close_handler = <callable>, .wake_period = <number>}
  *
  * 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 if (luaW_type(L_, kIdxTop) == LuaType::STRING) {
+            if (luaW_tostring(L_, kIdxTop) != "never") {
+                luaL_argerror(L_, 1, "invalid wake_period");
+            } else {
+                lua_pop(L_, 1);
+                lua_pushnumber(L_, 0);
+            }
+        }
+        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<int>(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{ luaW_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<int>(lua_tointeger(L_, _groupIdx)) };
-        luaL_argcheck(L_, _group >= 0 && _group < _nbKeepers, _groupIdx, "Group out of range");
+        auto const _nameType{ luaW_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<true>(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;
     }
-
 }