Add direct (same process) mode for packet stream

Signed-off-by: Andrew Opalach <andrew@akon.city>

3 files changed, 47 insertions, 16 deletions

diff --git a/src/util/thread/thread.h b/src/util/thread/thread.h
index 921de23..6ea5722 100644
--- a/src/util/thread/thread.h
+++ b/src/util/thread/thread.h
@@ -2,10 +2,10 @@
 
 #include <al/types.h>
 
-#ifndef _WIN32
-#include <pthread.h>
-#else
+#ifdef NAUNET_ON_WINDOWS
 #include "../../winwrap.h"
+#else
+#include <pthread.h>
 #endif
 
 #ifdef NAUNET_HAS_THREAD_CANCEL
@@ -15,27 +15,41 @@ enum {
 };
 #endif
 
-struct nn_thread {
-#ifndef _WIN32
-    pthread_t thread;
+#ifdef NAUNET_ON_WINDOWS
+enum {
+    NNWT_THREAD_SCHED_OTHER = 0,
+    NNWT_THREAD_SCHED_FIFO,
+    NNWT_THREAD_SCHED_RR
+};
 #else
+enum {
+    NNWT_THREAD_SCHED_OTHER = SCHED_OTHER,
+    NNWT_THREAD_SCHED_FIFO = SCHED_FIFO,
+    NNWT_THREAD_SCHED_RR = SCHED_RR
+};
+#endif
+
+struct nn_thread {
+#ifdef NAUNET_ON_WINDOWS
     HANDLE thread;
+#else
+    pthread_t thread;
 #endif
 };
 
 struct nn_mutex {
-#ifndef _WIN32
-    pthread_mutex_t mutex;
-#else
+#ifdef NAUNET_ON_WINDOWS
     CRITICAL_SECTION mutex;
+#else
+    pthread_mutex_t mutex;
 #endif
 };
 
 struct nn_cond {
-#ifndef _WIN32
-    pthread_cond_t cond;
-#else
+#ifdef NAUNET_ON_WINDOWS
     CONDITION_VARIABLE cond;
+#else
+    pthread_cond_t cond;
 #endif
     bool condition;
 };
@@ -44,12 +58,12 @@ typedef f64 nn_os_tstamp;
 
 #define NNWT_TS_FROM_USEC(usec) ((usec) * 1e-6)
 
-#ifndef _WIN32
-#define NNWT_THREADCALL
-typedef void * nn_thread_result;
-#else
+#ifdef NAUNET_ON_WINDOWS
 #define NNWT_THREADCALL __stdcall
 typedef unsigned long nn_thread_result;
+#else
+#define NNWT_THREADCALL
+typedef void * nn_thread_result;
 #endif
 
 typedef nn_thread_result (NNWT_THREADCALL * nn_thread_func)(void *);
@@ -57,6 +71,9 @@ typedef nn_thread_result (NNWT_THREADCALL * nn_thread_func)(void *);
 void nn_thread_sleep(nn_os_tstamp delay);
 
 void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userdata);
+
+void nn_thread_set_priority(s32 policy, s32 priority);
+
 void nn_thread_join(struct nn_thread *thread);
 #ifdef NAUNET_HAS_THREAD_CANCEL
 void nn_thread_cancel(struct nn_thread *thread);
diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c
index 5fc1081..f07fcb4 100644
--- a/src/util/thread/thread_linux.c
+++ b/src/util/thread/thread_linux.c
@@ -17,6 +17,14 @@ void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userd
     pthread_create(&thread->thread, NULL, func, userdata);
 }
 
+void nn_thread_set_priority(s32 policy, s32 priority)
+{
+    pthread_t this = pthread_self();
+    struct sched_param params = { 0 };
+    params.sched_priority = priority;
+    pthread_setschedparam(this, policy, &params);
+}
+
 void nn_thread_join(struct nn_thread *thread)
 {
     pthread_join(thread->thread, NULL);
diff --git a/src/util/thread/thread_windows.c b/src/util/thread/thread_windows.c
index 4104bb3..ca9be80 100644
--- a/src/util/thread/thread_windows.c
+++ b/src/util/thread/thread_windows.c
@@ -12,6 +12,12 @@ void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userd
     thread->thread = CreateThread(NULL, 0, func, userdata, 0, NULL);
 }
 
+void nn_thread_set_priority(s32 policy, s32 priority)
+{
+    (void)policy;
+    (void)priority;
+}
+
 void nn_thread_join(struct nn_thread *thread)
 {
     WaitForSingleObject(thread->thread, INFINITE);