Merge pull request #5208 from bunnei/service-threads

Service threads

8 files changed, 237 insertions, 126 deletions

diff --git a/src/core/hle/kernel/hle_ipc.cpp b/src/core/hle/kernel/hle_ipc.cpp
index e75e80ad0..83decf6cf 100644
--- a/src/core/hle/kernel/hle_ipc.cpp
+++ b/src/core/hle/kernel/hle_ipc.cpp
@@ -46,43 +46,6 @@ void SessionRequestHandler::ClientDisconnected(
     boost::range::remove_erase(connected_sessions, server_session);
 }
 
-std::shared_ptr<WritableEvent> HLERequestContext::SleepClientThread(
-    const std::string& reason, u64 timeout, WakeupCallback&& callback,
-    std::shared_ptr<WritableEvent> writable_event) {
-    // Put the client thread to sleep until the wait event is signaled or the timeout expires.
-
-    if (!writable_event) {
-        // Create event if not provided
-        const auto pair = WritableEvent::CreateEventPair(kernel, "HLE Pause Event: " + reason);
-        writable_event = pair.writable;
-    }
-
-    Handle event_handle = InvalidHandle;
-    {
-        KScopedSchedulerLockAndSleep lock(kernel, event_handle, thread.get(), timeout);
-        thread->SetHLECallback(
-            [context = *this, callback](std::shared_ptr<Thread> thread) mutable -> bool {
-                ThreadWakeupReason reason = thread->GetSignalingResult() == RESULT_TIMEOUT
-                                                ? ThreadWakeupReason::Timeout
-                                                : ThreadWakeupReason::Signal;
-                callback(thread, context, reason);
-                context.WriteToOutgoingCommandBuffer(*thread);
-                return true;
-            });
-        const auto readable_event{writable_event->GetReadableEvent()};
-        writable_event->Clear();
-        thread->SetHLESyncObject(readable_event.get());
-        thread->SetStatus(ThreadStatus::WaitHLEEvent);
-        thread->SetSynchronizationResults(nullptr, RESULT_TIMEOUT);
-        readable_event->AddWaitingThread(thread);
-    }
-    thread->SetHLETimeEvent(event_handle);
-
-    is_thread_waiting = true;
-
-    return writable_event;
-}
-
 HLERequestContext::HLERequestContext(KernelCore& kernel, Core::Memory::Memory& memory,
                                      std::shared_ptr<ServerSession> server_session,
                                      std::shared_ptr<Thread> thread)
diff --git a/src/core/hle/kernel/hle_ipc.h b/src/core/hle/kernel/hle_ipc.h
index c31a65476..b112e1ebd 100644
--- a/src/core/hle/kernel/hle_ipc.h
+++ b/src/core/hle/kernel/hle_ipc.h
@@ -129,23 +129,6 @@ public:
     using WakeupCallback = std::function<void(
         std::shared_ptr<Thread> thread, HLERequestContext& context, ThreadWakeupReason reason)>;
 
-    /**
-     * Puts the specified guest thread to sleep until the returned event is signaled or until the
-     * specified timeout expires.
-     * @param reason Reason for pausing the thread, to be used for debugging purposes.
-     * @param timeout Timeout in nanoseconds after which the thread will be awoken and the callback
-     * invoked with a Timeout reason.
-     * @param callback Callback to be invoked when the thread is resumed. This callback must write
-     * the entire command response once again, regardless of the state of it before this function
-     * was called.
-     * @param writable_event Event to use to wake up the thread. If unspecified, an event will be
-     * created.
-     * @returns Event that when signaled will resume the thread and call the callback function.
-     */
-    std::shared_ptr<WritableEvent> SleepClientThread(
-        const std::string& reason, u64 timeout, WakeupCallback&& callback,
-        std::shared_ptr<WritableEvent> writable_event = nullptr);
-
     /// Populates this context with data from the requesting process/thread.
     ResultCode PopulateFromIncomingCommandBuffer(const HandleTable& handle_table,
                                                  u32_le* src_cmdbuf);
