Merge pull request #458 from Tetralux/linux-threads

Implement core:thread and core:sync on Unix using pthreads

3 files changed, 194 insertions, 17 deletions

diff --git a/core/thread/thread.odin b/core/thread/thread.odin
new file mode 100644
index 000000000..c326b30f1
--- /dev/null
+++ b/core/thread/thread.odin
@@ -0,0 +1,15 @@
+package thread;
+
+import "core:runtime";
+
+Thread_Proc :: #type proc(^Thread);
+
+Thread :: struct {
+	using specific:   Thread_Os_Specific,
+	procedure:        Thread_Proc,
+	data:             rawptr,
+	user_index:       int,
+
+	init_context:     runtime.Context,
+	use_init_context: bool,
+}
\ No newline at end of file
diff --git a/core/thread/thread_unix.odin b/core/thread/thread_unix.odin
new file mode 100644
index 000000000..a05c8c2af
--- /dev/null
+++ b/core/thread/thread_unix.odin
@@ -0,0 +1,153 @@
+// +build linux, darwin
+package thread;
+
+import "core:sys/unix"
+import "core:sync"
+
+// 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,
+
+	// 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,
+}
+
+Thread_Priority :: enum {
+	Normal,
+	Low,
+	High,
+}
+
+//
+// Creates a thread which will run the given procedure.
+// It then waits for `start` to be called.
+//
+// You may provide a slice of bytes to use as the stack for the new thread,
+// but if you do, you are expected to set up the guard pages yourself.
+//
+// The stack must also be aligned appropriately for the platform.
+// We require it's at least 16 bytes aligned to help robustness; other
+// platforms may require page-size alignment.
+// Note also that pthreads requires the stack is at least 6 OS pages in size:
+// 4 are required by pthreads, and two extra for guards pages that will be applied.
+//
+create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^Thread {
+	__linux_thread_entry_proc :: proc "c" (t: rawptr) -> rawptr {
+		t := (^Thread)(t);
+		sync.condition_wait_for(&t.start_gate);
+		sync.condition_destroy(&t.start_gate);
+		t.start_gate = {};
+
+		c := context;
+		if t.use_init_context {
+			c = t.init_context;
+		}
+		context = c;
+
+		t.procedure(t);
+		sync.atomic_store(&t.done, true, .Sequentially_Consistent);
+		return nil;
+	}
+
+	attrs: unix.pthread_attr_t;
+	if unix.pthread_attr_init(&attrs) != 0 do return nil; // NOTE(tetra, 2019-11-01): POSIX OOM.
+	defer unix.pthread_attr_destroy(&attrs);
+
+	// NOTE(tetra, 2019-11-01): These only fail if their argument is invalid.
+	assert(unix.pthread_attr_setdetachstate(&attrs, unix.PTHREAD_CREATE_JOINABLE) == 0);
+	assert(unix.pthread_attr_setinheritsched(&attrs, unix.PTHREAD_EXPLICIT_SCHED) == 0);
+
+	thread := new(Thread);
+	if thread == nil do return nil;
+
+	// Set thread priority.
+	policy: i32;
+	res := unix.pthread_attr_getschedpolicy(&attrs, &policy);
+	assert(res == 0);
+	params: unix.sched_param;
+	res = unix.pthread_attr_getschedparam(&attrs, &params);
+	fmt.println(params.sched_priority);
+	assert(res == 0);
+	low := unix.sched_get_priority_min(policy);
+	high := unix.sched_get_priority_max(policy);
+	switch priority {
+	case .Low:
+		params.sched_priority = low + 1;
+	case .High:
+		params.sched_priority = high;
+	}
+	fmt.println(low, high, params.sched_priority);
+	res = unix.pthread_attr_setschedparam(&attrs, &params);
+	assert(res == 0);
+
+	sync.condition_init(&thread.start_gate);
+	if unix.pthread_create(&thread.unix_thread, &attrs, __linux_thread_entry_proc, thread) != 0 {
+		free(thread);
+		return nil;
+	}
+	thread.procedure = procedure;
+
+	return thread;
+}
+
+start :: proc(t: ^Thread) {
+	if sync.atomic_swap(&t.started, true, .Sequentially_Consistent) do return;
+	sync.condition_signal(&t.start_gate);
+}
+
+is_done :: proc(t: ^Thread) -> bool {
+	return sync.atomic_load(&t.done, .Sequentially_Consistent);
+}
+
+join :: proc(t: ^Thread) {
+	if unix.pthread_equal(unix.pthread_self(), t.unix_thread) do return;
+	// if unix.pthread_self().x == t.unix_thread.x do return;
+
+	// 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 sync.atomic_swap(&t.already_joined, true, .Sequentially_Consistent) {
+		for {
+			if sync.atomic_load(&t.done, .Sequentially_Consistent) do return;
+			sync.yield_processor();
+		}
+	}
+
+	// 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 sync.atomic_load(&t.done, .Sequentially_Consistent) do return;
+
+	ret := unix.pthread_join(t.unix_thread, nil);
+	assert(ret == 0, "cannot join thread");
+	assert(sync.atomic_load(&t.done, .Sequentially_Consistent), "thread not done after join");
+}
+
+import "core:fmt"
+destroy :: proc(t: ^Thread) {
+	join(t);
+	t.unix_thread = {};
+	free(t);
+}
diff --git a/core/thread/thread_windows.odin b/core/thread/thread_windows.odin
index 743f0fec8..5b956940f 100644
--- a/core/thread/thread_windows.odin
+++ b/core/thread/thread_windows.odin
@@ -1,27 +1,29 @@
 package thread
 
