From 0e6de5673b2689b50ebe98a2607d94fcdf9e768d Mon Sep 17 00:00:00 2001 From: Dale Weiler Date: Fri, 11 Mar 2022 08:06:23 -0500 Subject: [PATCH 1/5] fix thread data races --- core/thread/thread_unix.odin | 120 ++++++++++++----------------------- 1 file changed, 40 insertions(+), 80 deletions(-) diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin index b6679bbc2..35b887de1 100644 --- a/core/thread/thread_unix.odin +++ b/core/thread/thread_unix.odin @@ -4,33 +4,22 @@ package thread import "core:runtime" import "core:intrinsics" -import "core:sync" +import sync "core:sync/sync2" import "core:sys/unix" +Thread_State :: enum u8 { + Started, + Joined, + Done, +} + // NOTE(tetra): Aligned here because of core/unix/pthread_linux.odin/pthread_t. // Also see core/sys/darwin/mach_darwin.odin/semaphore_t. Thread_Os_Specific :: struct #align 16 { unix_thread: unix.pthread_t, // NOTE: very large on Darwin, small on Linux. - - // NOTE: pthread has a proc to query this, but it is marked - // as non-portable ("np") so we do this instead. - done: bool, - - // since libpthread doesn't seem to have a way to create a thread - // in a suspended state, we have it wait on this gate, which we - // signal to start it. - // destroyed after thread is started. - start_gate: sync.Condition, - start_mutex: sync.Mutex, - - // if true, the thread has been started and the start_gate has been destroyed. - started: bool, - - // NOTE: with pthreads, it is undefined behavior for multiple threads - // to call join on the same thread at the same time. - // this value is atomically updated to detect this. - // See the comment in `join`. - already_joined: bool, + cond: sync.Cond, + mutex: sync.Mutex, + flags: bit_set[Thread_State; u8], } // // Creates a thread which will run the given procedure. @@ -38,26 +27,31 @@ Thread_Os_Specific :: struct #align 16 { // _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^Thread { __linux_thread_entry_proc :: proc "c" (t: rawptr) -> rawptr { - context = runtime.default_context() t := (^Thread)(t) - sync.condition_wait_for(&t.start_gate) - sync.condition_destroy(&t.start_gate) - sync.mutex_destroy(&t.start_mutex) - t.start_gate = {} - t.start_mutex = {} - context = t.init_context.? or_else runtime.default_context() - + context = runtime.default_context() + + sync.lock(&t.mutex) + t.id = sync.current_thread_id() - t.procedure(t) - if t.init_context == nil { - if context.temp_allocator.data == &runtime.global_default_temp_allocator_data { - runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) - } + if .Started not_in t.flags { + sync.wait(&t.cond, &t.mutex) + } + + init_context := t.init_context + context = init_context.? or_else runtime.default_context() + + t.procedure(t) + + t.flags += { .Done } + + sync.unlock(&t.mutex) + + if init_context == nil && context.temp_allocator.data == &runtime.global_default_temp_allocator_data { + runtime.default_temp_allocator_destroy(auto_cast context.temp_allocator.data) } - intrinsics.atomic_store(&t.done, true) return nil } @@ -76,9 +70,6 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ return nil } thread.creation_allocator = context.allocator - - sync.mutex_init(&thread.start_mutex) - sync.condition_init(&thread.start_gate, &thread.start_mutex) // Set thread priority. policy: i32 @@ -97,65 +88,36 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ res = unix.pthread_attr_setschedparam(&attrs, ¶ms) assert(res == 0) + thread.procedure = procedure if unix.pthread_create(&thread.unix_thread, &attrs, __linux_thread_entry_proc, thread) != 0 { free(thread, thread.creation_allocator) - - sync.condition_destroy(&thread.start_gate) - sync.mutex_destroy(&thread.start_mutex) return nil } - thread.procedure = procedure - return thread } _start :: proc(t: ^Thread) { - if intrinsics.atomic_xchg(&t.started, true) { - return - } - sync.condition_signal(&t.start_gate) + sync.lock(&t.mutex) + t.flags += { .Started } + sync.signal(&t.cond) + sync.unlock(&t.mutex) } _is_done :: proc(t: ^Thread) -> bool { - return intrinsics.atomic_load(&t.done) + return intrinsics.atomic_and(&t.flags, { .Done }) != nil } _join :: proc(t: ^Thread) { - if unix.pthread_equal(unix.pthread_self(), t.unix_thread) { - return - } - // if unix.pthread_self().x == t.unix_thread.x do return; + sync.guard(&t.mutex) - // NOTE(tetra): It's apparently UB for multiple threads to join the same thread - // at the same time. - // If someone else already did, spin until the thread dies. - // See note on `already_joined` field. - // TODO(tetra): I'm not sure if we should do this, or panic, since I'm not - // sure it makes sense to need to join from multiple threads? - if intrinsics.atomic_xchg(&t.already_joined, true) { - for { - if intrinsics.atomic_load(&t.done) { - return - } - intrinsics.cpu_relax() - } - } - - // NOTE(tetra): If we're already dead, don't bother calling to pthread_join as that - // will just return 3 (ESRCH). - // We do this instead because I don't know if there is a danger - // that you may join a different thread from the one you called join on, - // if the thread handle is reused. - if intrinsics.atomic_load(&t.done) { + if .Joined in t.flags || unix.pthread_equal(unix.pthread_self(), t.unix_thread) { return } - ret_val: rawptr - _ = unix.pthread_join(t.unix_thread, &ret_val) - if !intrinsics.atomic_load(&t.done) { - panic("thread not done after join") - } + unix.pthread_join(t.unix_thread, nil) + + t.flags += { .Joined } } _join_multiple :: proc(threads: ..^Thread) { @@ -164,14 +126,12 @@ _join_multiple :: proc(threads: ..^Thread) { } } - _destroy :: proc(t: ^Thread) { _join(t) t.unix_thread = {} free(t, t.creation_allocator) } - _terminate :: proc(t: ^Thread, exit_code: int) { // TODO(bill) } From 7f845bb1655e671c19ae08d6be6b4c4e359a8152 Mon Sep 17 00:00:00 2001 From: Dale Weiler Date: Fri, 11 Mar 2022 08:30:03 -0500 Subject: [PATCH 2/5] fix for spurious wakeups --- core/thread/thread_unix.odin | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin index 35b887de1..8e1ab2b2c 100644 --- a/core/thread/thread_unix.odin +++ b/core/thread/thread_unix.odin @@ -35,7 +35,7 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ t.id = sync.current_thread_id() - if .Started not_in t.flags { + for (.Started not_in t.flags) { sync.wait(&t.cond, &t.mutex) } From 52df80dccd6a340a582b2b2a8ede682274fec0cd Mon Sep 17 00:00:00 2001 From: Dale Weiler Date: Fri, 11 Mar 2022 08:35:23 -0500 Subject: [PATCH 3/5] fix for mac & use atomic store on write side to avoid race --- core/thread/thread_unix.odin | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin index 8e1ab2b2c..e40091cca 100644 --- a/core/thread/thread_unix.odin +++ b/core/thread/thread_unix.odin @@ -44,7 +44,7 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ t.procedure(t) - t.flags += { .Done } + intrinsics.atomic_store(&t.flags, t.flags + { .Done }); sync.unlock(&t.mutex) @@ -105,7 +105,7 @@ _start :: proc(t: ^Thread) { } _is_done :: proc(t: ^Thread) -> bool { - return intrinsics.atomic_and(&t.flags, { .Done }) != nil + return .Done in intrinsics.atomic_load(&t.flags); } _join :: proc(t: ^Thread) { From 32ba5e7ad2e7d814835da2590462a586b33862d0 Mon Sep 17 00:00:00 2001 From: Dale Weiler Date: Fri, 11 Mar 2022 08:36:04 -0500 Subject: [PATCH 4/5] formatting --- core/thread/thread_unix.odin | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin index e40091cca..38bf8241d 100644 --- a/core/thread/thread_unix.odin +++ b/core/thread/thread_unix.odin @@ -44,7 +44,7 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ t.procedure(t) - intrinsics.atomic_store(&t.flags, t.flags + { .Done }); + intrinsics.atomic_store(&t.flags, t.flags + { .Done }) sync.unlock(&t.mutex) @@ -105,7 +105,7 @@ _start :: proc(t: ^Thread) { } _is_done :: proc(t: ^Thread) -> bool { - return .Done in intrinsics.atomic_load(&t.flags); + return .Done in intrinsics.atomic_load(&t.flags) } _join :: proc(t: ^Thread) { From 3da8fa9b27a399d465f616c779d24de00a60e4a1 Mon Sep 17 00:00:00 2001 From: Dale Weiler Date: Fri, 11 Mar 2022 08:41:03 -0500 Subject: [PATCH 5/5] can use sync.guard here --- core/thread/thread_unix.odin | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin index 38bf8241d..fbf89f122 100644 --- a/core/thread/thread_unix.odin +++ b/core/thread/thread_unix.odin @@ -98,10 +98,9 @@ _create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^ } _start :: proc(t: ^Thread) { - sync.lock(&t.mutex) + sync.guard(&t.mutex) t.flags += { .Started } sync.signal(&t.cond) - sync.unlock(&t.mutex) } _is_done :: proc(t: ^Thread) -> bool {