New linda:wake() and linda.status

5 files changed, 143 insertions, 25 deletions

diff --git a/CHANGES b/CHANGES
index 3a6df10..b8dff6d 100644
--- a/CHANGES
+++ b/CHANGES
@@ -47,7 +47,9 @@ CHANGE 2: BGe 04-Jul-24
             - linda :receive(), :send(), :get(), :set(), :limit() return nil, error in case of problem. Returned values in case of success change too.
             - linda:limit() can be used to read the value if no new limit is provided.
             - New __close metamethod that calls any suitable handler that was provided at Linda creation.
-            - linda:dump outputs <key>.limit as 'unlimited' instead of -1 for unlimited keys.
+            - linda:dump() outputs <key>.limit as 'unlimited' instead of -1 for unlimited keys.
+            - linda:wake() can wake up threads waiting for a Linda without doing any I/O on it.
+            - linda.status reads the cancel status of the Linda.
             - Deep userdata are an acceptable key to send data into (for example, another linda).
         - Full userdata conversion:
             - __lanesconvert added.
diff --git a/docs/index.html b/docs/index.html
index a8eadbf..b95bc0e 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -70,7 +70,7 @@
                         </p>
 
                         <p>
-                                This document was revised on 24-Sep-24, and applies to version <tt>4.0.0</tt>.
+                                This document was revised on 25-Oct-24, and applies to version <tt>4.0.0</tt>.
                         </p>
                 </font>
         </center>
@@ -1405,16 +1405,26 @@
 
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
         void = linda_h:cancel("read"|"write"|"both"|"none")
+        "cancelled"|"active" = linda_h.status
 </pre></td></tr></table>
 
 <p>
-        Signals the linda so that lanes waiting for read, write, or both, wake up.
+        <tt>linda:cancel()</tt> signals the linda so that lanes waiting for read, write, or both, wake up.
         All linda operations (including <tt>get()</tt> and <tt>set()</tt>) will return <tt>lanes.cancel_error</tt> as when the calling lane is <a href="#cancelling">soft-cancelled</a> as long as the linda is marked as cancelled.<br />
         <tt>"none"</tt> reset the linda's cancel status, but doesn't signal it.<br />
+        <tt>linda.status</tt> reads the current cancel status.
         If not void, the lane's cancel status overrides the linda's cancel status.
 </p>
 
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
+        void = linda_h:wake("read"|"write"|"both")
+</pre></td></tr></table>
+
+<p>
+        Signals the linda so that lanes waiting for read, write, or both, wake up. Does not change the current Linda's cancel status. Can be used to break early out of a blocking wait.
+</p>
+
+<table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
         light userdata = h:deep()
 </pre></td></tr></table>
 
