1 files changed, 52 insertions, 40 deletions

diff --git a/src/linda.cpp b/src/linda.cpp
index 5ee4768..ea1410e 100644
--- a/src/linda.cpp
+++ b/src/linda.cpp
@@ -61,8 +61,8 @@ class Linda : public DeepPrelude // Deep userdata MUST start with this header
 
     public:
 
-    SIGNAL_T read_happened;
-    SIGNAL_T write_happened;
+    std::condition_variable m_read_happened;
+    std::condition_variable m_write_happened;
     Universe* const U; // the universe this linda belongs to
     uintptr_t const group; // a group to control keeper allocation between lindas
     CancelRequest simulate_cancel{ CancelRequest::None };
@@ -81,17 +81,11 @@ class Linda : public DeepPrelude // Deep userdata MUST start with this header
     : U{ U_ }
     , group{ group_ << KEEPER_MAGIC_SHIFT }
     {
-        SIGNAL_INIT(&read_happened);
-        SIGNAL_INIT(&write_happened);
-
         setName(name_, len_);
     }
 
     ~Linda()
     {
-        // There aren't any lanes waiting on these lindas, since all proxies have been gc'ed. Right?
-        SIGNAL_FREE(&read_happened);
-        SIGNAL_FREE(&write_happened);
         if (std::holds_alternative<AllocatedName>(m_name))
         {
             AllocatedName& name = std::get<AllocatedName>(m_name);
@@ -216,15 +210,19 @@ LUAG_FUNC(linda_protected_call)
 LUAG_FUNC(linda_send)
 {
     Linda* const linda{ lua_toLinda<false>(L, 1) };
-    time_d timeout{ -1.0 };
+    std::chrono::time_point<std::chrono::steady_clock> until{ std::chrono::time_point<std::chrono::steady_clock>::max() };
     int key_i{ 2 }; // index of first key, if timeout not there
 
     if (lua_type(L, 2) == LUA_TNUMBER) // we don't want to use lua_isnumber() because of autocoercion
     {
-        timeout = SIGNAL_TIMEOUT_PREPARE(lua_tonumber(L, 2));
+        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);
+        }
         ++key_i;
     }
-    else if (lua_isnil(L, 2)) // alternate explicit "no timeout" by passing nil before the key
+    else if (lua_isnil(L, 2)) // alternate explicit "infinite timeout" by passing nil before the key
     {
         ++key_i;
     }
@@ -266,6 +264,7 @@ LUAG_FUNC(linda_send)
         lua_State* const KL{ K ? K->L : nullptr };
         if (KL == nullptr)
             return 0;
+
         STACK_CHECK_START_REL(KL, 0);
         for (bool try_again{ true };;)
         {
@@ -295,12 +294,12 @@ LUAG_FUNC(linda_send)
             if (ret)
             {
                 // Wake up ALL waiting threads
-                SIGNAL_ALL(&linda->write_happened);
+                linda->m_write_happened.notify_all();
                 break;
             }
 
             // instant timout to bypass the wait syscall
-            if (timeout == 0.0)
+            if (std::chrono::steady_clock::now() >= until)
             {
                 break; /* no wait; instant timeout */
             }
@@ -314,14 +313,17 @@ LUAG_FUNC(linda_send)
                     prev_status = lane->status; // RUNNING, most likely
                     ASSERT_L(prev_status == RUNNING); // but check, just in case
                     lane->status = WAITING;
-                    ASSERT_L(lane->waiting_on == nullptr);
-                    lane->waiting_on = &linda->read_happened;
+                    ASSERT_L(lane->m_waiting_on == nullptr);
+                    lane->m_waiting_on = &linda->m_read_happened;
                 }
                 // could not send because no room: wait until some data was read before trying again, or until timeout is reached
-                try_again = SIGNAL_WAIT(&linda->read_happened, &K->keeper_cs, timeout);
+                std::unique_lock<std::mutex> keeper_lock{ K->m_mutex, std::adopt_lock };
+                std::cv_status const status{ linda->m_read_happened.wait_until(keeper_lock, until) };
+                keeper_lock.release(); // we don't want to release the lock!
+                try_again = (status == std::cv_status::no_timeout); // detect spurious wakeups
                 if (lane != nullptr)
                 {
-                    lane->waiting_on = nullptr;
+                    lane->m_waiting_on = nullptr;
                     lane->status = prev_status;
                 }
             }
