Fixed race condition with thread deletion, update RT64

This commit is contained in:
Mr-Wiseguy 2024-04-27 15:06:26 -04:00
parent 598948bfb6
commit 7c37f8ef8e
8 changed files with 101 additions and 87 deletions

3
.gitignore vendored
View File

@ -53,3 +53,6 @@ imgui.ini
rt64.log rt64.log
node_modules/ node_modules/
# Recompiler Linux binary
N64Recomp

View File

@ -223,6 +223,7 @@ if (WIN32)
endif() endif()
if (LINUX) if (LINUX)
find_package(SDL2 REQUIRED)
find_package(X11 REQUIRED) find_package(X11 REQUIRED)
# Generate icon_bytes.c from the app icon PNG. # Generate icon_bytes.c from the app icon PNG.
@ -232,6 +233,11 @@ if (LINUX)
) )
target_sources(Zelda64Recompiled PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/icon_bytes.c) target_sources(Zelda64Recompiled PRIVATE ${CMAKE_CURRENT_BINARY_DIR}/icon_bytes.c)
message(STATUS "SDL2_FOUND = ${SDL2_FOUND}")
message(STATUS "SDL2_INCLUDE_DIRS = ${SDL2_INCLUDE_DIRS}")
target_include_directories(Zelda64Recompiled PRIVATE ${SDL2_INCLUDE_DIRS})
message(STATUS "X11_FOUND = ${X11_FOUND}") message(STATUS "X11_FOUND = ${X11_FOUND}")
message(STATUS "X11_Xrandr_FOUND = ${X11_Xrandr_FOUND}") message(STATUS "X11_Xrandr_FOUND = ${X11_Xrandr_FOUND}")
message(STATUS "X11_INCLUDE_DIR = ${X11_INCLUDE_DIR}") message(STATUS "X11_INCLUDE_DIR = ${X11_INCLUDE_DIR}")
@ -249,7 +255,7 @@ if (LINUX)
include_directories(${FREETYPE_LIBRARIES}) include_directories(${FREETYPE_LIBRARIES})
target_link_libraries(Zelda64Recompiled PRIVATE ${FREETYPE_LIBRARIES}) target_link_libraries(Zelda64Recompiled PRIVATE ${FREETYPE_LIBRARIES})
target_link_libraries(Zelda64Recompiled PRIVATE "-latomic") target_link_libraries(Zelda64Recompiled PRIVATE "-latomic -static-libstdc++")
endif() endif()
target_link_libraries(Zelda64Recompiled PRIVATE target_link_libraries(Zelda64Recompiled PRIVATE

@ -1 +1 @@
Subproject commit 404f38c8061b40f19d92febe172d035dc9a82a3e Subproject commit 92ae002acf9c344fed5575d6def79fb6a8207181

View File

@ -68,8 +68,7 @@ bool do_send(RDRAM_ARG PTR(OSMesgQueue) mq_, OSMesg msg, bool jam, bool block) {
while (MQ_IS_FULL(mq)) { while (MQ_IS_FULL(mq)) {
debug_printf("[Message Queue] Thread %d is blocked on send\n", TO_PTR(OSThread, ultramodern::this_thread())->id); debug_printf("[Message Queue] Thread %d is blocked on send\n", TO_PTR(OSThread, ultramodern::this_thread())->id);
ultramodern::thread_queue_insert(PASS_RDRAM GET_MEMBER(OSMesgQueue, mq_, blocked_on_send), ultramodern::this_thread()); ultramodern::thread_queue_insert(PASS_RDRAM GET_MEMBER(OSMesgQueue, mq_, blocked_on_send), ultramodern::this_thread());
ultramodern::run_next_thread(PASS_RDRAM1); ultramodern::run_next_thread_and_wait(PASS_RDRAM1);
ultramodern::wait_for_resumed(PASS_RDRAM1);
} }
} }
@ -107,8 +106,7 @@ bool do_recv(RDRAM_ARG PTR(OSMesgQueue) mq_, PTR(OSMesg) msg_, bool block) {
while (MQ_IS_EMPTY(mq)) { while (MQ_IS_EMPTY(mq)) {
debug_printf("[Message Queue] Thread %d is blocked on receive\n", TO_PTR(OSThread, ultramodern::this_thread())->id); debug_printf("[Message Queue] Thread %d is blocked on receive\n", TO_PTR(OSThread, ultramodern::this_thread())->id);
ultramodern::thread_queue_insert(PASS_RDRAM GET_MEMBER(OSMesgQueue, mq_, blocked_on_recv), ultramodern::this_thread()); ultramodern::thread_queue_insert(PASS_RDRAM GET_MEMBER(OSMesgQueue, mq_, blocked_on_recv), ultramodern::this_thread());
ultramodern::run_next_thread(PASS_RDRAM1); ultramodern::run_next_thread_and_wait(PASS_RDRAM1);
ultramodern::wait_for_resumed(PASS_RDRAM1);
} }
} }

