Revert "Remove signaled threads on TaskScheduler destruction on Windo… (#2568)

…ws (#2544)"

This reverts commit 7534e22.
Tests are intermittently failing on the CI. Will investigate further
#### Reference Issues/PRs
<!--Example: Fixes #1234. See also #3456.-->

#### What does this implement or fix?

#### 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,12 +19,16 @@ std::once_flag TaskScheduler::init_flag_;
 std::once_flag TaskScheduler::shutdown_flag_;
 bool TaskScheduler::forked_ = false;
 
-void TaskScheduler::stop_active_threads() {
-    std::call_once(shutdown_flag_, [] {
-        if (instance_) {
-            instance()->stop();
-        }
-    });
+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::reattach_instance() {
@@ -50,7 +54,6 @@ void TaskScheduler::init(){
 }
 
 TaskSchedulerPtrWrapper::~TaskSchedulerPtrWrapper() {
-    ptr_->stop_orphaned_threads();
     delete ptr_;
 }
 

cpp/arcticdb/async/task_scheduler.hpp

@@ -101,18 +101,6 @@ 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(), 0) == WAIT_OBJECT_0;
-            if (is_signaled) {
-                SchedulerType::threadList_.remove(thread);
-            }
-        }
-#endif
-    }
 };
 
 struct CGroupValues {
@@ -227,7 +215,8 @@ class TaskScheduler {
 
     static TaskScheduler* instance();
     static void reattach_instance();
-    static void stop_active_threads();
+    static void destroy_instance();
+    static void stop_and_destroy();
     static bool forked_;
     static bool is_forked();
     static void set_forked(bool);
@@ -240,8 +229,8 @@ class TaskScheduler {
 
     void stop() {
         ARCTICDB_DEBUG(log::schedule(), "Stopping task scheduler");
-        io_exec_.stop();
         cpu_exec_.stop();
+        io_exec_.stop();
     }
 
     void set_active_threads(size_t n) {
@@ -286,11 +275,6 @@ 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::stop_active_threads();
+    async::TaskScheduler::destroy_instance();
     storage::mongo::MongoInstance::destroy_instance();
     ModuleData::destroy_instance();
 }

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

@@ -5,9 +5,6 @@
 
 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
@@ -28,28 +25,3 @@ 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=10)
-
-    if proc.is_alive():
-        proc.terminate()
-        pytest.fail("os._exit did not exit within 5 seconds")
-
-    assert proc.exitcode == 0