linda:limit() and linda:set() return a second value, a string representing the fill status

6 files changed, 92 insertions, 46 deletions

diff --git a/docs/index.html b/docs/index.html
index 8e84fcb..148b5ab 100644
--- a/docs/index.html
+++ b/docs/index.html
@@ -1222,15 +1222,16 @@
 </p>
 
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
-        bool|(nil,[lanes.cancel_error|"timeout"]) = h:limit(key, &lt;limit&gt;)
-        &gt;limit&lt; = h:limit(key)
+        bool,string|(nil,[lanes.cancel_error|"timeout"]) = h:limit(key, &lt;limit&gt;)
+        (number|string),string = h:limit(key)
 </pre></td></tr></table>
 
 <p>
         By default, queue sizes are unlimited but limits can be enforced using the <tt>limit()</tt> method. This can be useful to balance execution speeds in a producer/consumer scenario.<br />
         A limit of 0 is allowed to block everything. <tt>"unlimited"</tt> removes the limit.<br />
-        If the key was full but the limit change added some room, <tt>limit()</tt> returns <tt>true</tt> and the Linda is signalled so that <tt>send()</tt>-blocked threads are awakened, else the return value is <tt>false</tt>.
-        If no limit is provided, <tt>limit()</tt> returns a single value, the current limit for the specified key.<br />
+        If the key was full but the limit change added some room, <tt>limit()</tt> first return value is <tt>true</tt> and the Linda is signalled so that <tt>send()</tt>-blocked threads are awakened, else the return value is <tt>false</tt>.
+        If no limit is provided, <tt>limit()</tt> first return value is the current limit for the specified key.<br />
+        The second returned value is a string representing the fill status relatively to the key's current limit (one of <tt>"over"</tt>, <tt>"under"</tt>, <tt>"exact"</tt>).
         Whether reading or writing, if the Linda is cancelled, <tt>limit()</tt> returns <tt>nil, lanes.cancel_error</tt>.
 </p>
 
@@ -1292,30 +1293,30 @@
 </p>
 
 <table border="1" bgcolor="#E0E0FF" cellpadding="10" style="width:50%"><tr><td><pre>
-        true|nil,lanes.cancel_error = linda_h:set(key [, val [, ...]])
+        (bool,string)|(nil,lanes.cancel_error) = linda_h:set(key [, val [, ...]])
 
-        [number,[val [, ...]]|nil,lanes.cancel_error] = linda_h:get(key [, count = 1])
+        (number,[val [, ...]])|(nil,lanes.cancel_error) = linda_h:get(key [, count = 1])
 </pre></td></tr></table>
 
 <p>
-        The table access methods are for accessing a slot without queuing or consuming. They can be used for making shared tables of storage among the lanes.<br />
-        Writing to a slot never blocks because it ignores the limit. It overwrites existing value and clears any possible queued entries.<br />
-        Reading doesn't block either because <tt>get()</tt> returns:
+        <tt>get()</tt>/<tt>set()</tt> and <tt>send()</tt>/<tt>receive()</tt> can be used together; reading a slot essentially peeks the next outcoming value of a queue.<br />
+        <tt>get()</tt>/<tt>set()</tt> are for accessing a key without queuing or consuming. They can be used for making shared tables of storage among the lanes.<br />
+        Writing to a key never blocks because it ignores the limit. It overwrites existing values and clears any possible queued entries.<br />
+</p>
+<p>
+        <tt>get()</tt> can read several values at once, and does not block. Return values ares:
         <ul>
                 <li><tt>nil, lanes.cancel_error</tt> in case of cancellation.</li>
                 <li><tt>number, val...</tt> where number is the actual count of items obtained from the linda (can be 0).</li>
         </ul>
