Remove signaled threads on TaskScheduler destruction on Windows with timeout (#2582)

#### Reference Issues/PRs
<!--Example: Fixes #1234. See also #3456.-->

#### What does this implement or fix?
This reapplies the reverted PR #2544 with two changes that previously
caused flaky tests:
1. Increase test timeout from 5 to 10 seconds. Previously some
successful tests took up to 4+ seconds so it is reasonable to increase
the limit.
2. Increase the timeout of windows api call. Previously the function
immediatly returned which caused some rare cases where the function was
wrongly returning that the thread is not signaled and thus causing the
tests to fail.

I've tested both before the change and after.
Before:
* Out of 100 repetitions of all param combinations of the test ~ 4 were
failing

After
* Out of ~200~ 1000 repetitions of all parameter combinations - no
failures


For easier review here is the diff with the reverted PR:
https://github.com/man-group/ArcticDB/compare/threads_already_dead..threads_already_dead_after_revert

For safety on the CI I additionally increased the test timeout to 15
seconds.
#### Any other comments?

#### Checklist

<details>
  <summary>
   Checklist for code changes...
  </summary>
 
- [ ] Have you updated the relevant docstrings, documentation and
copyright notice?
- [ ] Is this contribution tested against [all ArcticDB's
features](../docs/mkdocs/docs/technical/contributing.md)?
- [ ] Do all exceptions introduced raise appropriate [error
messages](https://docs.arcticdb.io/error_messages/)?
 - [ ] Are API changes highlighted in the PR description?
- [ ] Is the PR labelled as enhancement or bug so it appears in
autogenerated release notes?
</details>

<!--
Thanks for contributing a Pull Request to ArcticDB! Please ensure you
have taken a look at:
- ArcticDB's Code of Conduct:
https://github.com/man-group/ArcticDB/blob/master/CODE_OF_CONDUCT.md
- ArcticDB's Contribution Licensing:
https://github.com/man-group/ArcticDB/blob/master/docs/mkdocs/docs/technical/contributing.md#contribution-licensing
-->

Author: ognyanstoimenov

cpp/arcticdb/async/task_scheduler.cpp

@@ -19,16 +19,12 @@ std::once_flag TaskScheduler::init_flag_;
 std::once_flag TaskScheduler::shutdown_flag_;
 bool TaskScheduler::forked_ = false;
 
-void TaskScheduler::destroy_instance() {
-    std::call_once(TaskScheduler::shutdown_flag_, &TaskScheduler::stop_and_destroy);
-}
-
-void TaskScheduler::stop_and_destroy() {
-    if(TaskScheduler::instance_) {
-        TaskScheduler::instance()->stop();
-
-        TaskScheduler::instance_.reset();
-    }
+void TaskScheduler::stop_active_threads() {
+    std::call_once(shutdown_flag_, [] {
+        if (instance_) {
+            instance()->stop();
+        }
+    });
 }
 
 void TaskScheduler::reattach_instance() {
@@ -54,6 +50,7 @@ void TaskScheduler::init(){
 }
 
 TaskSchedulerPtrWrapper::~TaskSchedulerPtrWrapper() {
+    ptr_->stop_orphaned_threads();
     delete ptr_;
 }
 

cpp/arcticdb/async/task_scheduler.hpp

@@ -101,6 +101,18 @@ struct SchedulerWrapper : public SchedulerType {
     void ensure_active_threads() {
         SchedulerType::ensureActiveThreads();
     }
+
+    void stop_orphaned_threads() {
+#ifdef _WIN32
+        const auto to_remove_copy = SchedulerType::threadList_.get();
+        for (const auto& thread : to_remove_copy) {
+            const bool is_signaled = WaitForSingleObject(thread->handle.native_handle(), 5000) == WAIT_OBJECT_0;
+            if (is_signaled) {
+                SchedulerType::threadList_.remove(thread);
+            }
+        }
+#endif
+    }
 };
 
 struct CGroupValues {
@@ -215,8 +227,7 @@ class TaskScheduler {
 
     static TaskScheduler* instance();
     static void reattach_instance();
-    static void destroy_instance();
-    static void stop_and_destroy();
+    static void stop_active_threads();
     static bool forked_;
     static bool is_forked();
     static void set_forked(bool);
@@ -229,8 +240,8 @@ class TaskScheduler {
 
     void stop() {
         ARCTICDB_DEBUG(log::schedule(), "Stopping task scheduler");
-        cpu_exec_.stop();
         io_exec_.stop();
+        cpu_exec_.stop();
     }
 
     void set_active_threads(size_t n) {
@@ -275,6 +286,11 @@ class TaskScheduler {
         return io_thread_count_;
     }
 
+    void stop_orphaned_threads() {
+        io_exec_.stop_orphaned_threads();
+        cpu_exec_.stop_orphaned_threads();
+    }
+
 private:
     std::string cgroup_folder_;
     size_t cpu_thread_count_;

cpp/arcticdb/util/global_lifetimes.cpp

@@ -65,7 +65,7 @@ std::shared_ptr<ModuleData> ModuleData::instance_;
 std::once_flag ModuleData::init_flag_;
 
 void shutdown_globals() {
-    async::TaskScheduler::destroy_instance();
+    async::TaskScheduler::stop_active_threads();
     storage::mongo::MongoInstance::destroy_instance();
     ModuleData::destroy_instance();
 }

python/tests/stress/arcticdb/version_store/test_deallocation.py

@@ -5,6 +5,9 @@
 
 As of the Change Date specified in that file, in accordance with the Business Source License, use of this software will be governed by the Apache License, version 2.0.
 """
+import os
+from multiprocessing import Process
+
 import pandas as pd
 import numpy as np
 import pytest
@@ -25,3 +28,28 @@ def test_many_version_store(basic_store_factory):
         version_store.write(symbol, df2)
         vit = version_store.read(symbol)
         assert_frame_equal(vit.data, df2)
+
+
+def killed_worker(lib, io_threads, cpu_threads):
+    if io_threads:
+        lib.delete_snapshot("snap")
+    if cpu_threads:
+        lib.read("sym")
+    os._exit(0)
+
+@pytest.mark.parametrize("io_threads_spawned_in_child", [True, False])
+@pytest.mark.parametrize("cpu_threads_spawned_in_child", [True, False])
+def test_os_exit_exits_within_timeout(lmdb_storage, lib_name, io_threads_spawned_in_child, cpu_threads_spawned_in_child):
+    lib = lmdb_storage.create_arctic().create_library(lib_name)
+    df = pd.DataFrame()
+    lib.write("sym", df)
+    lib.snapshot("snap")
+    proc = Process(target=killed_worker, args=(lib, io_threads_spawned_in_child, cpu_threads_spawned_in_child))
+    proc.start()
+    proc.join(timeout=15)
+
+    if proc.is_alive():
+        proc.terminate()
+        pytest.fail("os._exit did not exit within 15 seconds")
+
+    assert proc.exitcode == 0