threads, jobs and random numbers

README.md

@@ -32,9 +32,9 @@ brings the power of C++'s `std::vector` to C.
 input/output. It's built on a basic buffer type that you can flush to output
 data. This buffer type also acts as a **string builder**, by letting you append
 data to the buffer, then compile it to a string type.
-- [ ] A multi-threaded job system. You can also just use the cross-platform
+- [x] A multi-threaded job system. You can also just use the operating system's
 primitives.
-- [ ] A Randomness API. Not really sure what else there is to say, really.
+- [x] A Randomness API. Not really sure what else there is to say, really.
 - [ ] A 3D Math API with supports for Vectors and Matricies using SIMD.
 - [ ] A simple GUI initialization system that gets you a window with immediate-mode
 input and a Direct3D11 (Win32) / OpenGL (Linux) context to build a renderer with.

bastd.c

@@ -211,5 +211,7 @@ typedef U8 B8;
 #include "bastd/string.c"
 #include "bastd/buffer.c"
 #include "bastd/args.c"
+#include "bastd/job.c"
+#include "bastd/rand.c"
 
 #endif //BASTD_C

bastd/examples/job_random.c

@@ -0,0 +1,42 @@
+#define BASTD_CLI
+#include "../../bastd.c"
+
+typedef struct DataToPass DataToPass;
+struct DataToPass {	
+	I64 job_num;
+	I64 random_num;
+};
+
+CALLBACK_EXPORT void
+printJob(void *args)
+{
+	DataToPass *dtp = (DataToPass *)args;
+
+	U8 raw[256];
+	Buffer buf = BUFFER(raw, 256);
+	Buffer_appendS8(&buf, S8("Thread #"));
+	Buffer_appendI64(&buf, dtp->job_num);
+	Buffer_appendS8(&buf, S8(" | Random number is "));
+	Buffer_appendI64(&buf, dtp->random_num);
+	Buffer_appendU8(&buf, '\n');
+	Buffer_standardOutput(&buf);
+}
+
+CALLBACK_EXPORT os_ErrorCode
+os_entry(void)
+{	
+	m_Buddy buddy = m_Buddy_create(os_alloc(MEGA(2)), MEGA(2));
+	m_Allocator perm = m_BUDDY_ALLOCATOR(buddy);
+	JobQueue *queue = JobQueue_start(8, &perm);
+
+	for (I64 i = 0; i < 255; i++) {
+		DataToPass *dtp = m_MAKE(DataToPass, 1, &perm);
+		dtp->job_num = i;
+		dtp->random_num = rand_next() % 100;
+		JobQueue_dispatch(queue, JOB(printJob, dtp));
+	}
+
+	JobQueue_completeAll(queue);
+
+	return os_ErrorCode_success;
+}

bastd/job.c

@@ -0,0 +1,100 @@
+#ifndef BASTD_JOB_C
+#define BASTD_JOB_C
+
+typedef void (*JobProc)(void *args);
+
+typedef struct Job Job;
+struct Job {
+	JobProc proc;
+	void *args;
+};
+#define JOB(p, a) (Job){.proc = p, .args = (void *)a}
+
+#define MAX_NUM_JOBS 256
+typedef struct JobQueue JobQueue;
+struct JobQueue {
+	U64 next_read;
+	U64 next_write;
+	
+	U64 complete_len;
+	U64 complete_goal;
+
+	Job jobs[MAX_NUM_JOBS];
+	os_Semaphore semaphore;
+};
+
+FUNCTION B8
+JobQueue__nextJob(JobQueue *queue)
+{
+	B8 done = FALSE;
+	U64 next_read = queue->next_read;
+	U64 new_next_read = (next_read + 1) % MAX_NUM_JOBS;	
+
+	if (next_read != queue->next_write) {
+		U64 index = os_atomic_compareExchange64((I64 volatile *)&queue->next_read, new_next_read, next_read);
+		if (index == next_read) {
+			done = TRUE;
+			Job *job = queue->jobs + index;
+			job->proc(job->args);
+			os_atomic_increment64((I64 volatile *)&queue->complete_len);
+		}
+	}
+
+	return done;
+}
+
+CALLBACK_EXPORT U32
+JobQueue__threadProc(void *args)
+{
+	JobQueue *queue = (JobQueue *)args;
+	for (;;) {
+		if (!JobQueue__nextJob(queue)) {
+			os_Semaphore_wait(queue->semaphore);
+		}
+	}
+}
+
+FUNCTION JobQueue *
+JobQueue_start(int num_threads, m_Allocator *perm)
+{
+	JobQueue *res = m_MAKE(JobQueue, 1, perm);
+
+	res->semaphore = os_Semaphore_create(0, num_threads);
+
+	for (int i = 0; i < num_threads; i++) {
+		os_Thread t = os_Thread_start(JobQueue__threadProc, res);
+		os_Thread_detach(&t);
+	}
+
+	return res;
+}
+
+// Should be called on main thread for now. Maybe need to sync up later
+FUNCTION void
+JobQueue_dispatch(JobQueue *queue, Job job)
+{
+	U64 next_write = queue->next_write;
+	U64 new_next_write = (next_write + 1) % MAX_NUM_JOBS;
+	//ASSERT(new_next_write != queue->next_read, "This will override the current entry");
+	Job *job_to_write = queue->jobs + next_write;
+	*job_to_write = job;
+	queue->complete_goal++;
+
+	os_WRITE_BARRIER;
+
+	queue->next_write = new_next_write;
+	os_Semaphore_increment(queue->semaphore, 1);
+}
+
+FUNCTION void
+JobQueue_completeAll(JobQueue *queue)
+{
+	while (queue->complete_len != queue->complete_goal) {
+		JobQueue__nextJob(queue);
+	}
+
+	queue->complete_len = 0;
+	queue->complete_goal = 0;
+}
+
+#endif//BASTD_JOB_C