diff --git a/src/linda.cpp b/src/linda.cpp
index 15e7a2c..67526a7 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -233,6 +233,13 @@ 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");
+}
+
+// #################################################################################################
+
 void Linda::releaseKeeper(Keeper* const keeper_) const
 {
     if (keeper_) { // can be nullptr if we tried to acquire during shutdown
@@ -284,18 +291,20 @@ LUAG_FUNC(linda_cancel)
     // make sure we got 2 arguments: the linda and the cancellation mode
     luaL_argcheck(L_, lua_gettop(L_) <= 2, 2, "wrong number of arguments");
 
-    _linda->cancelRequest = CancelRequest::Soft;
-    if (_who == "both") { // tell everyone writers to wake up
+    if (_who == "both") { // tell everyone to wake up
+        _linda->cancelStatus = Linda::Status::Cancelled;
         _linda->writeHappened.notify_all();
         _linda->readHappened.notify_all();
     } else if (_who == "none") { // reset flag
-        _linda->cancelRequest = CancelRequest::None;
+        _linda->cancelStatus = Linda::Status::Active;
     } else if (_who == "read") { // tell blocked readers to wake up
+        _linda->cancelStatus = Linda::Status::Cancelled;
         _linda->writeHappened.notify_all();
     } else if (_who == "write") { // tell blocked writers to wake up
+        _linda->cancelStatus = Linda::Status::Cancelled;
         _linda->readHappened.notify_all();
     } else {
-        raise_luaL_error(L_, "unknown wake hint '%s'", _who);
+        raise_luaL_error(L_, "unknown wake hint '%s'", _who.data());
     }
     return 0;
 }
@@ -357,6 +366,53 @@ LUAG_FUNC(linda_concat)
 
 // #################################################################################################
 
+// If key is "status" return the linda cancel status
+static int linda_index_string(lua_State* L_)
+{
+    static constexpr StackIndex kIdxSelf{ 1 };
+    static constexpr StackIndex kIdxKey{ 2 };
+
+    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) };
+    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"
+    lua_rawget(L_, -2);                                                                            // L_: mt value
+    if (luaG_type(L_, kIdxTop) != LuaType::NIL) { // found something?
+        return 1; // done
+    }
+
+    lua_pop(L_, 2);                                                                                // L_:
+    if (_keystr == "status") {
+        _linda->pushCancelString(L_);                                                              // L_: "<status>"
+        return 1;
+    }
+    raise_luaL_error(L_, "unknown field '%s'", _keystr.data());
+}
+
+// #################################################################################################
+
+// linda:__index(key,usr) -> value
+static LUAG_FUNC(linda_index)
+{
+    static constexpr StackIndex kIdxKey{ 2 };
+    LUA_ASSERT(L_, lua_gettop(L_) == 2);
+
+    switch (luaG_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());
+    }
+}
+
+// #################################################################################################
+
 /*
  * [val] = linda_count( linda_ud, [key [, ...]])
  *
@@ -433,7 +489,7 @@ LUAG_FUNC(linda_get)
             CheckKeyTypes(L_, StackIndex{ 2 }, StackIndex{ 2 });
 
             KeeperCallResult _pushed;
-            if (_linda->cancelRequest == CancelRequest::None) {
+            if (_linda->cancelStatus == Linda::Active) {
                 Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(get), L_, _linda, StackIndex{ 2 });
             } else { // linda is cancelled
@@ -477,7 +533,7 @@ LUAG_FUNC(linda_limit)
             CheckKeyTypes(L_, StackIndex{ 2 }, StackIndex{ 2 });
 
             KeeperCallResult _pushed;
-            if (_linda->cancelRequest == CancelRequest::None) {
+            if (_linda->cancelStatus == Linda::Active) {
                 if (_unlimited) {
                     LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaG_tostring(L_, StackIndex{ 3 }) == "unlimited");
                     // inside the Keeper, unlimited is signified with a -1 limit (can't use nil because of nil kNilSentinel conversions!)
@@ -586,7 +642,9 @@ LUAG_FUNC(linda_receive)
                 if (_lane != nullptr) {
                     _cancel = _lane->cancelRequest;
                 }
-                _cancel = (_cancel != CancelRequest::None) ? _cancel : _linda->cancelRequest;
+                _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);
@@ -723,7 +781,9 @@ LUAG_FUNC(linda_send)
                     if (_lane != nullptr) {
                         _cancel = _lane->cancelRequest;
                     }
-                    _cancel = (_cancel != CancelRequest::None) ? _cancel : _linda->cancelRequest;
+                    _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);
@@ -826,7 +886,7 @@ LUAG_FUNC(linda_set)
             CheckKeyTypes(L_, StackIndex{ 2 }, StackIndex{ 2 });
 
             KeeperCallResult _pushed;
-            if (_linda->cancelRequest == CancelRequest::None) {
+            if (_linda->cancelStatus == Linda::Active) {
                 Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(set), L_, _linda, StackIndex{ 2 });
                 if (_pushed.has_value()) { // no error?
@@ -884,6 +944,33 @@ LUAG_FUNC(linda_towatch)
 
 // #################################################################################################
 
+/*
+ * (void) = linda_wake( linda_ud, "read"|"write"|"both")
+ *
+ * Signal linda so that waiting threads wake up as if their own lane was cancelled
+ */
+LUAG_FUNC(linda_wake)
+{
+    Linda* const _linda{ ToLinda<false>(L_, StackIndex{ 1 }) };
+    std::string_view const _who{ luaG_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");
+
+    if (_who == "both") { // tell everyone to wake up
+        _linda->writeHappened.notify_all();
+        _linda->readHappened.notify_all();
+    } else if (_who == "read") { // simulate a read to wake writers
+        _linda->writeHappened.notify_all();
+    } else if (_who == "write") { // simulate a write to wake readers
+        _linda->readHappened.notify_all();
+    } else {
+        raise_luaL_error(L_, "unknown wake hint '%s'", _who.data());
+    }
+    return 0;
+}
+
+// #################################################################################################
+
 namespace {
     namespace local {
         static luaL_Reg const sLindaMT[] = {
@@ -891,6 +978,7 @@ namespace {
             { "__close", LG_linda_close },
 #endif // LUA_VERSION_NUM >= 504
             { "__concat", LG_linda_concat },
+            { "__index", LG_linda_index },
             { "__tostring", LG_linda_tostring },
 #if HAVE_DECODA_SUPPORT()
             { "__towatch", LG_linda_towatch }, // Decoda __towatch support
@@ -904,6 +992,7 @@ namespace {
             { "receive", LG_linda_receive },
             { "send", LG_linda_send },
             { "set", LG_linda_set },
+            { "wake", LG_linda_wake },
             { nullptr, nullptr }
         };
     } // namespace local
@@ -985,7 +1074,7 @@ LUAG_FUNC(linda)
         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 }, 1);                                             // L_: name linda
+            lua_setiuservalue(L_, StackIndex{ 2 }, UserValueIndex{ 1 });                           // L_: name 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
diff --git a/src/linda.h b/src/linda.h
index c05fb14..02b0514 100644
--- a/src/linda.h
+++ b/src/linda.h
@@ -40,7 +40,15 @@ class Linda
         }
     };
 