diff --git a/src/core/hle/kernel/kernel.cpp b/src/core/hle/kernel/kernel.cpp
index 04cae3a43..e8ece8164 100644
--- a/src/core/hle/kernel/kernel.cpp
+++ b/src/core/hle/kernel/kernel.cpp
@@ -8,13 +8,14 @@
 #include <functional>
 #include <memory>
 #include <thread>
-#include <unordered_map>
+#include <unordered_set>
 #include <utility>
 
 #include "common/assert.h"
 #include "common/logging/log.h"
 #include "common/microprofile.h"
 #include "common/thread.h"
+#include "common/thread_worker.h"
 #include "core/arm/arm_interface.h"
 #include "core/arm/cpu_interrupt_handler.h"
 #include "core/arm/exclusive_monitor.h"
@@ -35,6 +36,7 @@
 #include "core/hle/kernel/physical_core.h"
 #include "core/hle/kernel/process.h"
 #include "core/hle/kernel/resource_limit.h"
+#include "core/hle/kernel/service_thread.h"
 #include "core/hle/kernel/shared_memory.h"
 #include "core/hle/kernel/synchronization.h"
 #include "core/hle/kernel/thread.h"
@@ -60,6 +62,8 @@ struct KernelCore::Impl {
         RegisterHostThread();
 
         global_scheduler_context = std::make_unique<Kernel::GlobalSchedulerContext>(kernel);
+        service_thread_manager =
+            std::make_unique<Common::ThreadWorker>(1, "yuzu:ServiceThreadManager");
 
         InitializePhysicalCores();
         InitializeSystemResourceLimit(kernel);
@@ -76,6 +80,12 @@ struct KernelCore::Impl {
     }
 
     void Shutdown() {
+        process_list.clear();
+
+        // Ensures all service threads gracefully shutdown
+        service_thread_manager.reset();
+        service_threads.clear();
+
         next_object_id = 0;
         next_kernel_process_id = Process::InitialKIPIDMin;
         next_user_process_id = Process::ProcessIDMin;
@@ -89,8 +99,6 @@ struct KernelCore::Impl {
 
         cores.clear();
 
-        process_list.clear();
-
         current_process = nullptr;
 
         system_resource_limit = nullptr;
@@ -103,10 +111,8 @@ struct KernelCore::Impl {
 
         exclusive_monitor.reset();
 
-        num_host_threads = 0;
-        std::fill(register_host_thread_keys.begin(), register_host_thread_keys.end(),
-                  std::thread::id{});
-        std::fill(register_host_thread_values.begin(), register_host_thread_values.end(), 0);
+        // Next host thead ID to use, 0-3 IDs represent core threads, >3 represent others
+        next_host_thread_id = Core::Hardware::NUM_CPU_CORES;
     }
 
     void InitializePhysicalCores() {
@@ -186,52 +192,46 @@ struct KernelCore::Impl {
         }
     }
 
+    /// Creates a new host thread ID, should only be called by GetHostThreadId
+    u32 AllocateHostThreadId(std::optional<std::size_t> core_id) {
+        if (core_id) {
+            // The first for slots are reserved for CPU core threads
+            ASSERT(*core_id < Core::Hardware::NUM_CPU_CORES);
+            return static_cast<u32>(*core_id);
+        } else {
+            return next_host_thread_id++;
+        }
+    }
+
+    /// Gets the host thread ID for the caller, allocating a new one if this is the first time
+    u32 GetHostThreadId(std::optional<std::size_t> core_id = std::nullopt) {
+        const thread_local auto host_thread_id{AllocateHostThreadId(core_id)};
+        return host_thread_id;
+    }
+
+    /// Registers a CPU core thread by allocating a host thread ID for it
     void RegisterCoreThread(std::size_t core_id) {
-        const std::thread::id this_id = std::this_thread::get_id();
+        ASSERT(core_id < Core::Hardware::NUM_CPU_CORES);
+        const auto this_id = GetHostThreadId(core_id);
         if (!is_multicore) {
             single_core_thread_id = this_id;
         }
-        const auto end =
-            register_host_thread_keys.begin() + static_cast<ptrdiff_t>(num_host_threads);
-        const auto it = std::find(register_host_thread_keys.begin(), end, this_id);
-        ASSERT(core_id < Core::Hardware::NUM_CPU_CORES);
-        ASSERT(it == end);
-        InsertHostThread(static_cast<u32>(core_id));
     }
 
+    /// Registers a new host thread by allocating a host thread ID for it
     void RegisterHostThread() {
-        const std::thread::id this_id = std::this_thread::get_id();
-        const auto end =
-            register_host_thread_keys.begin() + static_cast<ptrdiff_t>(num_host_threads);
-        const auto it = std::find(register_host_thread_keys.begin(), end, this_id);
-        if (it == end) {
-            InsertHostThread(registered_thread_ids++);
-        }
-    }
-
-    void InsertHostThread(u32 value) {
-        const size_t index = num_host_threads++;
-        ASSERT_MSG(index < NUM_REGISTRABLE_HOST_THREADS, "Too many host threads");
-        register_host_thread_values[index] = value;
-        register_host_thread_keys[index] = std::this_thread::get_id();
+        [[maybe_unused]] const auto this_id = GetHostThreadId();
     }
 
-    [[nodiscard]] u32 GetCurrentHostThreadID() const {
-        const std::thread::id this_id = std::this_thread::get_id();
+    [[nodiscard]] u32 GetCurrentHostThreadID() {
+        const auto this_id = GetHostThreadId();
         if (!is_multicore && single_core_thread_id == this_id) {
             return static_cast<u32>(system.GetCpuManager().CurrentCore());
         }
-        const auto end =
-            register_host_thread_keys.begin() + static_cast<ptrdiff_t>(num_host_threads);
-        const auto it = std::find(register_host_thread_keys.begin(), end, this_id);
-        if (it == end) {
-            return Core::INVALID_HOST_THREAD_ID;
-        }
-        return register_host_thread_values[static_cast<size_t>(
-            std::distance(register_host_thread_keys.begin(), it))];
+        return this_id;
     }
 
-    Core::EmuThreadHandle GetCurrentEmuThreadID() const {
+    [[nodiscard]] Core::EmuThreadHandle GetCurrentEmuThreadID() {
         Core::EmuThreadHandle result = Core::EmuThreadHandle::InvalidHandle();
         result.host_handle = GetCurrentHostThreadID();
         if (result.host_handle >= Core::Hardware::NUM_CPU_CORES) {
@@ -325,15 +325,8 @@ struct KernelCore::Impl {
     std::unique_ptr<Core::ExclusiveMonitor> exclusive_monitor;
     std::vector<Kernel::PhysicalCore> cores;
 
-    // 0-3 IDs represent core threads, >3 represent others
-    std::atomic<u32> registered_thread_ids{Core::Hardware::NUM_CPU_CORES};
-
-    // Number of host threads is a relatively high number to avoid overflowing
-    static constexpr size_t NUM_REGISTRABLE_HOST_THREADS = 64;
-    std::atomic<size_t> num_host_threads{0};
-    std::array<std::atomic<std::thread::id>, NUM_REGISTRABLE_HOST_THREADS>
-        register_host_thread_keys{};
-    std::array<std::atomic<u32>, NUM_REGISTRABLE_HOST_THREADS> register_host_thread_values{};
+    // Next host thead ID to use, 0-3 IDs represent core threads, >3 represent others
+    std::atomic<u32> next_host_thread_id{Core::Hardware::NUM_CPU_CORES};
 
     // Kernel memory management
     std::unique_ptr<Memory::MemoryManager> memory_manager;
@@ -345,12 +338,19 @@ struct KernelCore::Impl {
     std::shared_ptr<Kernel::SharedMemory> irs_shared_mem;
     std::shared_ptr<Kernel::SharedMemory> time_shared_mem;
 
+    // Threads used for services
+    std::unordered_set<std::shared_ptr<Kernel::ServiceThread>> service_threads;
+
+    // Service threads are managed by a worker thread, so that a calling service thread can queue up
+    // the release of itself
+    std::unique_ptr<Common::ThreadWorker> service_thread_manager;
+
     std::array<std::shared_ptr<Thread>, Core::Hardware::NUM_CPU_CORES> suspend_threads{};
     std::array<Core::CPUInterruptHandler, Core::Hardware::NUM_CPU_CORES> interrupts{};
     std::array<std::unique_ptr<Kernel::KScheduler>, Core::Hardware::NUM_CPU_CORES> schedulers{};
 
     bool is_multicore{};
-    std::thread::id single_core_thread_id{};
+    u32 single_core_thread_id{};
 
     std::array<u64, Core::Hardware::NUM_CPU_CORES> svc_ticks{};
 
@@ -639,4 +639,19 @@ void KernelCore::ExitSVCProfile() {
     MicroProfileLeave(MICROPROFILE_TOKEN(Kernel_SVC), impl->svc_ticks[core]);
 }
 
+std::weak_ptr<Kernel::ServiceThread> KernelCore::CreateServiceThread(const std::string& name) {
+    auto service_thread = std::make_shared<Kernel::ServiceThread>(*this, 1, name);
+    impl->service_thread_manager->QueueWork(
+        [this, service_thread] { impl->service_threads.emplace(service_thread); });
+    return service_thread;
+}
+
+void KernelCore::ReleaseServiceThread(std::weak_ptr<Kernel::ServiceThread> service_thread) {
+    impl->service_thread_manager->QueueWork([this, service_thread] {
+        if (auto strong_ptr = service_thread.lock()) {
+            impl->service_threads.erase(strong_ptr);
+        }
+    });
+}
+
 } // namespace Kernel
diff --git a/src/core/hle/kernel/kernel.h b/src/core/hle/kernel/kernel.h
index 5846c3f39..e3169f5a7 100644
--- a/src/core/hle/kernel/kernel.h
+++ b/src/core/hle/kernel/kernel.h
@@ -42,6 +42,7 @@ class Process;
 class ResourceLimit;
 class KScheduler;
 class SharedMemory;
+class ServiceThread;
 class Synchronization;
 class Thread;
 class TimeManager;
@@ -227,6 +228,22 @@ public:
 
     void ExitSVCProfile();
 
+    /**
+     * Creates an HLE service thread, which are used to execute service routines asynchronously.
+     * While these are allocated per ServerSession, these need to be owned and managed outside of
+     * ServerSession to avoid a circular dependency.
+     * @param name String name for the ServerSession creating this thread, used for debug purposes.
+     * @returns The a weak pointer newly created service thread.
+     */
+    std::weak_ptr<Kernel::ServiceThread> CreateServiceThread(const std::string& name);
+
+    /**
+     * Releases a HLE service thread, instructing KernelCore to free it. This should be called when
+     * the ServerSession associated with the thread is destroyed.
+     * @param service_thread Service thread to release.
+     */
+    void ReleaseServiceThread(std::weak_ptr<Kernel::ServiceThread> service_thread);
+
 private:
     friend class Object;
     friend class Process;
diff --git a/src/core/hle/kernel/server_session.cpp b/src/core/hle/kernel/server_session.cpp
index a35c8aa4b..b40fe3916 100644
--- a/src/core/hle/kernel/server_session.cpp
+++ b/src/core/hle/kernel/server_session.cpp
@@ -25,19 +25,19 @@
 namespace Kernel {
 
 ServerSession::ServerSession(KernelCore& kernel) : SynchronizationObject{kernel} {}
-ServerSession::~ServerSession() = default;
+
+ServerSession::~ServerSession() {
+    kernel.ReleaseServiceThread(service_thread);
+}
 
 ResultVal<std::shared_ptr<ServerSession>> ServerSession::Create(KernelCore& kernel,
                                                                 std::shared_ptr<Session> parent,
                                                                 std::string name) {
     std::shared_ptr<ServerSession> session{std::make_shared<ServerSession>(kernel)};
 
-    session->request_event =
-        Core::Timing::CreateEvent(name, [session](std::uintptr_t, std::chrono::nanoseconds) {
-            session->CompleteSyncRequest();
-        });
     session->name = std::move(name);
     session->parent = std::move(parent);
+    session->service_thread = kernel.CreateServiceThread(session->name);
 
     return MakeResult(std::move(session));
 }
@@ -142,16 +142,16 @@ ResultCode ServerSession::QueueSyncRequest(std::shared_ptr<Thread> thread,
         std::make_shared<HLERequestContext>(kernel, memory, SharedFrom(this), std::move(thread));
 
     context->PopulateFromIncomingCommandBuffer(kernel.CurrentProcess()->GetHandleTable(), cmd_buf);
-    request_queue.Push(std::move(context));
+
+    if (auto strong_ptr = service_thread.lock()) {
+        strong_ptr->QueueSyncRequest(*this, std::move(context));
+        return RESULT_SUCCESS;
+    }
 
     return RESULT_SUCCESS;
 }
 
-ResultCode ServerSession::CompleteSyncRequest() {
-    ASSERT(!request_queue.Empty());
-
-    auto& context = *request_queue.Front();
-
+ResultCode ServerSession::CompleteSyncRequest(HLERequestContext& context) {
     ResultCode result = RESULT_SUCCESS;
     // If the session has been converted to a domain, handle the domain request
     if (IsDomain() && context.HasDomainMessageHeader()) {
@@ -177,18 +177,13 @@ ResultCode ServerSession::CompleteSyncRequest() {
         }
     }
 
-    request_queue.Pop();
-
     return result;
 }
 
 ResultCode ServerSession::HandleSyncRequest(std::shared_ptr<Thread> thread,
                                             Core::Memory::Memory& memory,
                                             Core::Timing::CoreTiming& core_timing) {
-    const ResultCode result = QueueSyncRequest(std::move(thread), memory);
-    const auto delay = std::chrono::nanoseconds{kernel.IsMulticore() ? 0 : 20000};
-    core_timing.ScheduleEvent(delay, request_event, {});
-    return result;
+    return QueueSyncRequest(std::move(thread), memory);
 }
 
 } // namespace Kernel
diff --git a/src/core/hle/kernel/server_session.h b/src/core/hle/kernel/server_session.h
index d23e9ec68..e8d1d99ea 100644
--- a/src/core/hle/kernel/server_session.h
+++ b/src/core/hle/kernel/server_session.h
@@ -10,6 +10,7 @@
 #include <vector>
 
 #include "common/threadsafe_queue.h"
+#include "core/hle/kernel/service_thread.h"
 #include "core/hle/kernel/synchronization_object.h"
 #include "core/hle/result.h"
 
@@ -43,6 +44,8 @@ class Thread;
  * TLS buffer and control is transferred back to it.
  */
 class ServerSession final : public SynchronizationObject {
+    friend class ServiceThread;
+
 public:
     explicit ServerSession(KernelCore& kernel);
     ~ServerSession() override;
@@ -132,7 +135,7 @@ private:
     ResultCode QueueSyncRequest(std::shared_ptr<Thread> thread, Core::Memory::Memory& memory);
 
     /// Completes a sync request from the emulated application.
-    ResultCode CompleteSyncRequest();
+    ResultCode CompleteSyncRequest(HLERequestContext& context);
 
     /// Handles a SyncRequest to a domain, forwarding the request to the proper object or closing an
     /// object handle.
@@ -163,11 +166,8 @@ private:
     /// The name of this session (optional)
     std::string name;
 
-    /// Core timing event used to schedule the service request at some point in the future
-    std::shared_ptr<Core::Timing::EventType> request_event;
-
-    /// Queue of scheduled service requests
-    Common::MPSCQueue<std::shared_ptr<Kernel::HLERequestContext>> request_queue;
+    /// Thread to dispatch service requests
+    std::weak_ptr<ServiceThread> service_thread;
 };
 
 } // namespace Kernel
diff --git a/src/core/hle/kernel/service_thread.cpp b/src/core/hle/kernel/service_thread.cpp
new file mode 100644
index 000000000..ee46f3e21
--- /dev/null
+++ b/src/core/hle/kernel/service_thread.cpp
@@ -0,0 +1,110 @@
+// Copyright 2020 yuzu emulator team
+// Licensed under GPLv2 or any later version
+// Refer to the license.txt file included.
+
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <thread>
+#include <vector>
+#include <queue>
+
+#include "common/assert.h"
+#include "common/scope_exit.h"
+#include "common/thread.h"
+#include "core/core.h"
+#include "core/hle/kernel/kernel.h"
+#include "core/hle/kernel/server_session.h"
+#include "core/hle/kernel/service_thread.h"
+#include "core/hle/lock.h"
+#include "video_core/renderer_base.h"
+
+namespace Kernel {
+
+class ServiceThread::Impl final {
+public:
+    explicit Impl(KernelCore& kernel, std::size_t num_threads, const std::string& name);
+    ~Impl();
+
+    void QueueSyncRequest(ServerSession& session, std::shared_ptr<HLERequestContext>&& context);
+
+private:
+    std::vector<std::thread> threads;
+    std::queue<std::function<void()>> requests;
+    std::mutex queue_mutex;
+    std::condition_variable condition;
+    const std::string service_name;
+    bool stop{};
+};
+
+ServiceThread::Impl::Impl(KernelCore& kernel, std::size_t num_threads, const std::string& name)
+    : service_name{name} {
+    for (std::size_t i = 0; i < num_threads; ++i)
+        threads.emplace_back([this, &kernel] {
+            Common::SetCurrentThreadName(std::string{"yuzu:HleService:" + service_name}.c_str());
+
+            // Wait for first request before trying to acquire a render context
+            {
+                std::unique_lock lock{queue_mutex};
+                condition.wait(lock, [this] { return stop || !requests.empty(); });
+            }
+
+            kernel.RegisterHostThread();
+
+            while (true) {
+                std::function<void()> task;
+
+                {
+                    std::unique_lock lock{queue_mutex};
+                    condition.wait(lock, [this] { return stop || !requests.empty(); });
+                    if (stop || requests.empty()) {
+                        return;
+                    }
+                    task = std::move(requests.front());
+                    requests.pop();
+                }
+
+                task();
+            }
+        });
+}
+
+void ServiceThread::Impl::QueueSyncRequest(ServerSession& session,
+                                           std::shared_ptr<HLERequestContext>&& context) {
+    {
+        std::unique_lock lock{queue_mutex};
+
+        // ServerSession owns the service thread, so we cannot caption a strong pointer here in the
+        // event that the ServerSession is terminated.
+        std::weak_ptr<ServerSession> weak_ptr{SharedFrom(&session)};
+        requests.emplace([weak_ptr, context{std::move(context)}]() {
+            if (auto strong_ptr = weak_ptr.lock()) {
+                strong_ptr->CompleteSyncRequest(*context);
+            }
+        });
+    }
+    condition.notify_one();
+}
+
+ServiceThread::Impl::~Impl() {
+    {
+        std::unique_lock lock{queue_mutex};
+        stop = true;
+    }
+    condition.notify_all();
+    for (std::thread& thread : threads) {
+        thread.join();
+    }
+}
+
+ServiceThread::ServiceThread(KernelCore& kernel, std::size_t num_threads, const std::string& name)
+    : impl{std::make_unique<Impl>(kernel, num_threads, name)} {}
+
+ServiceThread::~ServiceThread() = default;
+
+void ServiceThread::QueueSyncRequest(ServerSession& session,
+                                     std::shared_ptr<HLERequestContext>&& context) {
+    impl->QueueSyncRequest(session, std::move(context));
+}
+
+} // namespace Kernel
diff --git a/src/core/hle/kernel/service_thread.h b/src/core/hle/kernel/service_thread.h
new file mode 100644
index 000000000..025ab8fb5
--- /dev/null
+++ b/src/core/hle/kernel/service_thread.h
@@ -0,0 +1,28 @@
+// Copyright 2020 yuzu emulator team
+// Licensed under GPLv2 or any later version
+// Refer to the license.txt file included.
+
+#pragma once
+
+#include <memory>
+#include <string>
+
+namespace Kernel {
+
+class HLERequestContext;
+class KernelCore;
+class ServerSession;
+
+class ServiceThread final {
+public:
+    explicit ServiceThread(KernelCore& kernel, std::size_t num_threads, const std::string& name);
+    ~ServiceThread();
+
+    void QueueSyncRequest(ServerSession& session, std::shared_ptr<HLERequestContext>&& context);
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> impl;
+};
+
+} // namespace Kernel