@@ -369,21 +371,24 @@ static constexpr UniqueKey BATCH_SENTINEL{ 0x2DDFEE0968C62AA7ull };
 LUAG_FUNC(linda_receive)
 {
     Linda* const linda{ lua_toLinda<false>(L, 1) };
-
-    time_d timeout{ -1.0 };
-    int key_i{ 2 };
+    std::chrono::time_point<std::chrono::steady_clock> until{ std::chrono::time_point<std::chrono::steady_clock>::max() };
+    int key_i{ 2 }; // index of first key, if timeout not there
 
     if (lua_type(L, 2) == LUA_TNUMBER) // we don't want to use lua_isnumber() because of autocoercion
     {
-        timeout = SIGNAL_TIMEOUT_PREPARE(lua_tonumber(L, 2));
+        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);
+        }
         ++key_i;
     }
-    else if (lua_isnil(L, 2)) // alternate explicit "no timeout" by passing nil before the key
+    else if (lua_isnil(L, 2)) // alternate explicit "infinite timeout" by passing nil before the key
     {
         ++key_i;
     }
 
-    keeper_api_t keeper_receive;
+    keeper_api_t selected_keeper_receive{ nullptr };
     int expected_pushed_min{ 0 }, expected_pushed_max{ 0 };
     // are we in batched mode?
     BATCH_SENTINEL.pushKey(L);
@@ -396,7 +401,7 @@ LUAG_FUNC(linda_receive)
         // make sure the keys are of a valid type
         check_key_types(L, key_i, key_i);
         // receive multiple values from a single slot
-        keeper_receive = KEEPER_API(receive_batched);
+        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);
         expected_pushed_max = (int) luaL_optinteger(L, key_i + 2, expected_pushed_min);
@@ -413,17 +418,20 @@ LUAG_FUNC(linda_receive)
         // make sure the keys are of a valid type
         check_key_types(L, key_i, lua_gettop(L));
         // receive a single value, checking multiple slots
-        keeper_receive = KEEPER_API(receive);
+        selected_keeper_receive = KEEPER_API(receive);
         // we expect a single (value, key) pair of returned values
         expected_pushed_min = expected_pushed_max = 2;
     }
 
     Lane* const lane{ LANE_POINTER_REGKEY.readLightUserDataValue<Lane>(L) };
     Keeper* const K{ which_keeper(linda->U->keepers, linda->hashSeed()) };
-    if (K == nullptr)
+    lua_State* const KL{ K ? K->L : nullptr };
+    if (KL == nullptr)
         return 0;
+
     CancelRequest cancel{ CancelRequest::None };
     int pushed{ 0 };
+    STACK_CHECK_START_REL(KL, 0);
     for (bool try_again{ true };;)
     {
         if (lane != nullptr)
@@ -439,7 +447,7 @@ LUAG_FUNC(linda_receive)
         }
 
         // all arguments of receive() but the first are passed to the keeper's receive function
-        pushed = keeper_call(linda->U, K->L, keeper_receive, L, linda, key_i);
+        pushed = keeper_call(linda->U, KL, selected_keeper_receive, L, linda, key_i);
         if (pushed < 0)
         {
             break;
@@ -451,11 +459,11 @@ LUAG_FUNC(linda_receive)
             keeper_toggle_nil_sentinels(L, lua_gettop(L) - pushed, LookupMode::FromKeeper);
             // To be done from within the 'K' locking area
             //
-            SIGNAL_ALL(&linda->read_happened);
+            linda->m_read_happened.notify_all();
             break;
         }
 