-import "core:runtime"
+import "core:sync"
 import "core:sys/win32"
 
-Thread_Proc :: #type proc(^Thread) -> int;
-
 Thread_Os_Specific :: struct {
 	win32_thread:    win32.Handle,
 	win32_thread_id: u32,
+	done: bool, // see note in `is_done`
 }
 
-Thread :: struct {
-	using specific:   Thread_Os_Specific,
-	procedure:        Thread_Proc,
-	data:             rawptr,
-	user_index:       int,
-
-	init_context:     runtime.Context,
-	use_init_context: bool,
+THREAD_PRIORITY_IDLE   :: -15;
+THREAD_PRIORITY_LOWEST :: -2;
+THREAD_PRIORITY_BELOW_NORMAL :: -1;
+THREAD_PRIORITY_NORMAL :: 0;
+THREAD_PRIORITY_ABOVE_NORMAL :: 1;
+THREAD_PRIORITY_HIGHEST :: 2;
+THREAD_PRIORITY_TIME_CRITICAL :: 15;
+
+Thread_Priority :: enum i32 {
+	Normal = THREAD_PRIORITY_NORMAL,
+	Low = THREAD_PRIORITY_LOWEST,
+	High = THREAD_PRIORITY_HIGHEST,
 }
 
-
-create :: proc(procedure: Thread_Proc) -> ^Thread {
+create :: proc(procedure: Thread_Proc, priority := Thread_Priority.Normal) -> ^Thread {
 	win32_thread_id: u32;
 
 	__windows_thread_entry_proc :: proc "c" (t: ^Thread) -> i32 {
@@ -31,7 +33,9 @@ create :: proc(procedure: Thread_Proc) -> ^Thread {
 		}
 		context = c;
 
-		return i32(t.procedure(t));
+		t.procedure(t);
+		sync.atomic_store(&t.done, true, .Sequentially_Consistent);
+		return 0;
 	}
 
 
@@ -47,6 +51,9 @@ create :: proc(procedure: Thread_Proc) -> ^Thread {
 	thread.win32_thread    = win32_thread;
 	thread.win32_thread_id = win32_thread_id;
 
+	ok := win32.set_thread_priority(win32_thread, i32(priority));
+	assert(ok == true);
+
 	return thread;
 }
 
@@ -55,8 +62,10 @@ start :: proc(using thread: ^Thread) {
 }
 
 is_done :: proc(using thread: ^Thread) -> bool {
-	res := win32.wait_for_single_object(win32_thread, 0);
-	return res != win32.WAIT_TIMEOUT;
+	// NOTE(tetra, 2019-10-31): Apparently using wait_for_single_object and
+	// checking if it didn't time out immediately, is not good enough,
+	// so we do it this way instead.
+	return sync.atomic_load(&done, .Sequentially_Consistent);
 }
 
 join :: proc(using thread: ^Thread) {
@@ -72,4 +81,4 @@ destroy :: proc(thread: ^Thread) {
 
 terminate :: proc(using thread : ^Thread, exit_code : u32) {
 	win32.terminate_thread(win32_thread, exit_code);
-}
+}
\ No newline at end of file