Fix race condition with ConditionVariable

Mutex#sleep acts like an atomic Mutex#unlock + Thread#sleep.

If some other thread sneaks in the middle of that and calls
Thread#wakeup, the wakeup can be lost because we're not quite sleeping
yet. `m_wakeup` is the "condition" for our std::condition_variable that
allows us to check if we should immediately wake up.

include/natalie/thread_object.hpp

@@ -124,7 +124,7 @@ class ThreadObject : public Object {
     Value run(Env *);
     Value wakeup(Env *);
 
-    Value sleep(Env *, float);
+    Value sleep(Env *, float, Thread::MutexObject * = nullptr);
 
     void set_value(Value value) { m_value = value; }
     Value value(Env *);
@@ -318,6 +318,7 @@ class ThreadObject : public Object {
 
     // This condition variable is used to wake a sleeping thread,
     // i.e. a thread where Kernel#sleep has been called.
+    bool m_wakeup { false };
     std::condition_variable m_sleep_cond;
     std::mutex m_sleep_lock;
 

src/thread/mutex_object.cpp

@@ -29,8 +29,7 @@ Value MutexObject::lock(Env *env) {
 
 Value MutexObject::sleep(Env *env, Value timeout) {
     if (!timeout || timeout->is_nil()) {
-        unlock(env);
-        ThreadObject::current()->sleep(env, -1.0);
+        ThreadObject::current()->sleep(env, -1.0, this);
         lock(env);
         return this;
     }
@@ -43,9 +42,8 @@ Value MutexObject::sleep(Env *env, Value timeout) {
     if (timeout_int < 0)
         env->raise("ArgumentError", "timeout must be positive");
 
-    unlock(env);
     const auto timeout_float = timeout->is_float() ? static_cast<float>(timeout->as_float()->to_double()) : static_cast<float>(timeout_int);
-    ThreadObject::current()->sleep(env, timeout_float);
+    ThreadObject::current()->sleep(env, timeout_float, this);
     lock(env);
 
     return Value::integer(timeout_int);

src/thread_object.cpp

@@ -3,6 +3,7 @@
 #include <signal.h>
 
 #include "natalie.hpp"
+#include "natalie/thread/mutex_object.hpp"
 #include "natalie/thread_object.hpp"
 
 static void set_stack_for_thread(Natalie::ThreadObject *thread_object) {
@@ -397,13 +398,14 @@ Value ThreadObject::wakeup(Env *env) {
 
     {
         std::unique_lock sleep_lock { m_sleep_lock };
+        m_wakeup = true;
         m_sleep_cond.notify_one();
     }
 
     return this;
 }
 
-Value ThreadObject::sleep(Env *env, float timeout) {
+Value ThreadObject::sleep(Env *env, float timeout, Thread::MutexObject *mutex_to_unlock) {
     timespec t_begin;
     if (::clock_gettime(CLOCK_MONOTONIC, &t_begin) < 0)
         env->raise_errno();
@@ -417,16 +419,23 @@ Value ThreadObject::sleep(Env *env, float timeout) {
         return Value::integer(elapsed);
     };
 
+    m_wakeup = false;
+    if (mutex_to_unlock)
+        mutex_to_unlock->unlock(env);
+
     if (timeout < 0.0) {
         {
             std::unique_lock sleep_lock { m_sleep_lock };
 
             check_exception(env);
 
-            Defer done_sleeping([] { ThreadObject::set_current_sleeping(false); });
-            ThreadObject::set_current_sleeping(true);
+            if (!m_wakeup) {
+                Defer done_sleeping([] { ThreadObject::set_current_sleeping(false); });
+                ThreadObject::set_current_sleeping(true);
 
-            m_sleep_cond.wait(sleep_lock);
+                m_sleep_cond.wait(sleep_lock);
+            }
+            m_wakeup = false;
         }
 
         check_exception(env);
@@ -440,10 +449,13 @@ Value ThreadObject::sleep(Env *env, float timeout) {
 
         check_exception(env);
 
-        Defer done_sleeping([] { ThreadObject::set_current_sleeping(false); });
-        ThreadObject::set_current_sleeping(true);
+        if (!m_wakeup) {
+            Defer done_sleeping([] { ThreadObject::set_current_sleeping(false); });
+            ThreadObject::set_current_sleeping(true);
 
-        m_sleep_cond.wait_for(sleep_lock, wait);
+            m_sleep_cond.wait_for(sleep_lock, wait);
+        }
+        m_wakeup = false;
     }
 
     check_exception(env);