[K/N] Extract SingleThreadExecutor.

^KT-48537

Merge-request: KT-MR-5338
Merged-by: Alexander Shabalin <Alexander.Shabalin@jetbrains.com>

kotlin-native/runtime/src/gc/cms/cpp/ConcurrentMarkAndSweepTest.cpp

@@ -19,6 +19,7 @@
 #include "GlobalData.hpp"
 #include "ObjectOps.hpp"
 #include "ObjectTestSupport.hpp"
+#include "SingleThreadExecutor.hpp"
 #include "TestSupport.hpp"
 #include "ThreadData.hpp"
 
@@ -650,101 +651,65 @@ namespace {
 
 class Mutator : private Pinned {
 public:
-    Mutator() : thread_(&Mutator::RunLoop, this) {}
+    Mutator() : executor_(MakeSingleThreadExecutorWithContext<Context>()) {}
-
-    ~Mutator() {
-        {
-            std::unique_lock guard(queueMutex_);
-            shutdownRequested_ = true;
-        }
-        queueCV_.notify_one();
-        thread_.join();
-        RuntimeAssert(queue_.empty(), "The queue must be empty, has size=%zu", queue_.size());
-        RuntimeAssert(memory_ == nullptr, "Memory must have been deinitialized");
-        RuntimeAssert(stackRoots_.empty(), "Stack roots must be empty, has size=%zu", stackRoots_.size());
-        RuntimeAssert(globalRoots_.empty(), "Global roots must be empty, has size=%zu", globalRoots_.size());
-    }
 
     template <typename F>
     [[nodiscard]] std::future<void> Execute(F&& f) {
-        std::packaged_task<void()> task([this, f = std::forward<F>(f)]() { f(*memory_->memoryState()->GetThreadData(), *this); });
+        return executor_.Execute(
-        auto future = task.get_future();
+                [this, f = std::forward<F>(f)] { f(*executor_.thread().context().memory_->memoryState()->GetThreadData(), *this); });
-        {
-            std::unique_lock guard(queueMutex_);
-            queue_.push_back(std::move(task));
-        }
-        queueCV_.notify_one();
-        return future;
     }
 
     StackObjectHolder& AddStackRoot() {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddStackRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddStackRoot can only be called in the mutator thread");
-        auto holder = make_unique<StackObjectHolder>(*memory_->memoryState()->GetThreadData());
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<StackObjectHolder>(*context.memory_->memoryState()->GetThreadData());
         auto& holderRef = *holder;
-        stackRoots_.push_back(std::move(holder));
+        context.stackRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     StackObjectHolder& AddStackRoot(ObjHeader* object) {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddStackRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddStackRoot can only be called in the mutator thread");
+        auto& context = executor_.thread().context();
         auto holder = make_unique<StackObjectHolder>(object);
         auto& holderRef = *holder;
-        stackRoots_.push_back(std::move(holder));
+        context.stackRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     GlobalObjectHolder& AddGlobalRoot() {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddGlobalRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddGlobalRoot can only be called in the mutator thread");
-        auto holder = make_unique<GlobalObjectHolder>(*memory_->memoryState()->GetThreadData());
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<GlobalObjectHolder>(*context.memory_->memoryState()->GetThreadData());
         auto& holderRef = *holder;
-        globalRoots_.push_back(std::move(holder));
+        context.globalRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     GlobalObjectHolder& AddGlobalRoot(ObjHeader* object) {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddGlobalRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddGlobalRoot can only be called in the mutator thread");
-        auto holder = make_unique<GlobalObjectHolder>(*memory_->memoryState()->GetThreadData(), object);
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<GlobalObjectHolder>(*context.memory_->memoryState()->GetThreadData(), object);
         auto& holderRef = *holder;
-        globalRoots_.push_back(std::move(holder));
+        context.globalRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
-    KStdVector<ObjHeader*> Alive() { return ::Alive(*memory_->memoryState()->GetThreadData()); }
+    KStdVector<ObjHeader*> Alive() { return ::Alive(*executor_.thread().context().memory_->memoryState()->GetThreadData()); }
 
 private:
-    void RunLoop() {
+    struct Context {
-        memory_ = make_unique<ScopedMemoryInit>();
+        KStdUniquePtr<ScopedMemoryInit> memory_;
-        AssertThreadState(memory_->memoryState(), ThreadState::kRunnable);
+        KStdVector<KStdUniquePtr<StackObjectHolder>> stackRoots_;
+        KStdVector<KStdUniquePtr<GlobalObjectHolder>> globalRoots_;
 
-        while (true) {
+        Context() : memory_(make_unique<ScopedMemoryInit>()) {
-            std::packaged_task<void()> task;
+            // SingleThreadExecutor must work in the runnable state, so that GC does not collect between tasks.
-            {
+            AssertThreadState(memory_->memoryState(), ThreadState::kRunnable);
-                std::unique_lock guard(queueMutex_);
-                queueCV_.wait(guard, [this]() { return !queue_.empty() || shutdownRequested_; });
-                if (shutdownRequested_) {
-                    globalRoots_.clear();
-                    stackRoots_.clear();
-                    memory_.reset();
-                    return;
-                }
-                task = std::move(queue_.front());
-                queue_.pop_front();
-            }
-            task();
         }
-    }
+    };
 
-    KStdUniquePtr<ScopedMemoryInit> memory_;
+    SingleThreadExecutor<ThreadWithContext<Context>> executor_;
-
-    // TODO: Consider full runtime init instead, and interact with initialized worker
-    std::condition_variable queueCV_;
-    std::mutex queueMutex_;
-    KStdDeque<std::packaged_task<void()>> queue_;
-    bool shutdownRequested_ = false;
-    std::thread thread_;
-
-    KStdVector<KStdUniquePtr<GlobalObjectHolder>> globalRoots_;
-    KStdVector<KStdUniquePtr<StackObjectHolder>> stackRoots_;
 };
 
 } // namespace

kotlin-native/runtime/src/gc/stms/cpp/SameThreadMarkAndSweepTest.cpp

@@ -19,6 +19,7 @@
 #include "GlobalData.hpp"
 #include "ObjectOps.hpp"
 #include "ObjectTestSupport.hpp"
+#include "SingleThreadExecutor.hpp"
 #include "TestSupport.hpp"
 #include "ThreadData.hpp"
 
@@ -646,101 +647,65 @@ namespace {
 
 class Mutator : private Pinned {
 public:
-    Mutator() : thread_(&Mutator::RunLoop, this) {}
+    Mutator() : executor_(MakeSingleThreadExecutorWithContext<Context>()) {}
-
-    ~Mutator() {
-        {
-            std::unique_lock guard(queueMutex_);
-            shutdownRequested_ = true;
-        }
-        queueCV_.notify_one();
-        thread_.join();
-        RuntimeAssert(queue_.empty(), "The queue must be empty, has size=%zu", queue_.size());
-        RuntimeAssert(memory_ == nullptr, "Memory must have been deinitialized");
-        RuntimeAssert(stackRoots_.empty(), "Stack roots must be empty, has size=%zu", stackRoots_.size());
-        RuntimeAssert(globalRoots_.empty(), "Global roots must be empty, has size=%zu", globalRoots_.size());
-    }
 
     template <typename F>
     [[nodiscard]] std::future<void> Execute(F&& f) {
-        std::packaged_task<void()> task([this, f = std::forward<F>(f)]() { f(*memory_->memoryState()->GetThreadData(), *this); });
+        return executor_.Execute(
-        auto future = task.get_future();
+                [this, f = std::forward<F>(f)] { f(*executor_.thread().context().memory_->memoryState()->GetThreadData(), *this); });
-        {
-            std::unique_lock guard(queueMutex_);
-            queue_.push_back(std::move(task));
-        }
-        queueCV_.notify_one();
-        return future;
     }
 
     StackObjectHolder& AddStackRoot() {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddStackRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddStackRoot can only be called in the mutator thread");
-        auto holder = make_unique<StackObjectHolder>(*memory_->memoryState()->GetThreadData());
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<StackObjectHolder>(*context.memory_->memoryState()->GetThreadData());
         auto& holderRef = *holder;
-        stackRoots_.push_back(std::move(holder));
+        context.stackRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     StackObjectHolder& AddStackRoot(ObjHeader* object) {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddStackRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddStackRoot can only be called in the mutator thread");
+        auto& context = executor_.thread().context();
         auto holder = make_unique<StackObjectHolder>(object);
         auto& holderRef = *holder;
-        stackRoots_.push_back(std::move(holder));
+        context.stackRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     GlobalObjectHolder& AddGlobalRoot() {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddGlobalRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddGlobalRoot can only be called in the mutator thread");
-        auto holder = make_unique<GlobalObjectHolder>(*memory_->memoryState()->GetThreadData());
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<GlobalObjectHolder>(*context.memory_->memoryState()->GetThreadData());
         auto& holderRef = *holder;
-        globalRoots_.push_back(std::move(holder));
+        context.globalRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
     GlobalObjectHolder& AddGlobalRoot(ObjHeader* object) {
-        RuntimeAssert(std::this_thread::get_id() == thread_.get_id(), "AddGlobalRoot can only be called in the mutator thread");
+        RuntimeAssert(std::this_thread::get_id() == executor_.thread().get_id(), "AddGlobalRoot can only be called in the mutator thread");
-        auto holder = make_unique<GlobalObjectHolder>(*memory_->memoryState()->GetThreadData(), object);
+        auto& context = executor_.thread().context();
+        auto holder = make_unique<GlobalObjectHolder>(*context.memory_->memoryState()->GetThreadData(), object);
         auto& holderRef = *holder;
-        globalRoots_.push_back(std::move(holder));
+        context.globalRoots_.push_back(std::move(holder));
         return holderRef;
     }
 
-    KStdVector<ObjHeader*> Alive() { return ::Alive(*memory_->memoryState()->GetThreadData()); }
+    KStdVector<ObjHeader*> Alive() { return ::Alive(*executor_.thread().context().memory_->memoryState()->GetThreadData()); }
 
 private:
-    void RunLoop() {
+    struct Context {
-        memory_ = make_unique<ScopedMemoryInit>();
+        KStdUniquePtr<ScopedMemoryInit> memory_;
-        AssertThreadState(memory_->memoryState(), ThreadState::kRunnable);
+        KStdVector<KStdUniquePtr<StackObjectHolder>> stackRoots_;
+        KStdVector<KStdUniquePtr<GlobalObjectHolder>> globalRoots_;
 
-        while (true) {
+        Context() : memory_(make_unique<ScopedMemoryInit>()) {
-            std::packaged_task<void()> task;
+            // SingleThreadExecutor must work in the runnable state, so that GC does not collect between tasks.
-            {
+            AssertThreadState(memory_->memoryState(), ThreadState::kRunnable);
-                std::unique_lock guard(queueMutex_);
-                queueCV_.wait(guard, [this]() { return !queue_.empty() || shutdownRequested_; });
-                if (shutdownRequested_) {
-                    globalRoots_.clear();
-                    stackRoots_.clear();
-                    memory_.reset();
-                    return;
-                }
-                task = std::move(queue_.front());
-                queue_.pop_front();
-            }
-            task();
         }
-    }
+    };
 
-    KStdUniquePtr<ScopedMemoryInit> memory_;
+    SingleThreadExecutor<ThreadWithContext<Context>> executor_;
-
-    // TODO: Consider full runtime init instead, and interact with initialized worker
-    std::condition_variable queueCV_;
-    std::mutex queueMutex_;
-    KStdDeque<std::packaged_task<void()>> queue_;
-    bool shutdownRequested_ = false;
-    std::thread thread_;
-
-    KStdVector<KStdUniquePtr<GlobalObjectHolder>> globalRoots_;
-    KStdVector<KStdUniquePtr<StackObjectHolder>> stackRoots_;
 };
 
 } // namespace

kotlin-native/runtime/src/main/cpp/SingleThreadExecutor.hpp

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the LICENSE file.
+ */
+
+#pragma once
+
+#include <condition_variable>
+#include <functional>
+#include <future>
+#include <mutex>
+#include <thread>
+
+#include "Types.h"
+#include "Utils.hpp"
+
+namespace kotlin {
+
+// TODO: Try to generalize enough, so that FinalizerProcessor is implementable in terms of it.
+//       Requirements: avoid heap allocations as much as possible.
+// TODO: Try to generalize so enough, that Worker.cpp can be written on top of this.
+//       Requirements: delayed tasks.
+
+// Thread that has a context attached to it. Context is created and destroyed on the thread.
+// Thread cannot join before the destructor, because otherwise `context()` will dangle.
+template <typename Context>
+class ThreadWithContext : private Pinned {
+public:
+    ThreadWithContext() = default;
+
+    template <typename ContextFactory, typename Function, typename... Args>
+    explicit ThreadWithContext(ContextFactory&& contextFactory, Function&& f, Args&&... args) :
+        thread_(
+                [this, contextFactory = std::forward<ContextFactory>(contextFactory), f = std::forward<Function>(f)](
+                        Args&&... args) mutable {
+                    auto context = contextFactory();
+                    {
+                        std::unique_lock guard(startMutex_);
+                        context_ = &context;
+                    }
+                    startCV_.notify_one();
+                    std::invoke(std::forward<Function>(f), std::forward<Args>(args)...);
+                    std::unique_lock guard(stopMutex_);
+                    stopCV_.wait(guard, [this] { return needsShutdown_; });
+                },
+                std::forward<Args>(args)...) {}
+
+    ~ThreadWithContext() {
+        {
+            std::unique_lock guard(stopMutex_);
+            needsShutdown_ = true;
+        }
+        stopCV_.notify_one();
+        thread_.join();
+    }
+
+    // Wait until thread is fully initialized and `context()` is created.
+    void waitInitialized() noexcept {
+        std::unique_lock guard(startMutex_);
+        startCV_.wait(guard, [this] { return context_ != nullptr; });
+    }
+
+    // May only be called after the thread has fully initialized. Use `WaitInitialized()` to be sure.
+    Context& context() const noexcept {
+        RuntimeAssert(context_ != nullptr, "context must be set");
+        return *context_;
+    }
+
+    std::thread::id get_id() const noexcept { return thread_.get_id(); }
+
+private:
+    std::condition_variable startCV_;
+    std::mutex startMutex_;
+    Context* context_ = nullptr;
+
+    // Need to keep thread alive for the entire lifetime of this object, because `context_` lifetime
+    // is bound to the thread.
+    std::condition_variable stopCV_;
+    std::mutex stopMutex_;
+    bool needsShutdown_ = false;
+
+    std::thread thread_;
+};
+
+// TODO: Replace with `std::jthread`.
+// A thread that always joins in the destructor
+class JoiningThread : private MoveOnly {
+public:
+    JoiningThread() = default;
+
+    template <typename Function, typename... Args>
+    explicit JoiningThread(Function&& f, Args&&... args) : thread_(std::forward<Function>(f), std::forward<Args>(args)...) {}
+
+    ~JoiningThread() { thread_.join(); }
+
+    std::thread::id get_id() const noexcept { return thread_.get_id(); }
+
+private:
+    std::thread thread_;
+};
+
+// Execute tasks on a single worker thread.
+// `Thread` must join in the destructor.
+template <typename Thread>
+class SingleThreadExecutor : private Pinned {
+public:
+    // Starts the worker thread immediately.
+    template <typename ThreadFactory>
+    explicit SingleThreadExecutor(ThreadFactory&& threadFactory) noexcept :
+        thread_(std::forward<ThreadFactory>(threadFactory)(&SingleThreadExecutor::RunLoop, this)) {}
+
+    SingleThreadExecutor() noexcept :
+        SingleThreadExecutor([](auto&& function, auto&&... args) {
+            return Thread(std::forward<decltype(function)>(function), std::forward<decltype(args)>(args)...);
+        }) {}
+
+    ~SingleThreadExecutor() {
+        {
+            std::unique_lock guard(workMutex_);
+            // Note: This can only happen in destructor, because otherwise `context_` will be a dangling
+            // pointer to the destroyed thread's stack.
+            shutdownRequested_ = true;
+        }
+        workCV_.notify_one();
+    }
+
+    Thread& thread() noexcept { return thread_; }
+
+    // Schedule task execution on the worker thread. The returned future is resolved when the task has completed.
+    // If `this` is destroyed before the task manages to complete, the returned future will fail with exception upon `.get()`.
+    // If the task moves the runtime into a runnable state, it should move it back into the native state.
+    template <typename Task>
+    [[nodiscard]] std::future<void> Execute(Task&& f) noexcept {
+        std::packaged_task<void()> task(std::forward<Task>(f));
+        auto future = task.get_future();
+        {
+            std::unique_lock guard(workMutex_);
+            queue_.push_back(std::move(task));
+        }
+        workCV_.notify_one();
+        return future;
+    }
+
+private:
+    void RunLoop() noexcept {
+        while (true) {
+            std::packaged_task<void()> task;
+            {
+                std::unique_lock guard(workMutex_);
+                workCV_.wait(guard, [this] { return !queue_.empty() || shutdownRequested_; });
+                if (shutdownRequested_) {
+                    return;
+                }
+                task = std::move(queue_.front());
+                queue_.pop_front();
+            }
+            task();
+        }
+    }
+
+    std::condition_variable workCV_;
+    std::mutex workMutex_;
+    KStdDeque<std::packaged_task<void()>> queue_;
+    bool shutdownRequested_ = false;
+
+    Thread thread_;
+};
+
+template <typename Context, typename ContextFactory>
+SingleThreadExecutor<ThreadWithContext<Context>> MakeSingleThreadExecutorWithContext(ContextFactory&& contextFactory) noexcept {
+    return SingleThreadExecutor<ThreadWithContext<Context>>(
+            [contextFactory = std::forward<ContextFactory>(contextFactory)](auto&& function, auto&&... args) mutable {
+                return ThreadWithContext<Context>(
+                        std::forward<ContextFactory>(contextFactory), std::forward<decltype(function)>(function),
+                        std::forward<decltype(args)>(args)...);
+            });
+}
+
+template <typename Context>
+SingleThreadExecutor<ThreadWithContext<Context>> MakeSingleThreadExecutorWithContext() noexcept {
+    return MakeSingleThreadExecutorWithContext<Context>([] { return Context(); });
+}
+
+} // namespace kotlin

kotlin-native/runtime/src/main/cpp/SingleThreadExecutorTest.cpp

@@ -0,0 +1,195 @@
+/*
+ * Copyright 2010-2021 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
+ * that can be found in the LICENSE file.
+ */
+
+#include "SingleThreadExecutor.hpp"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "KAssert.h"
+#include "TestSupport.hpp"
+
+using namespace kotlin;
+
+using testing::_;
+
+namespace {
+
+class PinnedContext : private Pinned {
+public:
+    struct ScopedMocks : private Pinned {
+        testing::StrictMock<testing::MockFunction<void(PinnedContext&)>> ctorMock;
+        testing::StrictMock<testing::MockFunction<void(PinnedContext&)>> dtorMock;
+
+        ScopedMocks() {
+            RuntimeAssert(PinnedContext::ctorMock == nullptr, "ctor mock must be null was %p", PinnedContext::ctorMock);
+            PinnedContext::ctorMock = &ctorMock;
+            RuntimeAssert(PinnedContext::dtorMock == nullptr, "dtor mock must be null was %p", PinnedContext::dtorMock);
+            PinnedContext::dtorMock = &dtorMock;
+        }
+
+        ~ScopedMocks() {
+            RuntimeAssert(PinnedContext::ctorMock == &ctorMock, "ctor mock must be %p was %p", &ctorMock, PinnedContext::ctorMock);
+            PinnedContext::ctorMock = nullptr;
+            RuntimeAssert(PinnedContext::dtorMock == &dtorMock, "dtor mock must be %p was %p", &dtorMock, PinnedContext::dtorMock);
+            PinnedContext::dtorMock = nullptr;
+        }
+    };
+
+    PinnedContext() { ctorMock->Call(*this); }
+
+    ~PinnedContext() { dtorMock->Call(*this); }
+
+private:
+    static testing::MockFunction<void(PinnedContext&)>* ctorMock;
+    static testing::MockFunction<void(PinnedContext&)>* dtorMock;
+};
+
+testing::MockFunction<void(PinnedContext&)>* PinnedContext::ctorMock = nullptr;
+testing::MockFunction<void(PinnedContext&)>* PinnedContext::dtorMock = nullptr;
+
+} // namespace
+
+TEST(ThreadWithContextTest, ContextThreadBound) {
+    PinnedContext::ScopedMocks mocks;
+    PinnedContext* createdContext = nullptr;
+    std::thread::id createdThread;
+    testing::StrictMock<testing::MockFunction<void()>> function;
+    EXPECT_CALL(mocks.ctorMock, Call(_)).WillOnce([&](PinnedContext& context) {
+        createdContext = &context;
+        createdThread = std::this_thread::get_id();
+    });
+    EXPECT_CALL(function, Call()).WillOnce([&] { EXPECT_THAT(std::this_thread::get_id(), createdThread); });
+    auto thread = ::make_unique<ThreadWithContext<PinnedContext>>([] { return PinnedContext(); }, function.AsStdFunction());
+    thread->waitInitialized();
+    testing::Mock::VerifyAndClearExpectations(&function);
+    testing::Mock::VerifyAndClearExpectations(&mocks.ctorMock);
+    EXPECT_THAT(createdThread, thread->get_id());
+    EXPECT_THAT(thread->context(), testing::Ref(*createdContext));
+
+    EXPECT_CALL(mocks.dtorMock, Call(testing::Ref(*createdContext))).WillOnce([&] {
+        EXPECT_THAT(std::this_thread::get_id(), createdThread);
+    });
+    thread.reset();
+    testing::Mock::VerifyAndClearExpectations(&mocks.dtorMock);
+}
+
+TEST(ThreadWithContextTest, WaitInitialized) {
+    PinnedContext::ScopedMocks mocks;
+    PinnedContext* createdContext = nullptr;
+    std::mutex ctorMutex;
+    EXPECT_CALL(mocks.ctorMock, Call(_)).WillOnce([&](PinnedContext& context) {
+        std::unique_lock guard(ctorMutex);
+        createdContext = &context;
+    });
+
+    testing::StrictMock<testing::MockFunction<void()>> function;
+    EXPECT_CALL(function, Call()).Times(0);
+    ctorMutex.lock();
+    auto thread = ::make_unique<ThreadWithContext<PinnedContext>>([] { return PinnedContext(); }, function.AsStdFunction());
+
+    std::atomic_bool initialized = false;
+    std::thread initializedWaiter([&] {
+        thread->waitInitialized();
+        initialized = true;
+    });
+
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    EXPECT_THAT(initialized.load(), false);
+    testing::Mock::VerifyAndClearExpectations(&function);
+
+    EXPECT_CALL(function, Call());
+    ctorMutex.unlock();
+    initializedWaiter.join();
+    testing::Mock::VerifyAndClearExpectations(&mocks.ctorMock);
+    testing::Mock::VerifyAndClearExpectations(&function);
+    EXPECT_THAT(initialized.load(), true);
+
+    EXPECT_THAT(thread->context(), testing::Ref(*createdContext));
+    EXPECT_CALL(mocks.dtorMock, Call(testing::Ref(*createdContext)));
+}
+
+TEST(SingleThreadExecutorTest, Execute) {
+    SingleThreadExecutor<JoiningThread> executor;
+
+    std::mutex taskMutex;
+    testing::StrictMock<testing::MockFunction<void()>> task;
+
+    EXPECT_CALL(task, Call()).WillOnce([&] { std::unique_lock guard(taskMutex); });
+    taskMutex.lock();
+    auto future = executor.Execute(task.AsStdFunction());
+
+    auto futureStatus = future.wait_for(std::chrono::milliseconds(10));
+    EXPECT_THAT(futureStatus, std::future_status::timeout);
+
+    taskMutex.unlock();
+    future.get();
+    testing::Mock::VerifyAndClearExpectations(&task);
+}
+
+TEST(SingleThreadExecutorTest, DropExecutorWithTasks) {
+    auto executor = make_unique<SingleThreadExecutor<JoiningThread>>();
+
+    std::mutex taskMutex;
+    testing::StrictMock<testing::MockFunction<void()>> task;
+
+    std::atomic_bool taskStarted = false;
+    EXPECT_CALL(task, Call()).WillOnce([&] {
+        taskStarted = true;
+        std::unique_lock guard(taskMutex);
+    });
+    taskMutex.lock();
+    auto future = executor->Execute(task.AsStdFunction());
+    while (!taskStarted) {}
+
+    KStdVector<std::pair<std::future<void>, bool>> newTasks;
+    constexpr size_t tasksCount = 100;
+    for (size_t i = 0; i < tasksCount; ++i) {
+        newTasks.push_back(std::make_pair(executor->Execute([&newTasks, i] { newTasks[i].second = true; }), false));
+    }
+
+    taskMutex.unlock();
+    executor.reset();
+
+    testing::Mock::VerifyAndClearExpectations(&task);
+    future.get();
+
+    // There's no guarantee whether any of those succeed, or any fail.
+    for (auto& [future, success] : newTasks) {
+        if (success) {
+            future.get();
+        } else {
+            EXPECT_THROW(future.get(), std::future_error);
+        }
+    }
+}
+
+TEST(SingleThreadExecutorTest, ExecuteFromManyThreads) {
+    struct Context {
+        KStdVector<int> result;
+    };
+    auto executor = MakeSingleThreadExecutorWithContext<Context>();
+
+    std::atomic_bool canStart = false;
+
+    KStdVector<int> expected;
+    KStdVector<std::thread> threads;
+    for (int i = 0; i < kDefaultThreadCount; ++i) {
+        expected.push_back(i);
+        threads.emplace_back([&, i] {
+            while (!canStart) {
+            }
+            executor.Execute([&] { executor.thread().context().result.push_back(i); }).get();
+        });
+    }
+
+    canStart = true;
+
+    for (auto& thread : threads) {
+        thread.join();
+    }
+
+    EXPECT_THAT(executor.thread().context().result, testing::UnorderedElementsAreArray(expected));
+}