View File

@ -1,21 +1,20 @@
#include "ultramodern.hpp" #include "ultramodern.hpp"
void ultramodern::run_next_thread(RDRAM_ARG1) {
if (thread_queue_empty(PASS_RDRAM running_queue)) {
throw std::runtime_error("No threads left to run!\n");
}
OSThread* to_run = TO_PTR(OSThread, thread_queue_pop(PASS_RDRAM running_queue));
debug_printf("[Scheduling] Resuming execution of thread %d\n", to_run->id);
to_run->context->running.signal();
}
void ultramodern::schedule_running_thread(RDRAM_ARG PTR(OSThread) t_) { void ultramodern::schedule_running_thread(RDRAM_ARG PTR(OSThread) t_) {
debug_printf("[Scheduling] Adding thread %d to the running queue\n", TO_PTR(OSThread, t_)->id); debug_printf("[Scheduling] Adding thread %d to the running queue\n", TO_PTR(OSThread, t_)->id);
thread_queue_insert(PASS_RDRAM running_queue, t_); thread_queue_insert(PASS_RDRAM running_queue, t_);
TO_PTR(OSThread, t_)->state = OSThreadState::QUEUED; TO_PTR(OSThread, t_)->state = OSThreadState::QUEUED;
} }
void swap_to_thread(RDRAM_ARG OSThread *to) {
debug_printf("[Scheduling] Thread %d giving execution to thread %d\n", TO_PTR(OSThread, ultramodern::this_thread())->id, to->id);
// Insert this thread in the running queue.
ultramodern::thread_queue_insert(PASS_RDRAM ultramodern::running_queue, ultramodern::this_thread());
TO_PTR(OSThread, ultramodern::this_thread())->state = OSThreadState::QUEUED;
// Unpause the target thread and wait for this one to be unpaused.
ultramodern::resume_thread_and_wait(PASS_RDRAM to);
}
void ultramodern::check_running_queue(RDRAM_ARG1) { void ultramodern::check_running_queue(RDRAM_ARG1) {
// Check if there are any threads in the running queue. // Check if there are any threads in the running queue.
if (!thread_queue_empty(PASS_RDRAM running_queue)) { if (!thread_queue_empty(PASS_RDRAM running_queue)) {
@ -25,38 +24,11 @@ void ultramodern::check_running_queue(RDRAM_ARG1) {
if (next_thread->priority > self->priority) { if (next_thread->priority > self->priority) {
ultramodern::thread_queue_pop(PASS_RDRAM running_queue); ultramodern::thread_queue_pop(PASS_RDRAM running_queue);
// Swap to the higher priority thread. // Swap to the higher priority thread.
ultramodern::swap_to_thread(PASS_RDRAM next_thread); swap_to_thread(PASS_RDRAM next_thread);
} }
} }
} }
void ultramodern::swap_to_thread(RDRAM_ARG OSThread *to) {
debug_printf("[Scheduling] Thread %d giving execution to thread %d\n", TO_PTR(OSThread, ultramodern::this_thread())->id, to->id);
// Insert this thread in the running queue.
thread_queue_insert(PASS_RDRAM running_queue, ultramodern::this_thread());
TO_PTR(OSThread, ultramodern::this_thread())->state = OSThreadState::QUEUED;
// Unpause the target thread and wait for this one to be unpaused.
ultramodern::resume_thread(to);
ultramodern::wait_for_resumed(PASS_RDRAM1);
}
void ultramodern::wait_for_resumed(RDRAM_ARG1) {
TO_PTR(OSThread, ultramodern::this_thread())->context->running.wait();
// If this thread was marked to be destroyed by another thre, destroy it again from its own context.
// This will actually destroy the thread instead of just marking it to be destroyed.
if (TO_PTR(OSThread, ultramodern::this_thread())->destroyed) {
osDestroyThread(PASS_RDRAM NULLPTR);
}
}
void ultramodern::resume_thread(OSThread *t) {
if (t->state != OSThreadState::QUEUED) {
assert(false && "Threads should only be resumed from the queued state!");
}
t->state = OSThreadState::RUNNING;
t->context->running.signal();
}
extern "C" void pause_self(RDRAM_ARG1) { extern "C" void pause_self(RDRAM_ARG1) {
while (true) { while (true) {
// Wait until an external message arrives, then allow the next thread to run. // Wait until an external message arrives, then allow the next thread to run.

View File

@ -86,7 +86,7 @@ void ultramodern::set_native_thread_name(const std::string& name) {
void ultramodern::set_native_thread_priority(ThreadPriority pri) { void ultramodern::set_native_thread_priority(ThreadPriority pri) {
// TODO linux thread priority // TODO linux thread priority
printf("set_native_thread_priority unimplemented\n"); // printf("set_native_thread_priority unimplemented\n");
// int nPriority = THREAD_PRIORITY_NORMAL; // int nPriority = THREAD_PRIORITY_NORMAL;
// // Convert ThreadPriority to Win32 priority // // Convert ThreadPriority to Win32 priority
@ -116,7 +116,43 @@ void ultramodern::set_native_thread_priority(ThreadPriority pri) {
std::atomic_int temporary_threads = 0; std::atomic_int temporary_threads = 0;
std::atomic_int permanent_threads = 0; std::atomic_int permanent_threads = 0;
static void _thread_func(RDRAM_ARG PTR(OSThread) self_, PTR(thread_func_t) entrypoint, PTR(void) arg) { void wait_for_resumed(RDRAM_ARG UltraThreadContext* thread_context) {
TO_PTR(OSThread, ultramodern::this_thread())->context->running.wait();
// If this thread's context was replaced by another thread or deleted, destroy it again from its own context.
// This will trigger thread cleanup instead.
if (TO_PTR(OSThread, ultramodern::this_thread())->context != thread_context) {
osDestroyThread(PASS_RDRAM NULLPTR);
}
}
void resume_thread(OSThread* t) {
debug_printf("[Thread] Resuming execution of thread %d\n", t->id);
t->context->running.signal();
}
void run_next_thread(RDRAM_ARG1) {
if (ultramodern::thread_queue_empty(PASS_RDRAM ultramodern::running_queue)) {
throw std::runtime_error("No threads left to run!\n");
}
OSThread* to_run = TO_PTR(OSThread, ultramodern::thread_queue_pop(PASS_RDRAM ultramodern::running_queue));
debug_printf("[Scheduling] Resuming execution of thread %d\n", to_run->id);
to_run->context->running.signal();
}
void ultramodern::run_next_thread_and_wait(RDRAM_ARG1) {
UltraThreadContext* cur_context = TO_PTR(OSThread, thread_self)->context;
run_next_thread(PASS_RDRAM1);
wait_for_resumed(PASS_RDRAM cur_context);
}
void ultramodern::resume_thread_and_wait(RDRAM_ARG OSThread *t) {
UltraThreadContext* cur_context = TO_PTR(OSThread, thread_self)->context;
resume_thread(t);
wait_for_resumed(PASS_RDRAM cur_context);
}
static void _thread_func(RDRAM_ARG PTR(OSThread) self_, PTR(thread_func_t) entrypoint, PTR(void) arg, UltraThreadContext* thread_context) {
OSThread *self = TO_PTR(OSThread, self_); OSThread *self = TO_PTR(OSThread, self_);
debug_printf("[Thread] Thread created: %d\n", self->id); debug_printf("[Thread] Thread created: %d\n", self->id);
thread_self = self_; thread_self = self_;
@ -135,26 +171,35 @@ static void _thread_func(RDRAM_ARG PTR(OSThread) self_, PTR(thread_func_t) entry
} }
// Signal the initialized semaphore to indicate that this thread can be started. // Signal the initialized semaphore to indicate that this thread can be started.
self->context->initialized.signal(); thread_context->initialized.signal();
debug_printf("[Thread] Thread waiting to be started: %d\n", self->id); debug_printf("[Thread] Thread waiting to be started: %d\n", self->id);
// Wait until the thread is marked as running. // Wait until the thread is marked as running.
ultramodern::wait_for_resumed(PASS_RDRAM1); wait_for_resumed(PASS_RDRAM thread_context);
// Make sure the thread wasn't replaced or destroyed before it was started.
if (self->context == thread_context) {
debug_printf("[Thread] Thread started: %d\n", self->id); debug_printf("[Thread] Thread started: %d\n", self->id);
try { try {
// Run the thread's function with the provided argument. // Run the thread's function with the provided argument.
run_thread_function(PASS_RDRAM entrypoint, self->sp, arg); run_thread_function(PASS_RDRAM entrypoint, self->sp, arg);
// The thread function terminated normally, so mark this thread as destroyed and run the next thread.
self->destroyed = true;
ultramodern::run_next_thread(PASS_RDRAM1);
} catch (ultramodern::thread_terminated& terminated) { } catch (ultramodern::thread_terminated& terminated) {
} }
}
else {
debug_printf("[Thread] Thread destroyed before being started: %d\n", self->id);
}
// Dispose of this thread after it completes and run the next queued thread. // Check if the thread hasn't been destroyed or replaced. If so, then the thread terminated or destroyed itself,
ultramodern::cleanup_thread(self); // so mark this thread as destroyed and run the next queued thread.
if (self->context == thread_context) {
self->context = nullptr;
run_next_thread(PASS_RDRAM1);
}
// Dispose of this thread now that it's completed or terminated.
ultramodern::cleanup_thread(thread_context);
// TODO fix these being hardcoded (this is only used for quicksaving) // TODO fix these being hardcoded (this is only used for quicksaving)
if ((self->id == 2 && self->priority == 5) || self->id == 13) { // slowly, flashrom if ((self->id == 2 && self->priority == 5) || self->id == 13) { // slowly, flashrom
@ -187,7 +232,7 @@ extern "C" void osStartThread(RDRAM_ARG PTR(OSThread) t_) {
// Otherwise, immediately start the thread and terminate this one. // Otherwise, immediately start the thread and terminate this one.
else { else {
t->state = OSThreadState::QUEUED; t->state = OSThreadState::QUEUED;
ultramodern::resume_thread(t); resume_thread(t);
//throw ultramodern::thread_terminated{}; //throw ultramodern::thread_terminated{};
} }
} }
@ -202,11 +247,11 @@ extern "C" void osCreateThread(RDRAM_ARG PTR(OSThread) t_, OSId id, PTR(thread_f
t->id = id; t->id = id;
t->state = OSThreadState::STOPPED; t->state = OSThreadState::STOPPED;
t->sp = sp - 0x10; // Set up the first stack frame t->sp = sp - 0x10; // Set up the first stack frame
t->destroyed = false;
// Spawn a new thread, which will immediately pause itself and wait until it's been started. // Spawn a new thread, which will immediately pause itself and wait until it's been started.
// Pass the context as an argument to the thread function to ensure that it can't get cleared before the thread captures its value.
t->context = new UltraThreadContext{}; t->context = new UltraThreadContext{};
t->context->host_thread = std::thread{_thread_func, PASS_RDRAM t_, entrypoint, arg}; t->context->host_thread = std::thread{_thread_func, PASS_RDRAM t_, entrypoint, arg, t->context};
} }
extern "C" void osStopThread(RDRAM_ARG PTR(OSThread) t_) { extern "C" void osStopThread(RDRAM_ARG PTR(OSThread) t_) {
@ -220,12 +265,6 @@ extern "C" void osDestroyThread(RDRAM_ARG PTR(OSThread) t_) {
OSThread* t = TO_PTR(OSThread, t_); OSThread* t = TO_PTR(OSThread, t_);
// Check if the thread is destroying itself (arg is null or thread_self) // Check if the thread is destroying itself (arg is null or thread_self)
if (t_ == thread_self) { if (t_ == thread_self) {
// Check if the thread was destroyed by another thread. If it wasn't, then this thread destroyed itself and a new thread
// needs to be run.
if (!t->destroyed) {
t->destroyed = true;
ultramodern::run_next_thread(PASS_RDRAM1);
}
throw ultramodern::thread_terminated{}; throw ultramodern::thread_terminated{};
} }
// Otherwise if the thread isn't stopped, remove it from its currrent queue., // Otherwise if the thread isn't stopped, remove it from its currrent queue.,
@ -233,10 +272,11 @@ extern "C" void osDestroyThread(RDRAM_ARG PTR(OSThread) t_) {
ultramodern::thread_queue_remove(PASS_RDRAM t->queue, t_); ultramodern::thread_queue_remove(PASS_RDRAM t->queue, t_);
} }
// Check if the thread has already been destroyed to prevent destroying it again. // Check if the thread has already been destroyed to prevent destroying it again.
if (!t->destroyed) { UltraThreadContext* cur_context = t->context;
if (cur_context != nullptr) {
// Mark the target thread as destroyed and resume it. When it starts it'll check this and terminate itself instead of resuming. // Mark the target thread as destroyed and resume it. When it starts it'll check this and terminate itself instead of resuming.
t->destroyed = true; t->context = nullptr;
t->context->running.signal(); cur_context->running.signal();
} }
} }
@ -277,20 +317,18 @@ PTR(OSThread) ultramodern::this_thread() {
} }
static std::thread thread_cleaner_thread; static std::thread thread_cleaner_thread;
static moodycamel::BlockingConcurrentQueue<OSThread*> deleted_threads{}; static moodycamel::BlockingConcurrentQueue<UltraThreadContext*> deleted_threads{};
extern std::atomic_bool exited; extern std::atomic_bool exited;
void thread_cleaner_func() { void thread_cleaner_func() {
using namespace std::chrono_literals; using namespace std::chrono_literals;
while (!exited) { while (!exited) {
OSThread* to_delete; UltraThreadContext* to_delete;
if (deleted_threads.wait_dequeue_timed(to_delete, 10ms)) { if (deleted_threads.wait_dequeue_timed(to_delete, 10ms)) {
printf("Deleting thread %d\n", to_delete->id); debug_printf("[Cleanup] Deleting thread context %p\n", to_delete);
UltraThreadContext* cur_context = to_delete->context;
to_delete->context = nullptr;
cur_context->host_thread.join(); to_delete->host_thread.join();
delete cur_context; delete to_delete;
} }
} }
} }
@ -299,8 +337,8 @@ void ultramodern::init_thread_cleanup() {
thread_cleaner_thread = std::thread{thread_cleaner_func}; thread_cleaner_thread = std::thread{thread_cleaner_func};
} }
void ultramodern::cleanup_thread(OSThread *t) { void ultramodern::cleanup_thread(UltraThreadContext *cur_context) {
deleted_threads.enqueue(t); deleted_threads.enqueue(cur_context);
} }
void ultramodern::join_thread_cleaner_thread() { void ultramodern::join_thread_cleaner_thread() {

View File

@ -103,7 +103,6 @@ typedef struct OSThread_t {
int32_t pad3; int32_t pad3;
UltraThreadContext* context; // An actual pointer regardless of platform UltraThreadContext* context; // An actual pointer regardless of platform
int32_t sp; int32_t sp;
bool destroyed;
} OSThread; } OSThread;
typedef u32 OSEvent; typedef u32 OSEvent;

View File

@ -77,12 +77,10 @@ void wait_for_external_message(RDRAM_ARG1);
// Thread scheduling. // Thread scheduling.
void check_running_queue(RDRAM_ARG1); void check_running_queue(RDRAM_ARG1);
void wait_for_resumed(RDRAM_ARG1); void run_next_thread_and_wait(RDRAM_ARG1);
void run_next_thread(RDRAM_ARG1); void resume_thread_and_wait(RDRAM_ARG OSThread* t);
void swap_to_thread(RDRAM_ARG OSThread *to);
void resume_thread(OSThread* t);
void schedule_running_thread(RDRAM_ARG PTR(OSThread) t); void schedule_running_thread(RDRAM_ARG PTR(OSThread) t);
void cleanup_thread(OSThread *t); void cleanup_thread(UltraThreadContext* thread_context);
uint32_t permanent_thread_count(); uint32_t permanent_thread_count();
uint32_t temporary_thread_count(); uint32_t temporary_thread_count();
struct thread_terminated : std::exception {}; struct thread_terminated : std::exception {};