+    enum class Status
+    {
+        Active,
+        Cancelled
+    };
+    using enum Status;
+
     private:
+
     static constexpr size_t kEmbeddedNameLength = 24;
     using EmbeddedName = std::array<char, kEmbeddedNameLength>;
     // depending on the name length, it is either embedded inside the Linda, or allocated separately
@@ -53,7 +61,7 @@ class Linda
     std::condition_variable writeHappened{};
     Universe* const U{ nullptr }; // the universe this linda belongs to
     KeeperIndex const keeperIndex{ -1 }; // the keeper associated to this linda
-    CancelRequest cancelRequest{ CancelRequest::None };
+    Status cancelStatus{ Status::Active };
 
     public:
     [[nodiscard]] static void* operator new(size_t size_, Universe* U_) noexcept { return U_->internalAllocator.alloc(size_); }
@@ -89,6 +97,7 @@ class Linda
     };
     void releaseKeeper(Keeper* keeper_) const;
     [[nodiscard]] static int ProtectedCall(lua_State* L_, lua_CFunction f_);
+    void pushCancelString(lua_State* L_) const;
     [[nodiscard]] KeeperOperationInProgress startKeeperOperation(lua_State* const L_) { return KeeperOperationInProgress{ *this, L_ }; };
     [[nodiscard]] Keeper* whichKeeper() const { return U->keepers.getKeeper(keeperIndex); }
 };
diff --git a/src/lindafactory.cpp b/src/lindafactory.cpp
index cb801dd..11e2cff 100644
--- a/src/lindafactory.cpp
+++ b/src/lindafactory.cpp
@@ -41,25 +41,33 @@ static constexpr std::string_view kLindaMetatableName{ "Linda" };
 
 void LindaFactory::createMetatable(lua_State* L_) const
 {
+    static constexpr std::string_view kIndex{ "__index" };
+
     STACK_CHECK_START_REL(L_, 0);
-    lua_newtable(L_);
-    // metatable is its own index
-    lua_pushvalue(L_, -1);
-    lua_setfield(L_, -2, "__index");
+    lua_newtable(L_);                                                                              // L_: mt
 
     // protect metatable from external access
-    luaG_pushstring(L_, kLindaMetatableName);
-    lua_setfield(L_, -2, "__metatable");
+    luaG_pushstring(L_, kLindaMetatableName);                                                      // L_: mt "<name>"
+    lua_setfield(L_, -2, "__metatable");                                                           // L_: mt
 
     // the linda functions
     luaG_registerlibfuncs(L_, mLindaMT);
 
     // some constants
-    kLindaBatched.pushKey(L_);
-    lua_setfield(L_, -2, "batched");
-
-    kNilSentinel.pushKey(L_);
-    lua_setfield(L_, -2, "null");
+    kLindaBatched.pushKey(L_);                                                                     // L_: mt kLindaBatched
+    lua_setfield(L_, -2, "batched");                                                               // L_: mt
+
+    kNilSentinel.pushKey(L_);                                                                      // L_: mt kNilSentinel
+    lua_setfield(L_, -2, "null");                                                                  // L_: mt
+
+    // if the metatable contains __index, leave it as is
+    if (luaG_getfield(L_, kIdxTop, kIndex) != LuaType::NIL) {                                      // L_: mt __index
+        lua_pop(L_, 1);                                                                            // L_: mt __index
+    } else {
+        // metatable is its own index
+        lua_pushvalue(L_, kIdxTop);                                                                // L_: mt mt
+        luaG_setfield(L_, StackIndex{ -2 }, kIndex);                                               // L_: mt
+    }
 
     STACK_CHECK(L_, 1);
 }