aboutsummaryrefslogtreecommitdiff
path: root/src/linda.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/linda.cpp')
-rw-r--r--src/linda.cpp169
1 files changed, 73 insertions, 96 deletions
diff --git a/src/linda.cpp b/src/linda.cpp
index fa28385..1f4b19d 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -47,7 +47,7 @@ namespace {
47 { 47 {
48 STACK_CHECK_START_REL(L_, 0); 48 STACK_CHECK_START_REL(L_, 0);
49 for (StackIndex const _i : std::ranges::iota_view{ start_, StackIndex{ end_ + 1 } }) { 49 for (StackIndex const _i : std::ranges::iota_view{ start_, StackIndex{ end_ + 1 } }) {
50 LuaType const _t{ luaG_type(L_, _i) }; 50 LuaType const _t{ luaW_type(L_, _i) };
51 switch (_t) { 51 switch (_t) {
52 case LuaType::BOOLEAN: 52 case LuaType::BOOLEAN:
53 case LuaType::NUMBER: 53 case LuaType::NUMBER:
@@ -109,13 +109,13 @@ namespace {
109 { 109 {
110 Linda* const _linda{ ToLinda<OPT>(L_, idx_) }; 110 Linda* const _linda{ ToLinda<OPT>(L_, idx_) };
111 if (_linda != nullptr) { 111 if (_linda != nullptr) {
112 luaG_pushstring(L_, "Linda: "); 112 luaW_pushstring(L_, "Linda: ");
113 std::string_view const _lindaName{ _linda->getName() }; 113 std::string_view const _lindaName{ _linda->getName() };
114 if (!_lindaName.empty()) { 114 if (!_lindaName.empty()) {
115 luaG_pushstring(L_, _lindaName); 115 luaW_pushstring(L_, _lindaName);
116 } else { 116 } else {
117 // obfuscate the pointer so that we can't read the value with our eyes out of a script 117 // obfuscate the pointer so that we can't read the value with our eyes out of a script
118 luaG_pushstring(L_, "%p", _linda->obfuscated()); 118 luaW_pushstring(L_, "%p", _linda->obfuscated());
119 } 119 }
120 lua_concat(L_, 2); 120 lua_concat(L_, 2);
121 return 1; 121 return 1;
@@ -132,7 +132,7 @@ namespace {
132 StackIndex _key_i{ 2 }; // index of first slot, if timeout not there 132 StackIndex _key_i{ 2 }; // index of first slot, if timeout not there
133 133
134 std::chrono::time_point<std::chrono::steady_clock> _until{ std::chrono::time_point<std::chrono::steady_clock>::max() }; 134 std::chrono::time_point<std::chrono::steady_clock> _until{ std::chrono::time_point<std::chrono::steady_clock>::max() };
135 if (luaG_type(L_, StackIndex{ 2 }) == LuaType::NUMBER) { // we don't want to use lua_isnumber() because of autocoercion 135 if (luaW_type(L_, StackIndex{ 2 }) == LuaType::NUMBER) { // we don't want to use lua_isnumber() because of autocoercion
136 lua_Duration const _duration{ lua_tonumber(L_, 2) }; 136 lua_Duration const _duration{ lua_tonumber(L_, 2) };
137 if (_duration.count() >= 0.0) { 137 if (_duration.count() >= 0.0) {
138 _until = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(_duration); 138 _until = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(_duration);
@@ -147,6 +147,42 @@ namespace {
147 } 147 }
148 148
149 // ############################################################################################# 149 // #############################################################################################
150 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_)
151 {
152 Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
153 if (lane_ != nullptr) {
154 // change status of lane to "waiting"
155 _prev_status = lane_->status.load(std::memory_order_acquire); // Running, most likely
156 LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
157 LUA_ASSERT(L_, lane_->waiting_on == nullptr);
158 lane_->waiting_on = &waitingOn_;
159 lane_->status.store(Lane::Waiting, std::memory_order_release);
160 }
161
162 // wait until the final target date by small increments, interrupting regularly so that we can check for cancel requests,
163 // in case some timing issue caused a cancel request to be issued, and the condvar signalled, before we actually wait for it
164 auto const [_forceTryAgain, _until_check_cancel] = std::invoke([until_, wakePeriod = linda_->getWakePeriod()] {
165 auto _until_check_cancel{ std::chrono::time_point<std::chrono::steady_clock>::max() };
166 if (wakePeriod.count() > 0.0f) {
167 _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(wakePeriod);
168 }
169 bool const _forceTryAgain{ _until_check_cancel < until_ };
170 return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : until_);
171 });
172
173 // operation can't complete: wake when it is signalled to be possible, or when timeout is reached
174 std::unique_lock<std::mutex> _guard{ keeper_->mutex, std::adopt_lock };
175 std::cv_status const _status{ waitingOn_.wait_until(_guard, _until_check_cancel) };
176 _guard.release(); // we don't want to unlock the mutex on exit!
177 bool const _try_again{ _forceTryAgain || (_status == std::cv_status::no_timeout) }; // detect spurious wakeups
178 if (lane_ != nullptr) {
179 lane_->waiting_on = nullptr;
180 lane_->status.store(_prev_status, std::memory_order_release);
181 }
182 return _try_again;
183 }
184
185 // #############################################################################################
150 186
151 // the implementation for linda:receive() and linda:receive_batched() 187 // the implementation for linda:receive() and linda:receive_batched()
152 static int ReceiveInternal(lua_State* const L_, bool const batched_) 188 static int ReceiveInternal(lua_State* const L_, bool const batched_)
@@ -201,6 +237,7 @@ namespace {
201 _cancel = (_cancel != CancelRequest::None) 237 _cancel = (_cancel != CancelRequest::None)
202 ? _cancel 238 ? _cancel
203 : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None); 239 : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
240
204 // if user wants to cancel, or looped because of a timeout, the call returns without sending anything 241 // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
205 if (!_try_again || _cancel != CancelRequest::None) { 242 if (!_try_again || _cancel != CancelRequest::None) {
206 _pushed.emplace(0); 243 _pushed.emplace(0);
@@ -227,38 +264,7 @@ namespace {
227 } 264 }
228 265
229 // nothing received, wait until timeout or signalled that we should try again 266 // nothing received, wait until timeout or signalled that we should try again
230 { 267 _try_again = WaitInternal(L_, _lane, _linda, _keeper, _linda->writeHappened, _until);
231 Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
232 if (_lane != nullptr) {
233 // change status of lane to "waiting"
234 _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
235 LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
236 LUA_ASSERT(L_, _lane->waiting_on == nullptr);
237 _lane->waiting_on = &_linda->writeHappened;
238 _lane->status.store(Lane::Waiting, std::memory_order_release);
239 }
240
241 // wait until the final target date by small increments, interrupting regularly so that we can check for cancel requests,
242 // in case some timing issue caused a cancel request to be issued, and the condvar signalled, before we actually wait for it
243 auto const [_forceTryAgain, _until_check_cancel] = std::invoke([_until, wakePeriod = _linda->getWakePeriod()] {
244 auto _until_check_cancel{ std::chrono::time_point<std::chrono::steady_clock>::max() };
245 if (wakePeriod.count() > 0.0f) {
246 _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(wakePeriod);
247 }
248 bool const _forceTryAgain{ _until_check_cancel < _until };
249 return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : _until);
250 });
251
252 // not enough data to read: wakeup when data was sent, or when timeout is reached
253 std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
254 std::cv_status const _status{ _linda->writeHappened.wait_until(_guard, _until_check_cancel) };
255 _guard.release(); // we don't want to unlock the mutex on exit!
256 _try_again = _forceTryAgain || (_status == std::cv_status::no_timeout); // detect spurious wakeups
257 if (_lane != nullptr) {
258 _lane->waiting_on = nullptr;
259 _lane->status.store(_prev_status, std::memory_order_release);
260 }
261 }
262 } 268 }
263 STACK_CHECK(_K, 0); 269 STACK_CHECK(_K, 0);
264 270
@@ -273,7 +279,7 @@ namespace {
273 if (_nbPushed == 0) { 279 if (_nbPushed == 0) {
274 // not enough data in the linda slot to fulfill the request, return nil, "timeout" 280 // not enough data in the linda slot to fulfill the request, return nil, "timeout"
275 lua_pushnil(L_); 281 lua_pushnil(L_);
276 luaG_pushstring(L_, "timeout"); 282 luaW_pushstring(L_, "timeout");
277 return 2; 283 return 2;
278 } 284 }
279 return _nbPushed; 285 return _nbPushed;
@@ -332,6 +338,7 @@ Keeper* Linda::acquireKeeper() const
332 Keeper* const _keeper{ whichKeeper() }; 338 Keeper* const _keeper{ whichKeeper() };
333 if (_keeper) { 339 if (_keeper) {
334 _keeper->mutex.lock(); 340 _keeper->mutex.lock();
341 keeperOperationCount.fetch_add(1, std::memory_order_seq_cst);
335 } 342 }
336 return _keeper; 343 return _keeper;
337} 344}
@@ -344,16 +351,16 @@ Linda* Linda::CreateTimerLinda(lua_State* const L_)
344 // Initialize 'timerLinda'; a common Linda object shared by all states 351 // Initialize 'timerLinda'; a common Linda object shared by all states
345 lua_pushcfunction(L_, LG_linda); // L_: lanes.linda 352 lua_pushcfunction(L_, LG_linda); // L_: lanes.linda
346 lua_createtable(L_, 0, 3); // L_: lanes.linda {} 353 lua_createtable(L_, 0, 3); // L_: lanes.linda {}
347 luaG_pushstring(L_, "lanes-timer"); // L_: lanes.linda {} "lanes-timer" 354 luaW_pushstring(L_, "lanes-timer"); // L_: lanes.linda {} "lanes-timer"
348 luaG_setfield(L_, StackIndex{ -2 }, std::string_view{ "name" }); // L_: lanes.linda { .name="lanes-timer" } 355 luaW_setfield(L_, StackIndex{ -2 }, std::string_view{ "name" }); // L_: lanes.linda { .name="lanes-timer" }
349 lua_pushinteger(L_, 0); // L_: lanes.linda { .name="lanes-timer" } 0 356 lua_pushinteger(L_, 0); // L_: lanes.linda { .name="lanes-timer" } 0
350 luaG_setfield(L_, StackIndex{ -2 }, std::string_view{ "group" }); // L_: lanes.linda { .name="lanes-timer" .group = 0 } 357 luaW_setfield(L_, StackIndex{ -2 }, std::string_view{ "group" }); // L_: lanes.linda { .name="lanes-timer" .group = 0 }
351 // note that wake_period is not set (will default to the value in the universe) 358 // note that wake_period is not set (will default to the value in the universe)
352 lua_call(L_, 1, 1); // L_: linda 359 lua_call(L_, 1, 1); // L_: linda
353 STACK_CHECK(L_, 1); 360 STACK_CHECK(L_, 1);
354 361
355 // Proxy userdata contents is only a 'DeepPrelude*' pointer 362 // Proxy userdata contents is only a 'DeepPrelude*' pointer
356 auto const _timerLinda{ *luaG_tofulluserdata<Linda*>(L_, kIdxTop) }; 363 auto const _timerLinda{ *luaW_tofulluserdata<Linda*>(L_, kIdxTop) };
357 // increment refcount so that this linda remains alive as long as the universe exists. 364 // increment refcount so that this linda remains alive as long as the universe exists.
358 _timerLinda->refcount.fetch_add(1, std::memory_order_relaxed); 365 _timerLinda->refcount.fetch_add(1, std::memory_order_relaxed);
359 lua_pop(L_, 1); // L_: 366 lua_pop(L_, 1); // L_:
@@ -416,7 +423,6 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
416 // doing LindaFactory::deleteDeepObjectInternal -> keeper_call(clear) 423 // doing LindaFactory::deleteDeepObjectInternal -> keeper_call(clear)
417 lua_gc(L_, LUA_GCSTOP, 0); 424 lua_gc(L_, LUA_GCSTOP, 0);
418 425
419 LUA_ASSERT_CODE(auto const _koip{ _linda->startKeeperOperation(L_) });
420 // if we didn't do anything wrong, the keeper stack should be clean 426 // if we didn't do anything wrong, the keeper stack should be clean
421 LUA_ASSERT(L_, lua_gettop(_K) == 0); 427 LUA_ASSERT(L_, lua_gettop(_K) == 0);
422 428
@@ -424,7 +430,7 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
424 lua_pushcfunction(L_, f_); 430 lua_pushcfunction(L_, f_);
425 lua_insert(L_, 1); 431 lua_insert(L_, 1);
426 // do a protected call 432 // do a protected call
427 LuaError const _rc{ lua_pcall(L_, lua_gettop(L_) - 1, LUA_MULTRET, 0) }; 433 LuaError const _rc{ ToLuaError(lua_pcall(L_, lua_gettop(L_) - 1, LUA_MULTRET, 0)) };
428 // whatever happens, the keeper state stack must be empty when we are done 434 // whatever happens, the keeper state stack must be empty when we are done
429 lua_settop(_K, 0); 435 lua_settop(_K, 0);
430 436
@@ -446,7 +452,7 @@ int Linda::ProtectedCall(lua_State* const L_, lua_CFunction const f_)
446 452
447void Linda::pushCancelString(lua_State* L_) const 453void Linda::pushCancelString(lua_State* L_) const
448{ 454{
449 luaG_pushstring(L_, cancelStatus == Status::Cancelled ? "cancelled" : "active"); 455 luaW_pushstring(L_, cancelStatus == Status::Cancelled ? "cancelled" : "active");
450} 456}
451 457
452// ################################################################################################# 458// #################################################################################################
@@ -455,6 +461,7 @@ void Linda::releaseKeeper(Keeper* const keeper_) const
455{ 461{
456 if (keeper_) { // can be nullptr if we tried to acquire during shutdown 462 if (keeper_) { // can be nullptr if we tried to acquire during shutdown
457 assert(keeper_ == whichKeeper()); 463 assert(keeper_ == whichKeeper());
464 keeperOperationCount.fetch_sub(1, std::memory_order_seq_cst);
458 keeper_->mutex.unlock(); 465 keeper_->mutex.unlock();
459 } 466 }
460} 467}
@@ -498,7 +505,7 @@ void Linda::setName(std::string_view const& name_)
498LUAG_FUNC(linda_cancel) 505LUAG_FUNC(linda_cancel)
499{ 506{
500 Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) }; 507 Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
501 std::string_view const _who{ luaG_optstring(L_, StackIndex{ 2 }, "both") }; 508 std::string_view const _who{ luaW_optstring(L_, StackIndex{ 2 }, "both") };
502 // make sure we got 2 arguments: the linda and the cancellation mode 509 // make sure we got 2 arguments: the linda and the cancellation mode
503 luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments"); 510 luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments");
504 511
@@ -586,13 +593,13 @@ static int linda_index_string(lua_State* L_)
586 Linda* const _linda{ ToLinda<false>(L_, kIdxSelf) }; 593 Linda* const _linda{ ToLinda<false>(L_, kIdxSelf) };
587 LUA_ASSERT(L_, lua_gettop(L_) == 2); // L_: linda "key" 594 LUA_ASSERT(L_, lua_gettop(L_) == 2); // L_: linda "key"
588 595
589 std::string_view const _keystr{ luaG_tostring(L_, kIdxKey) }; 596 std::string_view const _keystr{ luaW_tostring(L_, kIdxKey) };
590 lua_settop(L_, 2); // keep only our original arguments on the stack 597 lua_settop(L_, 2); // keep only our original arguments on the stack
591 598
592 // look in metatable first 599 // look in metatable first
593 lua_getmetatable(L_, kIdxSelf); // L_: linda "key" mt 600 lua_getmetatable(L_, kIdxSelf); // L_: linda "key" mt
594 lua_replace(L_, -3); // L_: mt "key" 601 lua_replace(L_, -3); // L_: mt "key"
595 if (luaG_rawget(L_, StackIndex{ -2 }) != LuaType::NIL) { // found something? // L_: mt value 602 if (luaW_rawget(L_, StackIndex{ -2 }) != LuaType::NIL) { // found something? // L_: mt value
596 return 1; // done 603 return 1; // done
597 } 604 }
598 605
@@ -612,12 +619,12 @@ static LUAG_FUNC(linda_index)
612 static constexpr StackIndex kIdxKey{ 2 }; 619 static constexpr StackIndex kIdxKey{ 2 };
613 LUA_ASSERT(L_, lua_gettop(L_) == 2); 620 LUA_ASSERT(L_, lua_gettop(L_) == 2);
614 621
615 switch (luaG_type(L_, kIdxKey)) { 622 switch (luaW_type(L_, kIdxKey)) {
616 case LuaType::STRING: 623 case LuaType::STRING:
617 return linda_index_string(L_); // stack modification is undefined, returned value is at the top 624 return linda_index_string(L_); // stack modification is undefined, returned value is at the top
618 625
619 default: // unknown key 626 default: // unknown key
620 raise_luaL_error(L_, "Unsupported linda indexing key type %s", luaG_typename(L_, kIdxKey).data()); 627 raise_luaL_error(L_, "Unsupported linda indexing key type %s", luaW_typename(L_, kIdxKey).data());
621 } 628 }
622} 629}
623 630
@@ -760,7 +767,7 @@ LUAG_FUNC(linda_limit)
760 int const _nargs{ lua_gettop(L_) }; 767 int const _nargs{ lua_gettop(L_) };
761 luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments"); 768 luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments");
762 // make sure we got a numeric limit, or "unlimited", (or nothing) 769 // make sure we got a numeric limit, or "unlimited", (or nothing)
763 bool const _unlimited{ luaG_tostring(L_, StackIndex{ 3 }) == "unlimited" }; 770 bool const _unlimited{ luaW_tostring(L_, StackIndex{ 3 }) == "unlimited" };
764 LindaLimit const _val{ _unlimited ? std::numeric_limits<LindaLimit::type>::max() : static_cast<LindaLimit::type>(luaL_optinteger(L_, 3, 0)) }; 771 LindaLimit const _val{ _unlimited ? std::numeric_limits<LindaLimit::type>::max() : static_cast<LindaLimit::type>(luaL_optinteger(L_, 3, 0)) };
765 if (_val < 0) { 772 if (_val < 0) {
766 raise_luaL_argerror(L_, StackIndex{ 3 }, "limit must be >= 0"); 773 raise_luaL_argerror(L_, StackIndex{ 3 }, "limit must be >= 0");
@@ -771,23 +778,23 @@ LUAG_FUNC(linda_limit)
771 KeeperCallResult _pushed; 778 KeeperCallResult _pushed;
772 if (_linda->cancelStatus == Linda::Active) { 779 if (_linda->cancelStatus == Linda::Active) {
773 if (_unlimited) { 780 if (_unlimited) {
774 LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaG_tostring(L_, StackIndex{ 3 }) == "unlimited"); 781 LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaW_tostring(L_, StackIndex{ 3 }) == "unlimited");
775 // inside the Keeper, unlimited is signified with a -1 limit (can't use nil because of nil kNilSentinel conversions!) 782 // inside the Keeper, unlimited is signified with a -1 limit (can't use nil because of nil kNilSentinel conversions!)
776 lua_pop(L_, 1); // L_: linda slot 783 lua_pop(L_, 1); // L_: linda slot
777 lua_pushinteger(L_, -1); // L_: linda slot nil 784 lua_pushinteger(L_, -1); // L_: linda slot nil
778 } 785 }
779 Keeper* const _keeper{ _linda->whichKeeper() }; 786 Keeper* const _keeper{ _linda->whichKeeper() };
780 _pushed = keeper_call(_keeper->K, KEEPER_API(limit), L_, _linda, StackIndex{ 2 }); 787 _pushed = keeper_call(_keeper->K, KEEPER_API(limit), L_, _linda, StackIndex{ 2 });
781 LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 2) && luaG_type(L_, kIdxTop) == LuaType::STRING); 788 LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 2) && luaW_type(L_, kIdxTop) == LuaType::STRING);
782 if (_nargs == 3) { // 3 args: setting the limit 789 if (_nargs == 3) { // 3 args: setting the limit
783 // changing the limit: no error, boolean value saying if we should wake blocked writer threads 790 // changing the limit: no error, boolean value saying if we should wake blocked writer threads
784 LUA_ASSERT(L_, luaG_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN); // L_: bool string 791 LUA_ASSERT(L_, luaW_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN); // L_: bool string
785 if (lua_toboolean(L_, -2)) { 792 if (lua_toboolean(L_, -2)) {
786 _linda->readHappened.notify_all(); // To be done from within the 'K' locking area 793 _linda->readHappened.notify_all(); // To be done from within the 'K' locking area
787 } 794 }
788 } else { // 2 args: reading the limit 795 } else { // 2 args: reading the limit
789 // reading the limit: a number >=0 or "unlimited" 796 // reading the limit: a number >=0 or "unlimited"
790 LUA_ASSERT(L_, luaG_type(L_, StackIndex{ -2 }) == LuaType::NUMBER || luaG_tostring(L_, StackIndex{ -2 }) == "unlimited"); 797 LUA_ASSERT(L_, luaW_type(L_, StackIndex{ -2 }) == LuaType::NUMBER || luaW_tostring(L_, StackIndex{ -2 }) == "unlimited");
791 } 798 }
792 } else { // linda is cancelled 799 } else { // linda is cancelled
793 // do nothing and return nil,lanes.cancel_error 800 // do nothing and return nil,lanes.cancel_error
@@ -843,7 +850,7 @@ LUAG_FUNC(linda_restrict)
843 int const _nargs{ lua_gettop(L_) }; 850 int const _nargs{ lua_gettop(L_) };
844 luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments"); 851 luaL_argcheck(L_, _nargs == 2 || _nargs == 3, 2, "wrong number of arguments");
845 // make sure we got a known restrict mode, (or nothing) 852 // make sure we got a known restrict mode, (or nothing)
846 std::string_view const _mode{ luaG_tostring(L_, StackIndex{ 3 }) }; 853 std::string_view const _mode{ luaW_tostring(L_, StackIndex{ 3 }) };
847 if (!_mode.empty() && (_mode != "none" && _mode != "set/get" && _mode != "send/receive")) { 854 if (!_mode.empty() && (_mode != "none" && _mode != "set/get" && _mode != "send/receive")) {
848 raise_luaL_argerror(L_, StackIndex{ 3 }, "unknown restrict mode"); 855 raise_luaL_argerror(L_, StackIndex{ 3 }, "unknown restrict mode");
849 } 856 }
@@ -855,7 +862,7 @@ LUAG_FUNC(linda_restrict)
855 Keeper* const _keeper{ _linda->whichKeeper() }; 862 Keeper* const _keeper{ _linda->whichKeeper() };
856 _pushed = keeper_call(_keeper->K, KEEPER_API(restrict), L_, _linda, StackIndex{ 2 }); 863 _pushed = keeper_call(_keeper->K, KEEPER_API(restrict), L_, _linda, StackIndex{ 2 });
857 // we should get a single return value: the string describing the previous restrict mode 864 // we should get a single return value: the string describing the previous restrict mode
858 LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 1) && luaG_type(L_, kIdxTop) == LuaType::STRING); 865 LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 1) && luaW_type(L_, kIdxTop) == LuaType::STRING);
859 } else { // linda is cancelled 866 } else { // linda is cancelled
860 // do nothing and return nil,lanes.cancel_error 867 // do nothing and return nil,lanes.cancel_error
861 lua_pushnil(L_); 868 lua_pushnil(L_);
@@ -916,6 +923,7 @@ LUAG_FUNC(linda_send)
916 _cancel = (_cancel != CancelRequest::None) 923 _cancel = (_cancel != CancelRequest::None)
917 ? _cancel 924 ? _cancel
918 : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None); 925 : ((_linda->cancelStatus == Linda::Cancelled) ? CancelRequest::Soft : CancelRequest::None);
926
919 // if user wants to cancel, or looped because of a timeout, the call returns without sending anything 927 // if user wants to cancel, or looped because of a timeout, the call returns without sending anything
920 if (!_try_again || _cancel != CancelRequest::None) { 928 if (!_try_again || _cancel != CancelRequest::None) {
921 _pushed.emplace(0); 929 _pushed.emplace(0);
@@ -948,38 +956,7 @@ LUAG_FUNC(linda_send)
948 } 956 }
949 957
950 // storage limit hit, wait until timeout or signalled that we should try again 958 // storage limit hit, wait until timeout or signalled that we should try again
951 { 959 _try_again = WaitInternal(L_, _lane, _linda, _keeper, _linda->readHappened, _until);
952 Lane::Status _prev_status{ Lane::Error }; // prevent 'might be used uninitialized' warnings
953 if (_lane != nullptr) {
954 // change status of lane to "waiting"
955 _prev_status = _lane->status.load(std::memory_order_acquire); // Running, most likely
956 LUA_ASSERT(L_, _prev_status == Lane::Running); // but check, just in case
957 LUA_ASSERT(L_, _lane->waiting_on == nullptr);
958 _lane->waiting_on = &_linda->readHappened;
959 _lane->status.store(Lane::Waiting, std::memory_order_release);
960 }
961
962 // wait until the final target date by small increments, interrupting regularly so that we can check for cancel requests,
963 // in case some timing issue caused a cancel request to be issued, and the condvar signalled, before we actually wait for it
964 auto const [_forceTryAgain, _until_check_cancel] = std::invoke([_until, wakePeriod = _linda->getWakePeriod()] {
965 auto _until_check_cancel{ std::chrono::time_point<std::chrono::steady_clock>::max() };
966 if (wakePeriod.count() > 0.0f) {
967 _until_check_cancel = std::chrono::steady_clock::now() + std::chrono::duration_cast<std::chrono::steady_clock::duration>(wakePeriod);
968 }
969 bool const _forceTryAgain{ _until_check_cancel < _until };
970 return std::make_tuple(_forceTryAgain, _forceTryAgain ? _until_check_cancel : _until);
971 });
972
973 // could not send because no room: wait until some data was read before trying again, or until timeout is reached
974 std::unique_lock<std::mutex> _guard{ _keeper->mutex, std::adopt_lock };
975 std::cv_status const status{ _linda->readHappened.wait_until(_guard, _until_check_cancel) };
976 _guard.release(); // we don't want to unlock the mutex on exit!
977 _try_again = _forceTryAgain || (status == std::cv_status::no_timeout); // detect spurious wakeups
978 if (_lane != nullptr) {
979 _lane->waiting_on = nullptr;
980 _lane->status.store(_prev_status, std::memory_order_release);
981 }
982 }
983 } 960 }
984 STACK_CHECK(_K, 0); 961 STACK_CHECK(_K, 0);
985 962
@@ -1005,7 +982,7 @@ LUAG_FUNC(linda_send)
1005 } else { 982 } else {
1006 // not enough room in the Linda slot to fulfill the request, return nil, "timeout" 983 // not enough room in the Linda slot to fulfill the request, return nil, "timeout"
1007 lua_pushnil(L_); 984 lua_pushnil(L_);
1008 luaG_pushstring(L_, "timeout"); 985 luaW_pushstring(L_, "timeout");
1009 return 2; 986 return 2;
1010 } 987 }
1011 } 988 }
@@ -1040,7 +1017,7 @@ LUAG_FUNC(linda_set)
1040 if (kRestrictedChannel.equals(L_, kIdxTop)) { 1017 if (kRestrictedChannel.equals(L_, kIdxTop)) {
1041 raise_luaL_error(L_, "Key is restricted"); 1018 raise_luaL_error(L_, "Key is restricted");
1042 } 1019 }
1043 LUA_ASSERT(L_, _pushed.value() == 2 && luaG_type(L_, kIdxTop) == LuaType::STRING && luaG_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN); 1020 LUA_ASSERT(L_, _pushed.value() == 2 && luaW_type(L_, kIdxTop) == LuaType::STRING && luaW_type(L_, StackIndex{ -2 }) == LuaType::BOOLEAN);
1044 1021
1045 if (_has_data) { 1022 if (_has_data) {
1046 // we put some data in the slot, tell readers that they should wake 1023 // we put some data in the slot, tell readers that they should wake
@@ -1102,7 +1079,7 @@ LUAG_FUNC(linda_towatch)
1102LUAG_FUNC(linda_wake) 1079LUAG_FUNC(linda_wake)
1103{ 1080{
1104 Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) }; 1081 Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
1105 std::string_view const _who{ luaG_optstring(L_, StackIndex{ 2 }, "both") }; 1082 std::string_view const _who{ luaW_optstring(L_, StackIndex{ 2 }, "both") };
1106 // make sure we got 2 arguments: the linda and the wake targets 1083 // make sure we got 2 arguments: the linda and the wake targets
1107 luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments"); 1084 luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments");
1108 1085
@@ -1178,8 +1155,8 @@ LUAG_FUNC(linda)
1178 if (lua_isnil(L_, kIdxTop)) { 1155 if (lua_isnil(L_, kIdxTop)) {
1179 lua_pop(L_, 1); 1156 lua_pop(L_, 1);
1180 lua_pushnumber(L_, _U->lindaWakePeriod.count()); 1157 lua_pushnumber(L_, _U->lindaWakePeriod.count());
1181 } else if (luaG_type(L_, kIdxTop) == LuaType::STRING) { 1158 } else if (luaW_type(L_, kIdxTop) == LuaType::STRING) {
1182 if (luaG_tostring(L_, kIdxTop) != "never") { 1159 if (luaW_tostring(L_, kIdxTop) != "never") {
1183 luaL_argerror(L_, 1, "invalid wake_period"); 1160 luaL_argerror(L_, 1, "invalid wake_period");
1184 } else { 1161 } else {
1185 lua_pop(L_, 1); 1162 lua_pop(L_, 1);
@@ -1201,7 +1178,7 @@ LUAG_FUNC(linda)
1201 1178
1202#if LUA_VERSION_NUM >= 504 // to-be-closed support starts with Lua 5.4 1179#if LUA_VERSION_NUM >= 504 // to-be-closed support starts with Lua 5.4
1203 lua_getfield(L_, 1, "close_handler"); // L_: {} wake_period group close_handler 1180 lua_getfield(L_, 1, "close_handler"); // L_: {} wake_period group close_handler
1204 LuaType const _handlerType{ luaG_type(L_, kIdxTop) }; 1181 LuaType const _handlerType{ luaW_type(L_, kIdxTop) };
1205 if (_handlerType == LuaType::NIL) { 1182 if (_handlerType == LuaType::NIL) {
1206 lua_pop(L_, 1); // L_: {} wake_period group 1183 lua_pop(L_, 1); // L_: {} wake_period group
1207 } else if (_handlerType == LuaType::USERDATA || _handlerType == LuaType::TABLE) { 1184 } else if (_handlerType == LuaType::USERDATA || _handlerType == LuaType::TABLE) {
@@ -1212,7 +1189,7 @@ LUAG_FUNC(linda)
1212 } 1189 }
1213#endif // LUA_VERSION_NUM >= 504 1190#endif // LUA_VERSION_NUM >= 504
1214 1191
1215 auto const _nameType{ luaG_getfield(L_, StackIndex{ 1 }, "name") }; // L_: {} wake_period group [close_handler] name 1192 auto const _nameType{ luaW_getfield(L_, StackIndex{ 1 }, "name") }; // L_: {} wake_period group [close_handler] name
1216 luaL_argcheck(L_, _nameType == LuaType::NIL || _nameType == LuaType::STRING, 1, "name is not a string"); 1193 luaL_argcheck(L_, _nameType == LuaType::NIL || _nameType == LuaType::STRING, 1, "name is not a string");
1217 lua_replace(L_, 1); // L_: name wake_period group [close_handler] 1194 lua_replace(L_, 1); // L_: name wake_period group [close_handler]
1218 } 1195 }