schedulers: Fix std::thread clean-up race in newthread

In rare situations the old lifetime.add callback would cause
std::system_error to be thrown after unsubscribing due to
the std::thread being non-joinable.

Bug: 143745681
Change-Id: If64f3eef678ba5246baaca5be06e9d13c799922c

1 files changed, 11 insertions, 7 deletions

diff --git a/Rx/v2/src/rxcpp/schedulers/rx-newthread.hpp b/Rx/v2/src/rxcpp/schedulers/rx-newthread.hpp
index 5145e92..7c93469 100644
--- a/Rx/v2/src/rxcpp/schedulers/rx-newthread.hpp
+++ b/Rx/v2/src/rxcpp/schedulers/rx-newthread.hpp
@@ -37,6 +37,16 @@ private:
 
             virtual ~new_worker_state()
             {
+                // Ensure that std::thread is no longer joinable,
+                // otherwise the destructor will call std::terminate.
+                if (!worker.joinable()) {
+                    return;
+                }
+                if (worker.get_id() != std::this_thread::get_id()) {
+                    worker.join();
+                } else {
+                    worker.detach();
+                }
             }
 
             explicit new_worker_state(composite_subscription cs)
@@ -76,13 +86,7 @@ private:
                 if (!keepAlive->q.empty()) std::terminate();
                 keepAlive->wake.notify_one();
 
-                if (keepAlive->worker.joinable() && keepAlive->worker.get_id() != std::this_thread::get_id()) {
-                    guard.unlock();
-                    keepAlive->worker.join();
-                }
-                else {
-                    keepAlive->worker.detach();
-                }
+                // ~new_worker_state cleans up the std::thread
             });
 
             state->worker = tf([keepAlive](){