Merge pull request #364 from sandstorm-io/fix-thread-detach

Fix that Thread::detach() prematurely deleted objects the thread might be using.

c++/src/kj/thread-test.c++

+// Copyright (c) 2016 Sandstorm Development Group, Inc. and contributors
+// Licensed under the MIT License:
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+#include "thread.h"
+#include "test.h"
+#include <atomic>
+
+#if _WIN32
+#define NOGDI
+#include <windows.h>
+#undef NOGDI
+#else
+#include <unistd.h>
+#endif
+
+namespace kj {
+namespace {
+
+#if _WIN32
+inline void delay() { Sleep(10); }
+#else
+inline void delay() { usleep(10000); }
+#endif
+
+KJ_TEST("detaching thread doesn't delete function") {
+  struct Functor {
+    // Functor that sets *b = true on destruction, not counting moves.
+
+    std::atomic<bool>* destroyed;
+    const std::atomic<bool>* canExit;
+
+    Functor(std::atomic<bool>* destroyed, const std::atomic<bool>* canExit)
+        : destroyed(destroyed), canExit(canExit) {}
+
+    ~Functor() {
+      if (destroyed != nullptr) *destroyed = true;
+    }
+    KJ_DISALLOW_COPY(Functor);
+    Functor(Functor&& other): destroyed(other.destroyed), canExit(other.canExit) {
+      other.destroyed = nullptr;
+    }
+    Functor& operator=(Functor&& other) = delete;
+
+    void operator()() {
+      while (!*canExit) delay();
+    }
+  };
+
+  std::atomic<bool> destroyed(false);
+  std::atomic<bool> canExit(false);
+  Functor f(&destroyed, &canExit);
+
+  kj::Thread(kj::mv(f)).detach();
+
+  // detach() should not have destroyed the function.
+  KJ_ASSERT(!destroyed);
+
+  // Delay a bit to make sure the thread has had time to start up, and then make sure the function
+  // still isn't destroyed.
+  delay();
+  delay();
+  KJ_ASSERT(!destroyed);
+
+  // Notify the thread that it's safe to exit.
+  canExit = true;
+  while (!destroyed) {
+    delay();
+  }
+}
+
+}  // namespace
+}  // namespace kj

c++/src/kj/thread.c++

@@ -33,8 +33,8 @@ namespace kj {
 
 #if _WIN32
 
-Thread::Thread(Function<void()> func): func(kj::mv(func)) {
-  threadHandle = CreateThread(nullptr, 0, &runThread, this, 0, nullptr);
+Thread::Thread(Function<void()> func): state(new ThreadState { kj::mv(func), nullptr, 2 }) {
+  threadHandle = CreateThread(nullptr, 0, &runThread, state, 0, nullptr);
   KJ_ASSERT(threadHandle != nullptr, "CreateThread failed.");
 }
 
@@ -54,23 +54,24 @@ void Thread::detach() {
 }
 
 DWORD Thread::runThread(void* ptr) {
-  Thread* thread = reinterpret_cast<Thread*>(ptr);
+  ThreadState* state = reinterpret_cast<ThreadState*>(ptr);
   KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
-    thread->func();
+    state->func();
   })) {
-    thread->exception = kj::mv(*exception);
+    state->exception = kj::mv(*exception);
   }
+  state->unref();
   return 0;
 }
 
 #else  // _WIN32
 
-Thread::Thread(Function<void()> func): func(kj::mv(func)) {
+Thread::Thread(Function<void()> func): state(new ThreadState { kj::mv(func), nullptr, 2 }) {
   static_assert(sizeof(threadId) >= sizeof(pthread_t),
                 "pthread_t is larger than a long long on your platform.  Please port.");
 
   int pthreadResult = pthread_create(reinterpret_cast<pthread_t*>(&threadId),
-                                     nullptr, &runThread, this);
+                                     nullptr, &runThread, state);
   if (pthreadResult != 0) {
     KJ_FAIL_SYSCALL("pthread_create", pthreadResult);
   }
@@ -78,13 +79,17 @@ Thread::Thread(Function<void()> func): func(kj::mv(func)) {
 
 Thread::~Thread() noexcept(false) {
   if (!detached) {
+    KJ_DEFER(state->unref());
+
     int pthreadResult = pthread_join(*reinterpret_cast<pthread_t*>(&threadId), nullptr);
     if (pthreadResult != 0) {
       KJ_FAIL_SYSCALL("pthread_join", pthreadResult) { break; }
     }
 
-    KJ_IF_MAYBE(e, exception) {
-      kj::throwRecoverableException(kj::mv(*e));
+    KJ_IF_MAYBE(e, state->exception) {
+      Exception ecopy = kj::mv(*e);
+      state->exception = nullptr;  // don't complain of uncaught exception when deleting
+      kj::throwRecoverableException(kj::mv(ecopy));
     }
   }
 }
@@ -102,18 +107,37 @@ void Thread::detach() {
     KJ_FAIL_SYSCALL("pthread_detach", pthreadResult) { break; }
   }
   detached = true;
+  state->unref();
 }
 
 void* Thread::runThread(void* ptr) {
-  Thread* thread = reinterpret_cast<Thread*>(ptr);
+  ThreadState* state = reinterpret_cast<ThreadState*>(ptr);
   KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
-    thread->func();
+    state->func();
   })) {
-    thread->exception = kj::mv(*exception);
+    state->exception = kj::mv(*exception);
   }
+  state->unref();
   return nullptr;
 }
 
 #endif  // _WIN32, else
 
+void Thread::ThreadState::unref() {
+#if _MSC_VER
+  if (_InterlockedDecrement_rel(&refcount)) {
+    _ReadBarrier();
+#else
+  if (__atomic_sub_fetch(&refcount, 1, __ATOMIC_RELEASE) == 0) {
+    __atomic_thread_fence(__ATOMIC_ACQUIRE);
+#endif
+
+    KJ_IF_MAYBE(e, exception) {
+      KJ_LOG(ERROR, "uncaught exception thrown by detached thread", *e);
+    }
+
+    delete this;
+  }
+}
+
 }  // namespace kj

c++/src/kj/thread.h

@@ -55,13 +55,22 @@ public:
   //   the Thread object and the thread itself will need to share a refcounted object.
 
 private:
-  Function<void()> func;
+  struct ThreadState {
+    Function<void()> func;
+    kj::Maybe<kj::Exception> exception;
+
+    int refcount;
+    // Owned by the parent thread and the child thread.
+
+    void unref();
+  };
+  ThreadState* state;
+
 #if _WIN32
   void* threadHandle;
 #else
   unsigned long long threadId;  // actually pthread_t
 #endif
-  kj::Maybe<kj::Exception> exception;
   bool detached = false;
 
 #if _WIN32