-        if (timeout == 0.0)
+        if (std::chrono::steady_clock::now() >= until)
         {
             break; /* instant timeout */
         }
@@ -469,18 +477,22 @@ LUAG_FUNC(linda_receive)
                 prev_status = lane->status; // RUNNING, most likely
                 ASSERT_L(prev_status == RUNNING); // but check, just in case
                 lane->status = WAITING;
-                ASSERT_L(lane->waiting_on == nullptr);
-                lane->waiting_on = &linda->write_happened;
+                ASSERT_L(lane->m_waiting_on == nullptr);
+                lane->m_waiting_on = &linda->m_write_happened;
             }
             // not enough data to read: wakeup when data was sent, or when timeout is reached
-            try_again = SIGNAL_WAIT(&linda->write_happened, &K->keeper_cs, timeout);
+            std::unique_lock<std::mutex> keeper_lock{ K->m_mutex, std::adopt_lock };
+            std::cv_status const status{ linda->m_write_happened.wait_until(keeper_lock, until) };
+            keeper_lock.release(); // we don't want to release the lock!
+            try_again = (status == std::cv_status::no_timeout); // detect spurious wakeups
             if (lane != nullptr)
             {
-                lane->waiting_on = nullptr;
+                lane->m_waiting_on = nullptr;
                 lane->status = prev_status;
             }
         }
     }
+    STACK_CHECK(KL, 0);
 
     if (pushed < 0)
     {
@@ -537,13 +549,13 @@ LUAG_FUNC(linda_set)
             if (has_value)
             {
                 // we put some data in the slot, tell readers that they should wake
-                SIGNAL_ALL(&linda->write_happened); // To be done from within the 'K' locking area
+                linda->m_write_happened.notify_all(); // To be done from within the 'K' locking area
             }
             if (pushed == 1)
             {
                 // the key was full, but it is no longer the case, tell writers they should wake
                 ASSERT_L(lua_type(L, -1) == LUA_TBOOLEAN && lua_toboolean(L, -1) == 1);
-                SIGNAL_ALL(&linda->read_happened); // To be done from within the 'K' locking area
+                linda->m_read_happened.notify_all(); // To be done from within the 'K' locking area
             }
         }
     }
@@ -648,7 +660,7 @@ LUAG_FUNC( linda_limit)
         if( pushed == 1)
         {
             ASSERT_L( lua_type( L, -1) == LUA_TBOOLEAN && lua_toboolean( L, -1) == 1);
-            SIGNAL_ALL( &linda->read_happened); // To be done from within the 'K' locking area
+            linda->m_read_happened.notify_all(); // To be done from within the 'K' locking area
         }
     }
     else // linda is cancelled
@@ -678,8 +690,8 @@ LUAG_FUNC(linda_cancel)
     linda->simulate_cancel = CancelRequest::Soft;
     if (strcmp(who, "both") == 0) // tell everyone writers to wake up
     {
-        SIGNAL_ALL(&linda->write_happened);
-        SIGNAL_ALL(&linda->read_happened);
+        linda->m_write_happened.notify_all();
+        linda->m_read_happened.notify_all();
     }
     else if (strcmp(who, "none") == 0) // reset flag
     {
@@ -687,11 +699,11 @@ LUAG_FUNC(linda_cancel)
     }
     else if (strcmp(who, "read") == 0) // tell blocked readers to wake up
     {
-        SIGNAL_ALL(&linda->write_happened);
+        linda->m_write_happened.notify_all();
     }
     else if (strcmp(who, "write") == 0) // tell blocked writers to wake up
     {
-        SIGNAL_ALL(&linda->read_happened);
+        linda->m_read_happened.notify_all();
     }
     else
     {