-        Table access and <tt>send()</tt>/<tt>receive()</tt> can be used together; reading a slot essentially peeks the next outcoming value of a queue.
-</p>
-
-<p>
-        <tt>set()</tt> signals the Linda for write if a value is stored. If nothing special happens, <tt>set() </tt>returns nothing.<br />
-        If the key was full but the new data count of the key after <tt>set()</tt> is below its limit, <tt>set()</tt> returns <tt>true</tt> and the Linda is also signaled for read so that <tt>send()</tt>-blocked threads are awakened.
 </p>
-
 <p>
-        <tt>set()</tt> can write several values at the specified key, writing <tt>nil</tt> values is now possible, and clearing the contents at the specified key is done by not providing any value.<br />
-        Also, <tt>get()</tt> can read several values at once. If the key contains no data, <tt>get()</tt> returns no value. This can be used to separate the case when reading stored <tt>nil</tt> values.
+        <tt>set()</tt> can write several values at the specified key. Writing <tt>nil</tt> values is possible, and clearing the contents at the specified key is done by not providing any value.<br />
+        If <tt>set()</tt> actually stores data, the Linda is signalled for write, so that <tt>receive()</tt>-blocked Lanes are awakened.<br />
+        Clearing the contents of a non-existent key does not create it!<br />
+        If the key was full but the new data count of the key after <tt>set()</tt> is below its limit, <tt>set()</tt> first return value is <tt>true</tt> and the Linda is also signaled for read, so that <tt>send()</tt>-blocked Lanes are awakened.<br />
+        If the key was not already full, nothing additional happens, and <tt>set()</tt> first return value is <tt>false</tt>.<br />
+        The second return value is a string representing the fill status relatively to the key's current limit (one of <tt>"over"</tt>, <tt>"under"</tt>, <tt>"exact"</tt>).
 </p>
 
 <p>
diff --git a/src/keeper.cpp b/src/keeper.cpp
index ae09b37..cfeebbb 100644
--- a/src/keeper.cpp
+++ b/src/keeper.cpp
@@ -66,6 +66,10 @@ class KeyUD
     static constexpr int kContentsTableIndex{ 1 };
 
     public:
+    static constexpr std::string_view kUnder{ "under" };
+    static constexpr std::string_view kExact{ "exact" };
+    static constexpr std::string_view kOver{ "over" };
+
     int first{ 1 };
     int count{ 0 };
     LindaLimit limit{ -1 };
@@ -83,6 +87,8 @@ class KeyUD
     [[nodiscard]] int pop(KeeperState K_, int minCount_, int maxCount_); // keepercall_receive[_batched]
     void prepareAccess(KeeperState K_, int idx_) const;
     [[nodiscard]] bool push(KeeperState K_, int count_, bool enforceLimit_); // keepercall_send and keepercall_set
+    void pushFillStatus(KeeperState K_) const;
+    static void PushFillStatus(KeeperState K_, KeyUD const* key_);
     [[nodiscard]] bool reset(KeeperState K_);
 };
 
