Add direct (same process) mode for packet stream

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

10 files changed, 104 insertions, 49 deletions

diff --git a/src/util/error.h b/src/util/error.h
index 898119c..95ae282 100644
--- a/src/util/error.h
+++ b/src/util/error.h
@@ -10,14 +10,14 @@ extern __thread char errorbuf[NNWT_ERROR_STR_MAXLEN];
 
 static inline char *nn_strerror(s32 errnum)
 {
-#ifndef _WIN32
+#ifdef NAUNET_ON_WINDOWS
+    strerror_s(errorbuf, NNWT_ERROR_STR_MAXLEN, errnum);
+#else
 #if (_POSIX_C_SOURCE >= 200112L) && ! _GNU_SOURCE
     if (strerror_r(errnum, errorbuf, NNWT_ERROR_STR_MAXLEN) != 0) errorbuf[0] = '\0';
 #else
     return strerror_r(errnum, errorbuf, NNWT_ERROR_STR_MAXLEN);
 #endif
-#else
-    strerror_s(errorbuf, NNWT_ERROR_STR_MAXLEN, errnum);
 #endif
     return errorbuf;
 }
diff --git a/src/util/file/file.h b/src/util/file/file.h
index 30b44b6..f79bfe5 100644
--- a/src/util/file/file.h
+++ b/src/util/file/file.h
@@ -6,7 +6,9 @@
 #ifdef NAUNET_NEEDS_STDIO_ASSIST
 #include <stdio.h>
 #else
-#ifndef _WIN32
+#ifdef NAUNET_ON_WINDOWS
+// TODO
+#else
 #include <dirent.h>
 #include <fcntl.h>
 #include <unistd.h>
@@ -31,21 +33,24 @@ struct nn_file {
 #ifdef NAUNET_NEEDS_STDIO_ASSIST
     FILE *file;
 #else
-#ifndef _WIN32
-    s32 fd;
+#ifdef NAUNET_ON_WINDOWS
+    // TODO
 #else
+    s32 fd;
 #endif
     bool locked;
 #endif
-    size_t filesize;
+    str path;
+    size_t size;
 };
 
 struct nn_dir {
 #ifdef NAUNET_NEEDS_STDIO_ASSIST
 #else
-#ifndef _WIN32
-    DIR *dir;
+#ifdef NAUNET_ON_WINDOWS
+    // TODO
 #else
+    DIR *dir;
 #endif
 #endif
     str path;
@@ -55,9 +60,9 @@ struct nn_dir_entry {
     u8 type;
 #ifdef NAUNET_NEEDS_STDIO_ASSIST
 #else
-#ifndef _WIN32
-    struct dirent *entry;
+#ifdef NAUNET_ON_WINDOWS
 #else
+    struct dirent *entry;
 #endif
 #endif
     str name;
diff --git a/src/util/file/file_linux.c b/src/util/file/file_linux.c
index a362637..bbc4c67 100644
--- a/src/util/file/file_linux.c
+++ b/src/util/file/file_linux.c
@@ -49,14 +49,16 @@ bool nn_file_open(struct nn_file *file, str *path, s32 flags)
         close(file->fd);
         return false;
     }
-    file->filesize = sb.st_size;
+    file->size = sb.st_size;
 
     file->locked = flags & NNWT_FILE_LOCK;
-    if (file->locked && !lock_file_internal(file->fd, file->filesize)) {
+    if (file->locked && !lock_file_internal(file->fd, file->size)) {
         close(file->fd);
         return false;
     }
 
+    al_str_clone(&file->path, path);
+
     return true;
 }
 
@@ -94,13 +96,13 @@ bool nn_file_truncate(struct nn_file *file, off_t size)
         al_log_error("file", "ftruncate(%jd) failed (%s).", (intmax_t)size, nn_strerror(errno));
         return false;
     }
-    file->filesize = size;
+    file->size = size;
     return true;
 }
 
 size_t nn_file_get_filesize(struct nn_file *file)
 {
-    return file->filesize;
+    return file->size;
 }
 
 #define file_io_loop(call, fail_case) \