bastd/os.c

@@ -9,6 +9,26 @@ FUNCTION U64 os_getFileSize(int fd);
 FUNCTION int os_openFile(U8 *filename, B8 always_create);
 FUNCTION B8 os_closeFile(int fd);
 
+typedef struct os_Thread os_Thread;
+struct os_Thread; // Depends on the OS
+
+typedef U32 (*os_ThreadProc)(void *);
+FUNCTION os_Thread os_Thread_start(os_ThreadProc proc, void *ctx);
+FUNCTION void os_Thread_detach(os_Thread *t);
+FUNCTION void os_Thread_join(os_Thread *t);
+
+FUNCTION I64 os_atomic_compareExchange64(I64 volatile *dst, I64 exchange, I64 compare);
+FUNCTION I64 os_atomic_increment64(I64 volatile *dst);
+FUNCTION I64 os_atomic_decrement64(I64 volatile *dst);
+#define os_WRITE_BARRIER // Depends on the OS
+#define os_READ_BARRIER // Depends on the OS
+
+typedef struct os_Semaphore os_Semaphore;
+struct os_Semaphore; // Depends on the OS
+FUNCTION os_Semaphore os_Semaphore_create(I32 initial_count, I32 max_count);
+FUNCTION B8 os_Semaphore_increment(os_Semaphore semaphore);
+FUNCTION void os_Semaphore_wait(os_Semaphore semaphore);
+
 typedef int os_ErrorCode;
 enum {
 	os_ErrorCode_success = 0,
@@ -16,6 +36,8 @@ enum {
 };
 CALLBACK_EXPORT os_ErrorCode os_entry(void);
 
+FUNCTION U64 os_rdtsc(void);
+
 #if defined(OS_WINDOWS)
 
 #include "os_windows.c"

bastd/os_windows.c

@@ -2,10 +2,13 @@
 #define BASTD_OS_WINDOWS_C
 
 #include <windows.h>
+#include <intrin.h>
 #include <shellapi.h>
 #include <io.h>
 
 #define os_DEBUGBREAK() __debugbreak();
+#define os_WRITE_BARRIER _WriteBarrier()
+#define os_READ_BARRIER _ReadBarrier()
 
 FUNCTION void
 os_abort(char *msg)
@@ -22,7 +25,6 @@ os_alloc(U64 cap)
 	return VirtualAlloc(NIL, cap, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
 }
 
-
 FUNCTION B8
 os_write(int fd, U8 *buf, int len)
 {
@@ -106,6 +108,84 @@ os_getArgcAndArgv(int *argc)
     return argv;
 }
 
+struct os_Thread {
+    HANDLE raw;
+};
+
+FUNCTION os_Thread
+os_Thread_start(os_ThreadProc proc, void *ctx)
+{
+    os_Thread res = {0};
+    res.raw = CreateThread(NIL, 0, proc, ctx, 0, NIL);
+    return res;
+}
+
+FUNCTION void
+os_Thread_detach(os_Thread *t)
+{
+    if (t->raw == NIL) return;
+    CloseHandle(t->raw);
+    t->raw = NIL;
+}
+
+FUNCTION void
+os_Thread_join(os_Thread *t)
+{
+    if (t->raw == NIL) return;
+    WaitForSingleObject(t->raw, INFINITE);
+    os_Thread_detach(t);
+}
+
+FUNCTION I64
+os_atomic_compareExchange64(I64 volatile *dst, I64 exchange, I64 compare)
+{
+    return InterlockedCompareExchange64(dst, exchange, compare);
+}
+
+FUNCTION I64
+os_atomic_increment64(I64 volatile *dst)
+{
+    return InterlockedIncrement64(dst);
+}
+
+FUNCTION I64
+os_atomic_decrement64(I64 volatile *dst)
+{
+    return InterlockedDecrement64(dst);
+}
+
+struct os_Semaphore {
+    HANDLE raw;
+};
+
+FUNCTION os_Semaphore
+os_Semaphore_create(I32 initial_count, I32 max_count)
+{
+    os_Semaphore res = {0};
+
+    res.raw = CreateSemaphoreExA(NIL, initial_count, max_count, NIL, 0, SEMAPHORE_ALL_ACCESS);
+    
+    return res;
+}
+
+FUNCTION B8
+os_Semaphore_increment(os_Semaphore semaphore, I32 count)
+{
+    return ReleaseSemaphore(semaphore.raw, count, NIL);
+}
+
+FUNCTION void
+os_Semaphore_wait(os_Semaphore semaphore)
+{
+    WaitForSingleObjectEx(semaphore.raw, INFINITE, FALSE);
+}
+
+FUNCTION U64
+os_rdtsc(void)
+{
+    return __rdtsc();
+}
+
 #if defined(BASTD_CLI)
 
 CALLBACK_EXPORT int

bastd/rand.c

@@ -0,0 +1,102 @@
+#ifndef BASTD_RAND_C
+#define BASTD_RAND_C
+
+// standard xoshiro256**, idk what any of ts means but it works :P
+
+FUNCTION inline U64
+rand_rotl(U64 x, int k)
+{
+	return (x << k) | (x >> (64 - k));
+}
+
+GLOBAL_VAR U64 rand_seed[4];
+
+FUNCTION U64
+rand_generateSeed(U64 x)
+{
+	// SplitMix64
+	U64 z = (x += 0x9e3779b97f4a7c15);
+	z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9;
+	z = (z ^ (z >> 27)) * 0x94d049bb133111eb;
+	return z ^ (z >> 31);
+}
+
+FUNCTION U64
+rand_next(void)
+{
+	U64 rdtscp_val = os_rdtsc();
+	for (int i = 0; i < COUNT_OF(rand_seed); i++) {
+		if (rand_seed[i] == 0) {
+			rand_seed[i] = rand_generateSeed(rdtscp_val);
+		}
+	}
+
+	U64 res = rand_rotl(rand_seed[1] * 5, 7) * 9;
+
+	U64 t = rand_seed[1] << 17;
+
+	rand_seed[2] ^= rand_seed[0];
+	rand_seed[3] ^= rand_seed[1];
+	rand_seed[1] ^= rand_seed[2];
+	rand_seed[0] ^= rand_seed[3];
+
+	rand_seed[2] ^= t;
+
+	rand_seed[3] = rand_rotl(rand_seed[3], 45);
+
+	return res;
+}
+
+FUNCTION void
+rand_jump(void) {
+	LOCAL_PERSIST const U64 JUMP[] = { 0x180ec6d33cfd0aba, 0xd5a61266f0c9392c, 0xa9582618e03fc9aa, 0x39abdc4529b1661c };
+
+	U64 s0 = 0;
+	U64 s1 = 0;
+	U64 s2 = 0;
+	U64 s3 = 0;
+	for(int i = 0; i < COUNT_OF(JUMP); i++) {
+		for(int b = 0; b < 64; b++) {
+			if (JUMP[i] & UINT64_C(1) << b) {
+				s0 ^= rand_seed[0];
+				s1 ^= rand_seed[1];
+				s2 ^= rand_seed[2];
+				s3 ^= rand_seed[3];
+			}
+			rand_next();	
+		}
+	}
+		
+	rand_seed[0] = s0;
+	rand_seed[1] = s1;
+	rand_seed[2] = s2;
+	rand_seed[3] = s3;
+}
+
+FUNCTION void
+rand_longJump(void) {
+	LOCAL_PERSIST const U64 LONG_JUMP[] = { 0x76e15d3efefdcbbf, 0xc5004e441c522fb3, 0x77710069854ee241, 0x39109bb02acbe635 };
+
+	U64 s0 = 0;
+	U64 s1 = 0;
+	U64 s2 = 0;
+	U64 s3 = 0;
+	for(int i = 0; i < COUNT_OF(LONG_JUMP); i++) {
+		for(int b = 0; b < 64; b++) {
+			if (LONG_JUMP[i] & UINT64_C(1) << b) {
+				s0 ^= rand_seed[0];
+				s1 ^= rand_seed[1];
+				s2 ^= rand_seed[2];
+				s3 ^= rand_seed[3];
+			}
+			rand_next();	
+		}
+	}
+		
+	rand_seed[0] = s0;
+	rand_seed[1] = s1;
+	rand_seed[2] = s2;
+	rand_seed[3] = s3;
+}
+
+#endif

build.bat

@@ -10,6 +10,7 @@ REM cl ..\bastd\examples\memory.c %debugflags%
 REM cl ..\bastd\examples\fib_slice.c %debugflags%
 REM cl ..\bastd\examples\string.c %debugflags%
 REM cl ..\bastd\examples\buffer.c %debugflags%
+REM cl ..\bastd\examples\job_random.c %debugflags%
 
 REM This builds your application with debug symbols.
 cl ..\main.c %debugflags%