@@ -232,6 +238,35 @@ bool KeyUD::push(KeeperState const K_, int const count_, bool const enforceLimit
 
 // #################################################################################################
 
+void KeyUD::pushFillStatus(KeeperState const K_) const
+{
+    if (limit < 0) {
+        luaG_pushstring(K_, kUnder);
+        return;
+    }
+    int const _delta{limit - count};
+    if (_delta < 0) {
+        luaG_pushstring(K_, kOver);
+    } else if (_delta > 0) {
+        luaG_pushstring(K_, kUnder);
+    } else {
+        luaG_pushstring(K_, kExact);
+    }
+}
+
+// #################################################################################################
+
+void KeyUD::PushFillStatus(KeeperState const K_, KeyUD const* const key_)
+{
+    if (key_) {
+        key_->pushFillStatus(K_);                                                                  // _K: ... <fill status>
+    } else {
+        luaG_pushstring(K_, KeyUD::kUnder);                                                        // _K: ... "under"
+    }
+}
+
+// #################################################################################################
+
 // expects 'this' on top of the stack
 bool KeyUD::reset(KeeperState const K_)
 {
@@ -394,10 +429,11 @@ int keepercall_get(lua_State* const L_)
 // #################################################################################################
 
 // in: linda key [n|nil]
-// out: true or nil
+// out: boolean, <fill status: string>
 int keepercall_limit(lua_State* const L_)
 {
     KeeperState const _K{ L_ };
+    STACK_CHECK_START_ABS(_K, lua_gettop(_K));
     // no limit to set, means we read and return the current limit instead
     bool const _reading{ lua_gettop(_K) == 2 };
     LindaLimit const _limit{ static_cast<LindaLimit::type>(luaL_optinteger(_K, 3, -1)) }; // -1 if we read nil because the argument is absent
@@ -408,10 +444,12 @@ int keepercall_limit(lua_State* const L_)
     lua_rawget(_K, -3);                                                                            // _K: KeysDB key KeyUD|nil
     KeyUD* _key{ KeyUD::GetPtr(_K, -1) };
     if (_reading) {
+        // remove any clutter on the stack
+        lua_settop(_K, 0);                                                                         // _K:
         if (_key && _key->limit >= 0) {
-            lua_pushinteger(_K, _key->limit);                                                      // _K: KeysDB key KeyUD limit
+            lua_pushinteger(_K, _key->limit);                                                      // _K: limit
         } else { // if the key doesn't exist, it is unlimited by default
-            luaG_pushstring(_K, "unlimited");                                                      // _K: KeysDB key KeyUD "unlimited"
+            luaG_pushstring(_K, "unlimited");                                                      // _K: "unlimited"
         }
         // return a single value: the limit of the key
     } else {
@@ -426,7 +464,9 @@ int keepercall_limit(lua_State* const L_)
         // this is the case if we detect the key was full but it is no longer the case
         lua_pushboolean(_K, _key->changeLimit(_limit) ? 1 : 0);                                    // _K: bool
     }
-    return 1;
+    KeyUD::PushFillStatus(_K, _key);                                                               // _K: limit|bool <fill status>
+    STACK_CHECK(_K, 2);
+    return 2;
 }
 
 // #################################################################################################
@@ -538,12 +578,12 @@ int keepercall_set(lua_State* const L_)
     // retrieve KeysDB associated with the linda
     PushKeysDB(_K, 1);                                                                             // _K: linda key val... KeysDB
     lua_replace(_K, 1);                                                                            // _K: KeysDB key val...
+    lua_pushvalue(_K, 2);                                                                          // _K: KeysDB key val... key
+    lua_rawget(_K, 1);                                                                             // _K: KeysDB key val KeyUD|nil
+    KeyUD* _key{ KeyUD::GetPtr(_K, -1) };
 
-    if (lua_gettop(_K) == 2) { // no value to set                                                  // _K: KeysDB key
-        lua_pushvalue(_K, -1);                                                                     // _K: KeysDB key key
-        lua_rawget(_K, 1);                                                                         // _K: KeysDB key KeyUD|nil
+    if (lua_gettop(_K) == 3) { // no value to set                                                  // _K: KeysDB key KeyUD|nil
         // empty the KeyUD for the specified key: replace uservalue with a virgin table, reset counters, but leave limit unchanged!
-        KeyUD* const _key{ KeyUD::GetPtr(_K, -1) };
         if (_key != nullptr) { // might be nullptr if we set a nonexistent key to nil              // _K: KeysDB key KeyUD
             if (_key->limit < 0) { // KeyUD limit value is the default (unlimited): we can totally remove it
                 lua_pop(_K, 1);                                                                    // _K: KeysDB key
@@ -555,18 +595,17 @@ int keepercall_set(lua_State* const L_)
                 _should_wake_writers = _key->reset(_K);
             }
         }
+        lua_settop(_K, 0); // we are done, remove everything                                       // _K:
     } else { // set/replace contents stored at the specified key?
-        int const _count{ lua_gettop(_K) - 2 };                                                    // number of items we want to store
-        lua_pushvalue(_K, 2);                                                                      // _K: KeysDB key val... key
-        lua_rawget(_K, 1);                                                                         // _K: KeysDB key val... KeyUD|nil
-        KeyUD* _key{ KeyUD::GetPtr(_K, -1) };
-        if (_key == nullptr) { // can be nullptr if we store a value at a new key                  // KeysDB key val... nil
-            // no need to wake writers in that case, because a writer can't wait on an inexistent key
+        int const _count{ lua_gettop(_K) - 3 };  // number of items we want to store               // _K: KeysDB key val... KeyUD|nil
+        if (_key == nullptr) { // can be nullptr if we store a value at a new key                  // _K: KeysDB key val... nil
+            assert(lua_isnil(_K, -1));
             lua_pop(_K, 1);                                                                        // _K: KeysDB key val...
             _key = KeyUD::Create(KeeperState{ _K });                                               // _K: KeysDB key val... KeyUD
             lua_pushvalue(_K, 2);                                                                  // _K: KeysDB key val... KeyUD key
             lua_pushvalue(_K, -2);                                                                 // _K: KeysDB key val... KeyUD key KeyUD
             lua_rawset(_K, 1);                                                                     // _K: KeysDB key val... KeyUD
+            // no need to wake writers, because a writer can't wait on an inexistent key
         } else {                                                                                   // _K: KeysDB key val... KeyUD
             // the KeyUD exists, we just want to update its contents
             // we create room if the KeyUD was full but we didn't refill it to the brim with new data
@@ -575,10 +614,12 @@ int keepercall_set(lua_State* const L_)
         // replace the key with the KeyUD in the stack
         lua_replace(_K, -2 - _count);                                                              // _K: KeysDB KeyUD val...
         [[maybe_unused]] bool const _pushed{ _key->push(_K, _count, false) };                      // _K: KeysDB
+        lua_pop(_K, 1);                                                                            // _K:
     }
-    // stack isn't the same here depending on what we did before, but that's not a problem
-    lua_pushboolean(_K, _should_wake_writers ? 1 : 0);                                             // _K: ... bool
-    return 1;
+    assert(lua_gettop(_K) == 0);
+    lua_pushboolean(_K, _should_wake_writers ? 1 : 0);                                             // _K: bool
+    KeyUD::PushFillStatus(_K, _key);                                                               // _K: bool <fill status>
+    return 2;
 }
 
 // #################################################################################################
diff --git a/src/lane.cpp b/src/lane.cpp
index d4175c6..ba24af3 100644
--- a/src/lane.cpp
+++ b/src/lane.cpp
@@ -435,7 +435,7 @@ static constexpr RegistryUniqueKey kStackTraceRegKey{ 0x3F327747CACAA904ull };
 [[nodiscard]] static int lane_error(lua_State* L_)
 {
     // error message (any type)
-    STACK_CHECK_START_ABS(L_, 1); // L_: some_error
+    STACK_CHECK_START_ABS(L_, 1);                                                                  // L_: some_error
 
     // Don't do stack survey for cancelled lanes.
     //
diff --git a/src/lanes.lua b/src/lanes.lua
index 6a4f149..48ebeb6 100644
--- a/src/lanes.lua
+++ b/src/lanes.lua
@@ -792,7 +792,8 @@ local genatomic = function(linda_, key_, initial_val_)
         if val ~= cancel_error then
             val = val + (diff_ or 1.0)
             -- set() releases the lock by emptying queue
-            if linda_:set(key_, val) == cancel_error then
+            local _res, _err = linda_:set(key_, val)
+            if _err == cancel_error then
                 val = cancel_error
             end
         end
diff --git a/src/linda.cpp b/src/linda.cpp
index f4dd7e7..13627aa 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -477,22 +477,24 @@ LUAG_FUNC(linda_limit)
 
             KeeperCallResult _pushed;
             if (_linda->cancelRequest == CancelRequest::None) {
-                Keeper* const _keeper{ _linda->whichKeeper() };
                 if (_unlimited) {
                     LUA_ASSERT(L_, lua_gettop(L_) == 3 && luaG_tostring(L_, 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 key
                     lua_pushinteger(L_, -1);                                                       // L_: linda key nil
                 }
+                Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(limit), L_, _linda, 2);
-                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 1));
+                LUA_ASSERT(L_, _pushed.has_value() && (_pushed.value() == 2) && luaG_type(L_, -1) == LuaType::STRING);
                 if (_nargs == 3) { // 3 args: setting the limit
-                    LUA_ASSERT(L_, luaG_type(L_, -1) == LuaType::BOOLEAN); // changing the limit: no error, boolean value saying if we should wake blocked writer threads
-                    if (lua_toboolean(L_, -1)) {
+                    // changing the limit: no error, boolean value saying if we should wake blocked writer threads
+                    LUA_ASSERT(L_, luaG_type(L_, -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
-                    LUA_ASSERT(L_, luaG_type(L_, -1) == LuaType::NUMBER || luaG_tostring(L_, -1) == "unlimited"); // reading the limit: a number >=0 or "unlimited"
+                    // reading the limit: a number >=0 or "unlimited"
+                    LUA_ASSERT(L_, luaG_type(L_, -2) == LuaType::NUMBER || luaG_tostring(L_, -2) == "unlimited");
                 }
             } else { // linda is cancelled
                 // do nothing and return nil,lanes.cancel_error
@@ -500,7 +502,7 @@ LUAG_FUNC(linda_limit)
                 kCancelError.pushKey(L_);
                 _pushed.emplace(2);
             }
-            // propagate pushed boolean if any
+            // propagate returned values
             return _pushed.value();
         }
     };
@@ -807,7 +809,7 @@ LUAG_FUNC(linda_send)
 // #################################################################################################
 
 /*
- * [true|nil,lanes.cancel_error] = linda:set(key_num|str|bool|lightuserdata [, value [, ...]])
+ * (boolean,string)|(nil,lanes.cancel_error) = linda:set(key_num|str|bool|lightuserdata [, value [, ...]])
  *
  * Set one or more value to Linda. Ignores limits.
  *
@@ -827,13 +829,13 @@ LUAG_FUNC(linda_set)
                 Keeper* const _keeper{ _linda->whichKeeper() };
                 _pushed = keeper_call(_keeper->K, KEEPER_API(set), L_, _linda, 2);
                 if (_pushed.has_value()) { // no error?
-                    LUA_ASSERT(L_, _pushed.value() == 1 && luaG_type(L_, -1) == LuaType::BOOLEAN);
+                    LUA_ASSERT(L_, _pushed.value() == 2 && luaG_type(L_, -1) == LuaType::STRING && luaG_type(L_, -2) == LuaType::BOOLEAN);
 
                     if (_has_data) {
                         // we put some data in the slot, tell readers that they should wake
                         _linda->writeHappened.notify_all(); // To be done from within the 'K' locking area
                     }
-                    if (lua_toboolean(L_, -1)) {
+                    if (lua_toboolean(L_, -2)) {
                         // the key was full, but it is no longer the case, tell writers they should wake
                         _linda->readHappened.notify_all(); // To be done from within the 'K' locking area
                     }
diff --git a/tests/atomic.lua b/tests/atomic.lua
index 2de8f52..511a19e 100644
--- a/tests/atomic.lua
+++ b/tests/atomic.lua
@@ -9,6 +9,7 @@ local lanes = require "lanes"
 local linda= lanes.linda()
 local key= "$"
 
+-- TODO: test what happens when we cancel the linda
 local f= lanes.genatomic( linda, key, 5 )
 
 local v