@@ -128,9 +130,9 @@ bool nn_file_read(struct nn_file *file, void *buf, size_t size)
 
 void *nn_file_mmap(struct nn_file *file)
 {
-    void *map = mmap(NULL, file->filesize, PROT_READ | PROT_WRITE, MAP_SHARED, file->fd, 0L);
+    void *map = mmap(NULL, file->size, PROT_READ | PROT_WRITE, MAP_SHARED, file->fd, 0L);
     if (map == MAP_FAILED) {
-        al_log_error("file", "mmap() failed (%s).", nn_strerror(errno));
+        al_log_error("file", "mmap(%.*s) failed (%s).", AL_STR_PRINTF(&file->path), nn_strerror(errno));
         return NULL;
     }
     return map;
@@ -138,10 +140,10 @@ void *nn_file_mmap(struct nn_file *file)
 
 void *nn_file_mremap(struct nn_file *file, size_t size, void *old_map)
 {
-    void *map = mremap(old_map, file->filesize, size, MREMAP_MAYMOVE);
+    void *map = mremap(old_map, file->size, size, MREMAP_MAYMOVE);
     if (map == MAP_FAILED) {
         al_log_error("file", "mremap(%p, %zu, %zu) failed (%s).",
-            old_map, file->filesize, size, nn_strerror(errno));
+            old_map, file->size, size, nn_strerror(errno));
         return NULL;
     }
     return map;
@@ -149,18 +151,19 @@ void *nn_file_mremap(struct nn_file *file, size_t size, void *old_map)
 
 void nn_file_munmap(struct nn_file *file, void *map)
 {
-    s32 ret = munmap(map, file->filesize);
+    s32 ret = munmap(map, file->size);
     if (ret != 0) {
-        al_log_error("file", "munmap(%p, %zu) failed (%s).", map, file->filesize, nn_strerror(errno));
+        al_log_error("file", "munmap(%p, %zu) failed (%s).", map, file->size, nn_strerror(errno));
     }
 }
 
 void nn_file_close(struct nn_file *file)
 {
     if (file->locked) {
-        unlock_file_internal(file->fd, file->filesize);
+        unlock_file_internal(file->fd, file->size);
     }
     close(file->fd);
+    al_str_free(&file->path);
 }
 
 bool nn_dir_open(struct nn_dir *dir, str *path)
@@ -178,8 +181,8 @@ bool nn_dir_open(struct nn_dir *dir, str *path)
 
 void nn_dir_close(struct nn_dir *dir)
 {
-    al_str_free(&dir->path);
     closedir(dir->dir);
+    al_str_free(&dir->path);
 }
 
 bool nn_dir_exists(str *path)
diff --git a/src/util/file/file_stdio.c b/src/util/file/file_stdio.c
index 8f6d7ae..2a18c81 100644
--- a/src/util/file/file_stdio.c
+++ b/src/util/file/file_stdio.c
@@ -8,16 +8,16 @@ bool nn_file_open(struct nn_file *file, str *path, s32 flags)
 {
     const char *mode = flags & NNWT_FILE_CREATE ? "wb+" : "rb+";
     char *c_str = al_str_to_c_str(path);
-#ifndef _WIN32
-    file->file = fopen(c_str, mode);
-#else
+#ifdef NAUNET_ON_WINDOWS
     errno_t ret = fopen_s(&file->file, c_str, mode);
+#else
+    file->file = fopen(c_str, mode);
 #endif
     al_free(c_str);
-#ifndef _WIN32
-    if (!file->file) {
-#else
+#ifdef NAUNET_ON_WINDOWS
     if (ret != 0) {
+#else
+    if (!file->file) {
 #endif
         al_log_error("file_stdio", "fopen(%.*s) failed (%s).", AL_STR_PRINTF(path), nn_strerror(errno));
         return false;
@@ -28,7 +28,7 @@ bool nn_file_open(struct nn_file *file, str *path, s32 flags)
         nn_file_close(file);
         return false;
     }
-    file->filesize = pos;
+    file->size = pos;
     rewind(file->file);
 
     return true;
@@ -59,7 +59,7 @@ bool nn_file_truncate(struct nn_file *file, off_t size)
 
 size_t nn_file_get_filesize(struct nn_file *file)
 {
-    return file->filesize;
+    return file->size;
 }
 
 bool nn_file_write(struct nn_file *file, void *buf, size_t size)
diff --git a/src/util/file/util.c b/src/util/file/util.c
index eab9893..93d0825 100644
--- a/src/util/file/util.c
+++ b/src/util/file/util.c
@@ -2,7 +2,7 @@
 
 s32 nn_file_read_as_str(struct nn_file *file, str *out)
 {
-    size_t size = file->filesize;
+    size_t size = file->size;
     al_str_sized(out, size);
     if (!nn_file_read(file, (void *)out->data, size)) {
         al_str_free(out);
@@ -14,7 +14,7 @@ s32 nn_file_read_as_str(struct nn_file *file, str *out)
 
 s32 nn_file_read_as_c_str(struct nn_file *file, char **out)
 {
-    size_t size = file->filesize;
+    size_t size = file->size;
     *out = al_malloc(size + 1);
     if (!nn_file_read(file, (void *)*out, size)) {
         al_free(*out);
diff --git a/src/util/packet.c b/src/util/packet.c
index 900016c..5fe63de 100644
--- a/src/util/packet.c
+++ b/src/util/packet.c
@@ -15,6 +15,7 @@ struct nn_packet *nn_packet_clone(struct nn_packet *packet)
     al_memcpy(nn_buffer_get_ptr(&c->buffer, 0), nn_buffer_get_ptr(&packet->buffer, 0), packet->buffer.size);
     c->windex = packet->windex;
     c->rindex = packet->rindex;
+    c->opaque = packet->opaque;
     return c;
 }
 
@@ -23,6 +24,7 @@ void nn_packet_reset(struct nn_packet *packet)
     nn_buffer_ensure_space(&packet->buffer, NNWT_PACKET_HEADER_LENGTH);
     packet->rindex = NNWT_PACKET_HEADER_LENGTH;
     packet->windex = NNWT_PACKET_HEADER_LENGTH;
+    packet->opaque = NULL;
 }
 
 void nn_packet_write_size(struct nn_packet *packet)
@@ -74,8 +76,17 @@ void nn_packet_write_buffer(struct nn_packet *packet, struct nn_buffer *buf)
         return r; \
     }
 
+#define DEFINE_PACKET_READ_FUNC_EXT(type) \
+    type nn_packet_peek_##type(struct nn_packet *packet) \
+    { \
+        type r; \
+        NNWT_PACKET_PEEK_TYPE(packet, type, r); \
+        return r; \
+    }
+
 DEFINE_PACKET_READ_FUNC(bool)
 DEFINE_PACKET_READ_FUNC(u8)
+DEFINE_PACKET_READ_FUNC_EXT(u8)
 DEFINE_PACKET_READ_FUNC(s8)
 DEFINE_PACKET_READ_FUNC(u16)
 DEFINE_PACKET_READ_FUNC(s16)
diff --git a/src/util/packet.h b/src/util/packet.h
index 41f0a77..4ae9776 100644
--- a/src/util/packet.h
+++ b/src/util/packet.h
@@ -13,6 +13,7 @@ struct nn_packet {
     struct nn_buffer buffer;
     u32 rindex;
     u32 windex;
+    void *opaque;
 };
 
 #define NNWT_PACKET_GET_ID(p) \
@@ -30,6 +31,9 @@ struct nn_packet {
     r = *((type *)nn_buffer_get_ptr(&(p)->buffer, (p)->rindex)); \
     (p)->rindex += (u32)sizeof(type);
 
+#define NNWT_PACKET_PEEK_TYPE(p, type, r) \
+    r = *((type *)nn_buffer_get_ptr(&(p)->buffer, (p)->rindex));
+
 #define NNWT_PACKET_READ_DATA(p, len, r) \
     r = (void *)nn_buffer_get_ptr(&(p)->buffer, (p)->rindex);  \
     (p)->rindex += (u32)len;
@@ -58,6 +62,7 @@ void nn_packet_write_buffer(struct nn_packet *packet, struct nn_buffer *buf);
 bool nn_packet_read_bool(struct nn_packet *packet);
 s8 nn_packet_read_s8(struct nn_packet *packet);
 u8 nn_packet_read_u8(struct nn_packet *packet);
+u8 nn_packet_peek_u8(struct nn_packet *packet);
 s16 nn_packet_read_s16(struct nn_packet *packet);
 u16 nn_packet_read_u16(struct nn_packet *packet);
 s32 nn_packet_read_s32(struct nn_packet *packet);
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);