summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/common.c51
-rw-r--r--src/curl/http.c79
-rw-r--r--src/curl/http.h8
-rw-r--r--src/ev_embed_compat.c2
-rw-r--r--src/fs_event/fs_event.h8
-rw-r--r--src/multiplex.c49
-rw-r--r--src/multiplex.h17
-rw-r--r--src/packet_cache.c14
-rw-r--r--src/packet_cache.h3
-rw-r--r--src/packet_pool.c68
-rw-r--r--src/packet_pool.h10
-rw-r--r--src/packet_stream.c140
-rw-r--r--src/packet_stream.h14
-rw-r--r--src/rpc.c48
-rw-r--r--src/rpc2.h13
-rw-r--r--src/socket/socket.h26
-rw-r--r--src/util/error.h6
-rw-r--r--src/util/file/file.h21
-rw-r--r--src/util/file/file_linux.c27
-rw-r--r--src/util/file/file_stdio.c16
-rw-r--r--src/util/file/util.c4
-rw-r--r--src/util/packet.c11
-rw-r--r--src/util/packet.h5
-rw-r--r--src/util/thread/thread.h49
-rw-r--r--src/util/thread/thread_linux.c8
-rw-r--r--src/util/thread/thread_windows.c6
26 files changed, 466 insertions, 237 deletions
diff --git a/src/common.c b/src/common.c
index a24592a..4f9baa5 100644
--- a/src/common.c
+++ b/src/common.c
@@ -1,15 +1,48 @@
#include <al/lib.h>
+#include <al/log.h>
#include <al/random.h>
#include <locale.h>
-#ifndef _WIN32
-#include <unistd.h>
-#else
+#ifdef NAUNET_ON_WINDOWS
#include "winwrap.h"
+#else
+#include <unistd.h>
#endif
#include "common.h"
+#define RESET "\033[0m"
+static inline const char *get_color_for_level(u8 level)
+{
+#define NORMAL "\x1B[0m"
+#define RED "\x1B[31m"
+#define GREEN "\x1B[32m"
+#define YELLOW "\x1B[33m"
+#define BLUE "\x1B[34m"
+#define MAGENTA "\x1B[35m"
+#define CYAN "\x1B[36m"
+#define WHITE "\x1B[37m"
+ switch (level) {
+ case AL_LOG_ERROR: return RED;
+ case AL_LOG_WARN: return YELLOW;
+ default: return "";
+ }
+}
+
+s32 al_print_default(void *userdata, u8 level, char *message)
+{
+ (void)userdata;
+ s32 ret = 0;
+#ifdef NAUNET_ON_WINDOWS
+ (void)level;
+ ret = al_printf("%*.*s\n", 0, AL_LOG_MESSAGE_SIZE, message);
+#else
+ ret = al_printf("%s%*.*s%s\n", get_color_for_level(level), 0, AL_LOG_MESSAGE_SIZE, message, RESET);
+#endif
+ al_free(message);
+ return ret;
+}
+
#ifdef AL_MEMORY_TRACKING
#include "util/thread/thread.h"
@@ -30,12 +63,14 @@ bool nn_common_init(void)
{
if (!setlocale(LC_ALL, "")) return false;
-#ifndef _WIN32
- al_set_page_size(sysconf(_SC_PAGESIZE));
-#else
+ al_set_print(al_print_default, NULL);
+
+#ifdef NAUNET_ON_WINDOWS
SYSTEM_INFO info;
GetSystemInfo(&info);
al_set_page_size(info.dwPageSize);
+#else
+ al_set_page_size(sysconf(_SC_PAGESIZE));
#endif
#ifdef AL_MEMORY_TRACKING
@@ -48,7 +83,7 @@ bool nn_common_init(void)
free, malloc_lock, malloc_unlock);
#endif
al_rand_init();
-#ifdef _WIN32
+#ifdef NAUNET_ON_WINDOWS
nn_windows_init();
#endif
@@ -57,7 +92,7 @@ bool nn_common_init(void)
void nn_common_close(void)
{
-#ifdef _WIN32
+#ifdef NAUNET_ON_WINDOWS
nn_windows_close();
#endif
#ifdef AL_MEMORY_TRACKING
diff --git a/src/curl/http.c b/src/curl/http.c
index 60d26f1..d040f53 100644
--- a/src/curl/http.c
+++ b/src/curl/http.c
@@ -6,17 +6,17 @@ static void check_status_codes(struct nn_curl *curl)
{
struct nn_http *http = (struct nn_http *)curl;
- if (http->status_code <= 0) {
+ if (http->status_code <= 0L) {
curl_easy_getinfo(curl->handle, CURLINFO_RESPONSE_CODE, &http->status_code);
- if (http->status_code > 0) {
- http->callback(http->userdata, NNWT_HTTP_RESPONSE_CODE, NULL, http->status_code);
+ if (http->status_code > 0L) {
+ http->callback(http->userdata, NNWT_HTTP_RESPONSE_CODE, NULL, &http->status_code);
}
}
- if (http->content_length < 0 && http->status_code == 200) {
+ if (http->content_length < 0L && http->status_code == 200L) {
curl_easy_getinfo(curl->handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &http->content_length);
- if (http->content_length >= 0) {
- http->callback(http->userdata, NNWT_HTTP_CONTENT_LENGTH, NULL, (s64)http->content_length);
+ if (http->content_length >= 0L) {
+ http->callback(http->userdata, NNWT_HTTP_CONTENT_LENGTH, NULL, &http->content_length);
}
}
}
@@ -35,7 +35,8 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
CURLcode code = msg->data.result;
if (code != CURLE_OK) {
al_log_error("http", "Curl error: %s.", curl_easy_strerror(code));
- http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, -1);
+ http->status_code = -1L;
+ http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
return;
}
@@ -45,26 +46,26 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
check_status_codes(curl);
- if (http->status_code == 200) {
- http->callback(http->userdata, NNWT_HTTP_FINISHED, NULL, http->status_code);
- } else if (http->status_code / 100 == 3) {
- http->callback(http->userdata, NNWT_HTTP_REDIRECT, NULL, http->status_code);
+ if (http->status_code == 200L) {
+ http->callback(http->userdata, NNWT_HTTP_FINISHED, NULL, &http->status_code);
+ } else if (http->status_code / 100L == 3L) {
+ http->callback(http->userdata, NNWT_HTTP_REDIRECT, NULL, &http->status_code);
char *redirect_url = NULL;
curl_easy_getinfo(curl->handle, CURLINFO_REDIRECT_URL, &redirect_url);
if (!redirect_url) {
- http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code);
+ http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
return;
}
nn_curl_set_url(curl, al_str_cr(redirect_url));
- http->status_code = -1;
- http->content_length = -1;
+ http->status_code = -1L;
+ http->content_length = -1L;
if (!nn_curl_add_handle(curl)) {
return;
}
} else {
- http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code);
+ http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, &http->status_code);
}
return;
@@ -81,8 +82,8 @@ static void http_handle_events(void *userdata, struct nn_curl *curl)
bool nn_http_init(struct nn_http *http)
{
- http->status_code = -1;
- http->content_length = -1;
+ http->status_code = -1L;
+ http->content_length = -1L;
http->headers = NULL;
return nn_curl_init(&http->curl);
}
@@ -124,40 +125,49 @@ static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, voi
{
struct nn_http *http = (struct nn_http *)userdata;
check_status_codes(&http->curl);
- return http->callback(http->userdata, NNWT_HTTP_WRITE, (u8 *)buffer, size * nmemb);
+ size_t n = size * nmemb;
+ http->callback(http->userdata, NNWT_HTTP_WRITE, (u8 *)buffer, &n);
+ return n;
}
static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata)
{
struct nn_http *http = (struct nn_http *)userdata;
- return http->callback(http->userdata, NNWT_HTTP_READ, (u8 *)buffer, size * nmemb);
+ size_t n = size * nmemb;
+ http->callback(http->userdata, NNWT_HTTP_READ, (u8 *)buffer, &n);
+ return n;
}
-static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0)
+static void request_callback(void *userdata, u8 op, u8 *buf, void *opaque)
{
struct nn_http_request *request = (struct nn_http_request *)userdata;
switch (op) {
case NNWT_HTTP_READ: {
- al_assert(request->pointer >= 0);
-
+ size_t n = *(size_t *)opaque;
size_t size = nn_buffer_get_size(&request->payload);
- if (request->pointer + int0 > (off_t)size) {
- int0 = size - request->pointer;
+ if (request->pointer + n > size) {
+ n = size - request->pointer;
}
- nn_buffer_read(&request->payload, buf, request->pointer, int0);
- request->pointer += int0;
+ nn_buffer_read(&request->payload, buf, request->pointer, n);
+ request->pointer += n;
+
+ *(size_t *)opaque = n;
break;
}
- case NNWT_HTTP_WRITE:
- nn_buffer_append(&request->response, buf, int0);
+ case NNWT_HTTP_WRITE: {
+ size_t n = *(size_t *)opaque;
+ nn_buffer_append(&request->response, buf, n);
break;
+ }
case NNWT_HTTP_RESPONSE_CODE:
break;
- case NNWT_HTTP_CONTENT_LENGTH:
- nn_buffer_ensure_space(&request->response, int0);
+ case NNWT_HTTP_CONTENT_LENGTH: {
+ curl_off_t length = *(curl_off_t *)opaque;
+ nn_buffer_ensure_space(&request->response, length);
break;
+ }
case NNWT_HTTP_REDIRECT:
break;
case NNWT_HTTP_FINISHED:
@@ -167,7 +177,6 @@ static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0)
request->callback(request->userdata, request, false);
break;
}
- return int0;
}
static bool init_request_internal(struct nn_http *http, u8 method)
@@ -201,7 +210,7 @@ static bool init_request_internal(struct nn_http *http, u8 method)
}
bool nn_http_request_stream(struct nn_http *http, u8 method, struct nn_event_loop *loop,
- size_t (*callback)(void *, u8, u8 *, s64), void *userdata)
+ void (*callback)(void *, u8, u8 *, void *), void *userdata)
{
struct nn_curl *curl = &http->curl;
al_assert(curl->loop == NULL);
@@ -230,9 +239,9 @@ bool nn_http_request(struct nn_http_request *request, u8 method, struct nn_event
request->http.callback = request_callback;
request->http.userdata = request;
- size_t payload = nn_buffer_get_size(&request->payload);
- if (payload > 0) {
- curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, (s64)payload);
+ long payload = (long)nn_buffer_get_size(&request->payload);
+ if (payload > 0L) {
+ curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, payload);
}
request->callback = callback;
diff --git a/src/curl/http.h b/src/curl/http.h
index d5efcf9..952a74a 100644
--- a/src/curl/http.h
+++ b/src/curl/http.h
@@ -26,16 +26,16 @@ enum {
struct nn_http {
struct nn_curl curl;
- s64 status_code;
+ long status_code;
curl_off_t content_length;
struct curl_slist *headers;
- size_t (*callback)(void *, u8, u8 *, s64);
+ void (*callback)(void *, u8, u8 *, void *);
void *userdata;
};
struct nn_http_request {
struct nn_http http;
- off_t pointer;
+ size_t pointer;
struct nn_buffer payload;
struct nn_buffer response;
void (*callback)(void *, struct nn_http_request *, bool);
@@ -49,7 +49,7 @@ void nn_http_add_header(struct nn_http *http, str *header);
void nn_http_set_range(struct nn_http *http, size_t start, size_t end);
void nn_http_close(struct nn_http *http);
bool nn_http_request_stream(struct nn_http *http, u8 method, struct nn_event_loop *loop,
- size_t (*callback)(void *, u8, u8 *, s64), void *userdata);
+ void (*callback)(void *, u8, u8 *, void *), void *userdata);
bool nn_http_request_init(struct nn_http_request *request);
bool nn_http_request(struct nn_http_request *request, u8 method, struct nn_event_loop *loop,
void (*callback)(void *, struct nn_http_request *, bool), void *userdata);
diff --git a/src/ev_embed_compat.c b/src/ev_embed_compat.c
index 2bf262a..d090e5e 100644
--- a/src/ev_embed_compat.c
+++ b/src/ev_embed_compat.c
@@ -2,7 +2,7 @@
#define EV_USE_NANOSLEEP 1
#define EV_USE_REALTIME 1
#define EV_USE_MONOTONIC 1
-#ifdef _WIN32
+#ifdef NAUNET_ON_WINDOWS
#define FD_SETSIZE 2048
#define EV_SELECT_IS_WINSOCKET 1
#else
diff --git a/src/fs_event/fs_event.h b/src/fs_event/fs_event.h
index 1e99b62..2cfef07 100644
--- a/src/fs_event/fs_event.h
+++ b/src/fs_event/fs_event.h
@@ -2,12 +2,12 @@
#include <al/str.h>
-#ifndef _WIN32
-#include <sys/inotify.h>
-#define NNWT_CLOSE_WRITE IN_CLOSE_WRITE
-#else
+#ifdef NAUNET_ON_WINDOWS
struct inotify_event {};
#define NNWT_CLOSE_WRITE 0
+#else
+#include <sys/inotify.h>
+#define NNWT_CLOSE_WRITE IN_CLOSE_WRITE
#endif
#include "../loop.h"
diff --git a/src/multiplex.c b/src/multiplex.c
index 45be88b..24729a2 100644
--- a/src/multiplex.c
+++ b/src/multiplex.c
@@ -1,7 +1,7 @@
#include "multiplex.h"
bool nn_multiplex_socket_init(struct nn_multiplex_socket *multi, u8 type,
- bool (*connection_callback)(void *, u8, struct nn_socket *), void *userdata)
+ bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata)
{
multi->sock.type = type;
if (!nn_socket_init(&multi->sock, NNWT_SOCKET_NONBLOCKING)) {
@@ -28,8 +28,9 @@ static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 reven
u8 id;
ssize_t ret = nn_socket_read(&sock, &id, sizeof(u8));
if (ret > 0) {
- nn_socket_set_blocking(&sock, false);
- if (multi->connection_callback(multi->userdata, id, &sock)) {
+ struct nn_packet_stream *stream = al_alloc_object(struct nn_packet_stream);
+ if (multi->connection_callback(multi->userdata, id, stream)) {
+ nn_packet_stream_from_socket(stream, multi->loop, &sock);
return;
}
}
@@ -60,3 +61,45 @@ void nn_multiplex_socket_close(struct nn_multiplex_socket *multi)
nn_socket_close(&multi->sock);
ev_io_stop(multi->loop->ev, &multi->event);
}
+
+static struct nn_multiplex_direct multiplex_direct_global = { 0 };
+
+void nn_multiplex_direct_init(bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata)
+{
+ multiplex_direct_global.closing_bridge = al_alloc_object(struct nn_packet_stream);
+ multiplex_direct_global.connection_callback = connection_callback;
+ multiplex_direct_global.userdata = userdata;
+}
+
+static void direct_connect(struct nn_packet_stream *client, u8 id)
+{
+ struct nn_packet_stream *server = al_alloc_object(struct nn_packet_stream);
+ nn_packet_stream_init(server, NULL, NULL, NULL);
+ multiplex_direct_global.connection_callback(multiplex_direct_global.userdata, id, server);
+ server->direct = client;
+ client->direct = server;
+ // Connected happens on the server first.
+ nn_packet_stream_connected(server);
+ nn_packet_stream_connected(client);
+}
+
+void nn_multiplex_direct_connect(struct nn_packet_stream *client, u8 id)
+{
+ client->id = id;
+ direct_connect(client, id);
+}
+
+void nn_multiplex_direct_reconnect(struct nn_packet_stream *client)
+{
+ direct_connect(client, client->id);
+}
+
+struct nn_packet_stream *nn_multiplex_direct_get_bridge(void)
+{
+ return multiplex_direct_global.closing_bridge;
+}
+
+void nn_multiplex_direct_close(void)
+{
+ al_free(multiplex_direct_global.closing_bridge);
+}
diff --git a/src/multiplex.h b/src/multiplex.h
index 6339798..264581a 100644
--- a/src/multiplex.h
+++ b/src/multiplex.h
@@ -5,16 +5,29 @@
#include "socket/socket.h"
#include "loop.h"
+#include "packet_stream.h"
+
+struct nn_multiplex_direct {
+ struct nn_packet_stream *closing_bridge;
+ bool (*connection_callback)(void *, u8 id, struct nn_packet_stream *stream);
+ void *userdata;
+};
struct nn_multiplex_socket {
struct nn_socket sock;
struct nn_event_loop *loop;
ev_io event;
- bool (*connection_callback)(void *, u8 id, struct nn_socket *);
+ bool (*connection_callback)(void *, u8 id, struct nn_packet_stream *stream);
void *userdata;
};
bool nn_multiplex_socket_init(struct nn_multiplex_socket *multi, u8 type,
- bool (*connection_callback)(void *, u8, struct nn_socket *), void *userdata);
+ bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata);
bool nn_multiplex_socket_listen(struct nn_multiplex_socket *multi, struct nn_event_loop *loop, str *addr, u16 port);
void nn_multiplex_socket_close(struct nn_multiplex_socket *multi);
+
+void nn_multiplex_direct_init(bool (*connection_callback)(void *, u8, struct nn_packet_stream *), void *userdata);
+void nn_multiplex_direct_connect(struct nn_packet_stream *client, u8 id);
+void nn_multiplex_direct_reconnect(struct nn_packet_stream *client);
+struct nn_packet_stream *nn_multiplex_direct_get_bridge(void);
+void nn_multiplex_direct_close(void);
diff --git a/src/packet_cache.c b/src/packet_cache.c
index 49f2952..3604eb0 100644
--- a/src/packet_cache.c
+++ b/src/packet_cache.c
@@ -51,6 +51,7 @@ bool nn_packet_cache_wait(struct nn_packet_cache *cache, u32 *count)
// If there was an active wait before calling disable(), expected
// behavior would be that wait() returns false.
if (cache->disabled) {
+ *count = 0;
nn_mutex_unlock(&cache->mutex);
return false;
}
@@ -60,13 +61,14 @@ bool nn_packet_cache_wait(struct nn_packet_cache *cache, u32 *count)
return true;
}
-struct nn_packet *nn_packet_cache_pop(struct nn_packet_cache *cache)
+struct nn_packet *nn_packet_cache_at(struct nn_packet_cache *cache, u32 index)
{
- struct nn_packet *packet = NULL;
- if (cache->cache.size > 0) {
- al_array_pop_at(cache->cache, 0, packet);
- }
- return packet;
+ return al_array_at(cache->cache, index);
+}
+
+void nn_packet_cache_lock(struct nn_packet_cache *cache)
+{
+ nn_mutex_lock(&cache->mutex);
}
void nn_packet_cache_unlock(struct nn_packet_cache *cache)
diff --git a/src/packet_cache.h b/src/packet_cache.h
index 1d1b83a..c31347d 100644
--- a/src/packet_cache.h
+++ b/src/packet_cache.h
@@ -17,7 +17,8 @@ void nn_packet_cache_init(struct nn_packet_cache *cache, u32 size);
bool nn_packet_cache_send_packet(struct nn_packet_cache *cache, struct nn_packet *packet);
void nn_packet_cache_flush(struct nn_packet_cache *cache);
bool nn_packet_cache_wait(struct nn_packet_cache *cache, u32 *count);
-struct nn_packet *nn_packet_cache_pop(struct nn_packet_cache *cache);
+struct nn_packet *nn_packet_cache_at(struct nn_packet_cache *cache, u32 index);
+void nn_packet_cache_lock(struct nn_packet_cache *cache);
void nn_packet_cache_unlock(struct nn_packet_cache *cache);
void nn_packet_cache_disable(struct nn_packet_cache *cache);
void nn_packet_cache_enable(struct nn_packet_cache *cache);
diff --git a/src/packet_pool.c b/src/packet_pool.c
index 3645f13..3f454cd 100644
--- a/src/packet_pool.c
+++ b/src/packet_pool.c
@@ -4,9 +4,6 @@ static void return_internal(struct nn_packet_pool *pool, struct nn_packet *packe
{
nn_packet_reset(packet);
al_array_push(pool->empty, packet);
- if (nn_cond_is_waiting(&pool->cond)) {
- nn_cond_signal(&pool->cond);
- }
}
static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents)
@@ -22,28 +19,23 @@ static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents)
return;
}
- while (pool->ready.size > 0) {
- struct nn_packet *packet;
- al_array_pop_at(pool->ready, 0, packet);
- u8 ret = pool->callback(pool->userdata, packet);
- switch (ret) {
- case NNWT_PACKET_POOL_KEEP:
- break;
- case NNWT_PACKET_POOL_RETURN:
- return_internal(pool, packet);
- break;
- }
- }
+ al_array_copy(pool->sending, pool->ready);
+ pool->ready.size = 0;
nn_mutex_unlock(&pool->mutex);
+
+ struct nn_packet *packet;
+ al_array_foreach(pool->sending, i, packet) {
+ pool->callback(pool->userdata, packet);
+ }
+ pool->sending.size = 0;
}
-void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size, struct nn_event_loop *loop,
- u8 (*callback)(void *, struct nn_packet *), void *userdata)
+void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size,
+ struct nn_event_loop *loop, void (*callback)(void *, struct nn_packet *), void *userdata)
{
pool->loop = loop;
- al_assert(size > 0);
pool->size = size;
al_array_init(pool->ready);
@@ -55,6 +47,8 @@ void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size, struct nn_event_
al_array_push(pool->empty, nn_packet_create());
}
+ al_array_init(pool->sending);
+
pool->disabled = false;
nn_cond_init(&pool->cond);
nn_mutex_init(&pool->mutex);
@@ -67,20 +61,16 @@ void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size, struct nn_event_
pool->userdata = userdata;
}
-// This defeats the entire purpose of a packet pool
-// but is useful for testing.
-//#define PACKET_POOL_GROW
-
struct nn_packet *nn_packet_pool_get(struct nn_packet_pool *pool)
{
nn_mutex_lock(&pool->mutex);
if (!pool->disabled && !pool->empty.size) {
-#ifdef PACKET_POOL_GROW
- al_array_push(pool->empty, nn_packet_create());
-#else
- nn_cond_wait(&pool->cond, &pool->mutex);
-#endif
+ if (pool->grow) {
+ al_array_push(pool->empty, nn_packet_create());
+ } else {
+ nn_cond_wait(&pool->cond, &pool->mutex);
+ }
}
if (pool->disabled) {
@@ -118,10 +108,21 @@ void nn_packet_pool_flush(struct nn_packet_pool *pool)
ev_async_send(pool->loop->ev, &pool->signal);
}
-void nn_packet_pool_return(struct nn_packet_pool *pool, struct nn_packet *packet)
+void nn_packet_pool_lock(struct nn_packet_pool *pool)
{
nn_mutex_lock(&pool->mutex);
+}
+
+void nn_packet_pool_return(struct nn_packet_pool *pool, struct nn_packet *packet)
+{
return_internal(pool, packet);
+}
+
+void nn_packet_pool_unlock(struct nn_packet_pool *pool)
+{
+ if (!pool->disabled && pool->empty.size > 0 && nn_cond_is_waiting(&pool->cond)) {
+ nn_cond_signal(&pool->cond);
+ }
nn_mutex_unlock(&pool->mutex);
}
@@ -169,13 +170,10 @@ void nn_packet_pool_free(struct nn_packet_pool *pool)
nn_cond_destroy(&pool->cond);
struct nn_packet *packet;
- al_array_foreach(pool->empty, i, packet) {
- nn_packet_free(packet);
- }
+ al_array_foreach(pool->empty, i, packet) nn_packet_free(packet);
+ al_array_foreach(pool->ready, i, packet) nn_packet_free(packet);
+ al_array_foreach(pool->sending, i, packet) nn_packet_free(packet);
al_array_free(pool->empty);
-
- al_array_foreach(pool->ready, i, packet) {
- nn_packet_free(packet);
- }
al_array_free(pool->ready);
+ al_array_free(pool->sending);
}
diff --git a/src/packet_pool.h b/src/packet_pool.h
index e20e29a..40e7169 100644
--- a/src/packet_pool.h
+++ b/src/packet_pool.h
@@ -24,22 +24,26 @@ enum {
struct nn_packet_pool {
struct nn_event_loop *loop;
u32 size;
+ bool grow;
array(struct nn_packet *) empty;
array(struct nn_packet *) ready;
+ array(struct nn_packet *) sending;
bool disabled;
struct nn_cond cond;
struct nn_mutex mutex;
ev_async signal;
- u8 (*callback)(void *, struct nn_packet *);
+ void (*callback)(void *, struct nn_packet *);
void *userdata;
};
-void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size, struct nn_event_loop *loop,
- u8 (*callback)(void *, struct nn_packet *), void *userdata);
+void nn_packet_pool_init(struct nn_packet_pool *pool, u32 size,
+ struct nn_event_loop *loop, void (*callback)(void *, struct nn_packet *), void *userdata);
struct nn_packet *nn_packet_pool_get(struct nn_packet_pool *pool);
void nn_packet_pool_submit(struct nn_packet_pool *pool, struct nn_packet *packet);
void nn_packet_pool_flush(struct nn_packet_pool *pool);
+void nn_packet_pool_lock(struct nn_packet_pool *pool);
void nn_packet_pool_return(struct nn_packet_pool *pool, struct nn_packet *packet);
+void nn_packet_pool_unlock(struct nn_packet_pool *pool);
void nn_packet_pool_disable(struct nn_packet_pool *pool);
void nn_packet_pool_enable(struct nn_packet_pool *pool);
struct nn_packet *nn_packet_pool_pop(struct nn_packet_pool *pool);
diff --git a/src/packet_stream.c b/src/packet_stream.c
index 54fcf94..e3dfa0a 100644
--- a/src/packet_stream.c
+++ b/src/packet_stream.c
@@ -1,3 +1,4 @@
+#include "multiplex.h"
#include "packet_stream.h"
enum {
@@ -15,31 +16,25 @@ static inline void init_io_state(struct nn_packet_stream *stream)
stream->in.index = 0;
al_array_init(stream->out.queue);
stream->out.packet = NULL;
- stream->out.active = false;
+ stream->out.running = false;
}
-bool nn_packet_stream_init(struct nn_packet_stream *stream, u8 type, u8 id,
+void nn_packet_stream_init(struct nn_packet_stream *stream,
bool (*connection_callback)(void *, struct nn_packet_stream *),
void (*connection_closed_callback)(void *, struct nn_packet_stream *), void *userdata)
{
- stream->sock.type = type;
- if (!nn_socket_init(&stream->sock, NNWT_SOCKET_NONBLOCKING)) {
- return false;
- }
-
- stream->id = id;
-
init_io_state(stream);
+ stream->direct = NULL;
+
stream->connection_callback = connection_callback;
stream->connection_closed_callback = connection_closed_callback;
stream->packet_callback = NULL;
stream->packet_sent_callback = NULL;
+ stream->packets_sent_callback = NULL;
stream->userdata = userdata;
stream->connect = PACKET_STREAM_DISCONNECTED;
-
- return true;
}
void nn_packet_stream_set_nodelay(struct nn_packet_stream *stream, s32 nodelay)
@@ -49,13 +44,13 @@ void nn_packet_stream_set_nodelay(struct nn_packet_stream *stream, s32 nodelay)
static void start_write_internal(struct nn_packet_stream *stream)
{
- stream->out.active = true;
+ stream->out.running = true;
ev_io_start(stream->loop->ev, &stream->wevent);
}
static void stop_write_internal(struct nn_packet_stream *stream)
{
- stream->out.active = false;
+ stream->out.running = false;
ev_io_stop(stream->loop->ev, &stream->wevent);
}
@@ -85,9 +80,14 @@ static void stop_internal(struct nn_packet_stream *stream)
stream->out.packet = NULL;
}
- struct nn_packet *packet;
- al_array_foreach(stream->out.queue, i, packet) {
- stream->packet_sent_callback(stream->userdata, packet);
+ if (stream->packets_sent_callback) {
+ u32 count = stream->out.queue.size;
+ stream->packets_sent_callback(stream->userdata, stream->out.queue.data, count);
+ } else {
+ struct nn_packet *packet;
+ al_array_foreach(stream->out.queue, i, packet) {
+ stream->packet_sent_callback(stream->userdata, packet);
+ }
}
stream->out.queue.size = 0;
@@ -106,7 +106,7 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
ssize_t ret = nn_socket_read(&stream->sock, ptr, size - stream->in.index);
if (ret <= 0 || stream->connect == PACKET_STREAM_DISCONNECTING) {
ev_io_stop(stream->loop->ev, &stream->revent);
- if (stream->out.active) {
+ if (stream->out.running) {
stop_write_internal(stream);
}
stop_internal(stream);
@@ -131,6 +131,17 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents)
}
}
+static bool stream_connected(struct nn_packet_stream *stream)
+{
+ stream->connect = PACKET_STREAM_CONNECTED;
+ al_assert(stream->corked);
+ if (stream->connection_callback(stream->userdata, stream)) {
+ stream->corked = false;
+ }
+ al_assert(stream->packet_callback && stream->packet_sent_callback);
+ return !stream->corked;
+}
+
static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
{
(void)loop;
@@ -149,14 +160,9 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents)
// Try again on next POLLOUT, not sure if this can actually happen.
return;
}
-
- stream->connect = PACKET_STREAM_CONNECTED;
- al_assert(stream->corked);
- if (stream->connection_callback(stream->userdata, stream)) {
- stream->corked = false;
+ if (stream_connected(stream)) {
ev_io_start(stream->loop->ev, &stream->revent);
}
- al_assert(stream->packet_callback && stream->packet_sent_callback);
}
if (!stream->out.packet) {
@@ -191,6 +197,7 @@ void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_eve
stream->loop = loop;
stream->sock = *sock;
+ nn_socket_set_blocking(&stream->sock, false);
init_io_state(stream);
@@ -200,17 +207,18 @@ void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_eve
stream->revent.data = stream;
ev_io_init(&stream->revent, stream_read_callback, nn_socket_get_fd(&stream->sock), EV_READ);
- stream->connect = PACKET_STREAM_CONNECTED;
- al_assert(stream->corked);
- if (stream->connection_callback(stream->userdata, stream)) {
- stream->corked = false;
+ if (stream_connected(stream)) {
ev_io_start(stream->loop->ev, &stream->revent);
}
- al_assert(stream->packet_callback && stream->packet_sent_callback);
}
static void do_connect_internal(struct nn_packet_stream *stream, str *addr, u16 port)
{
+ if (!nn_socket_init(&stream->sock, NNWT_SOCKET_NONBLOCKING)) {
+ stream->connection_closed_callback(stream->userdata, stream);
+ return;
+ }
+
if (!nn_socket_connect(&stream->sock, addr, port)) {
nn_socket_close(&stream->sock);
stream->connection_closed_callback(stream->userdata, stream);
@@ -228,26 +236,29 @@ static void do_connect_internal(struct nn_packet_stream *stream, str *addr, u16
start_write_internal(stream);
}
-void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop, str *addr, u16 port)
+void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop,
+ u8 id, u8 type, str *addr, u16 port)
{
+ stream->id = id;
stream->loop = loop;
+ stream->sock.type = type;
do_connect_internal(stream, addr, port);
}
void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port)
{
al_assert(stream->connect == PACKET_STREAM_DISCONNECTED);
-
- if (!nn_socket_init(&stream->sock, NNWT_SOCKET_NONBLOCKING)) {
- stream->connection_closed_callback(stream->userdata, stream);
- return;
- }
-
do_connect_internal(stream, addr, port);
}
+bool nn_packet_stream_connected(struct nn_packet_stream *stream)
+{
+ return stream_connected(stream);
+}
+
void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork)
{
+ if (stream->direct) return;
if (stream->connect == PACKET_STREAM_CONNECTED) {
if (cork && !stream->corked) {
ev_io_stop(stream->loop->ev, &stream->revent);
@@ -267,15 +278,52 @@ bool nn_packet_stream_send_packet(struct nn_packet_stream *stream, struct nn_pac
al_assert(stream->connect == PACKET_STREAM_CONNECTED);
nn_packet_write_size(packet);
+
+ if (stream->direct) {
+ struct nn_packet_stream *direct = stream->direct;
+ direct->packet_callback(direct->userdata, direct, packet);
+ return true;
+ }
+
al_array_push(stream->out.queue, packet);
- if (!stream->out.active) {
+ if (!stream->out.running) {
start_write_internal(stream);
}
return true;
}
+void nn_packet_stream_return_packet(struct nn_packet_stream *stream, struct nn_packet *packet)
+{
+ if (stream->direct) {
+ struct nn_packet_stream *direct = stream->direct;
+ direct->packet_sent_callback(direct->userdata, packet);
+ } else {
+ nn_packet_free(packet);
+ }
+}
+
+void nn_packet_stream_return_packets(struct nn_packet_stream *stream, struct nn_packet **packets, u32 count)
+{
+ if (stream->direct) {
+ struct nn_packet_stream *direct = stream->direct;
+ if (direct->packets_sent_callback) {
+ direct->packets_sent_callback(direct->userdata, packets, count);
+ } else {
+ for (u32 i = 0; i < count; i++) {
+ if (packets[i]) {
+ direct->packet_sent_callback(direct->userdata, packets[i]);
+ }
+ }
+ }
+ } else {
+ for (u32 i = 0; i < count; i++) {
+ if (packets[i]) nn_packet_free(packets[i]);
+ }
+ }
+}
+
void nn_packet_stream_disconnect(struct nn_packet_stream *stream)
{
al_assert(stream->connect != PACKET_STREAM_DISCONNECTING);
@@ -286,6 +334,24 @@ void nn_packet_stream_disconnect(struct nn_packet_stream *stream)
return;
}
+ if (stream->direct) {
+ struct nn_packet_stream *direct = stream->direct;
+ direct->connect = PACKET_STREAM_DISCONNECTED;
+ stream->connect = PACKET_STREAM_DISCONNECTED;
+ direct->corked = true;
+ stream->corked = true;
+ struct nn_packet_stream *bridge = nn_multiplex_direct_get_bridge();
+ bridge->connection_callback = NULL;
+ bridge->connection_closed_callback = NULL;
+ bridge->packet_sent_callback = direct->packet_sent_callback;
+ bridge->packets_sent_callback = direct->packets_sent_callback;
+ bridge->userdata = direct->userdata;
+ direct->connection_closed_callback(direct->userdata, direct);
+ stream->direct = bridge;
+ stream->connection_closed_callback(stream->userdata, stream);
+ return;
+ }
+
if (stream->connect == PACKET_STREAM_CONNECTING) {
// wevent is always active when connect = CONNECTING.
stop_write_internal(stream);
@@ -304,7 +370,7 @@ void nn_packet_stream_free(struct nn_packet_stream *stream)
{
al_assert(stream->in.packet);
// Loosely assert that stop_internal() has run.
- al_assert(!stream->out.active);
+ al_assert(!stream->out.running);
nn_packet_free(stream->in.packet);
al_array_free(stream->out.queue);
}
diff --git a/src/packet_stream.h b/src/packet_stream.h
index 40e5c3c..6eae962 100644
--- a/src/packet_stream.h
+++ b/src/packet_stream.h
@@ -17,30 +17,36 @@ struct nn_packet_stream {
bool corked;
struct {
struct nn_packet *packet;
- bool have_header;
size_t index;
+ bool have_header;
} in;
struct {
array(struct nn_packet *) queue;
struct nn_packet *packet;
- bool active;
size_t index;
+ bool running;
} out;
+ struct nn_packet_stream *direct;
bool (*connection_callback)(void *, struct nn_packet_stream *);
void (*connection_closed_callback)(void *, struct nn_packet_stream *);
void (*packet_callback)(void *, struct nn_packet_stream *, struct nn_packet *);
void (*packet_sent_callback)(void *, struct nn_packet *);
+ void (*packets_sent_callback)(void *, struct nn_packet **, u32);
void *userdata;
};
-bool nn_packet_stream_init(struct nn_packet_stream *stream, u8 type, u8 id,
+void nn_packet_stream_init(struct nn_packet_stream *stream,
bool (*connection_callback)(void *, struct nn_packet_stream *),
void (*connection_closed_callback)(void *, struct nn_packet_stream *), void *userdata);
void nn_packet_stream_set_nodelay(struct nn_packet_stream *stream, s32 no_delay);
void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_event_loop *loop, struct nn_socket *sock);
-void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop, str *addr, u16 port);
+void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop,
+ u8 id, u8 type, str *addr, u16 port);
void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port);
+bool nn_packet_stream_connected(struct nn_packet_stream *stream);
void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork);
bool nn_packet_stream_send_packet(struct nn_packet_stream *stream, struct nn_packet *packet);
+void nn_packet_stream_return_packet(struct nn_packet_stream *stream, struct nn_packet *packet);
+void nn_packet_stream_return_packets(struct nn_packet_stream *stream, struct nn_packet **packets, u32 count);
void nn_packet_stream_disconnect(struct nn_packet_stream *stream);
void nn_packet_stream_free(struct nn_packet_stream *stream);
diff --git a/src/rpc.c b/src/rpc.c
index cdac4c6..ae8b689 100644
--- a/src/rpc.c
+++ b/src/rpc.c
@@ -23,13 +23,14 @@ void nn_rpc_add_command(struct nn_rpc *rpc, struct nn_rpc_command *command)
static void packet_callback(void *userdata, struct nn_packet_stream *stream, struct nn_packet *packet)
{
struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata;
+ al_assert(conn->stream == stream);
s8 op = nn_packet_read_s8(packet);
u32 id = nn_packet_read_u32(packet);
if (op == -1) { // Response
struct nn_rpc_callback *callback;
al_array_foreach_ptr(conn->callbacks, i, callback) {
if (callback->id == id) {
- callback->callback(callback->userdata, packet);
+ callback->callback(callback->userdata, conn, packet);
al_array_remove_at(conn->callbacks, i);
return;
}
@@ -51,7 +52,7 @@ static void packet_callback(void *userdata, struct nn_packet_stream *stream, str
}
}
}
- nn_packet_free(packet);
+ nn_packet_stream_return_packet(stream, packet);
}
static void packet_sent_callback(void *userdata, struct nn_packet *packet)
@@ -61,7 +62,7 @@ static void packet_sent_callback(void *userdata, struct nn_packet *packet)
al_assert(conn->outgoing > 0);
conn->outgoing--;
if (conn->flushing && conn->outgoing == 0) {
- nn_packet_stream_disconnect(&conn->stream);
+ nn_packet_stream_disconnect(conn->stream);
}
}
@@ -89,7 +90,7 @@ static bool stream_connection_callback(void *userdata, struct nn_packet_stream *
static void stream_connection_closed_callback(void *userdata, struct nn_packet_stream *stream)
{
struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata;
- al_assert(stream == &conn->stream);
+ al_assert(stream == conn->stream);
conn->rpc->connection_closed_callback(conn->rpc->userdata, conn);
}
@@ -101,47 +102,41 @@ static inline void init_rpc_connection(struct nn_rpc *rpc, struct nn_rpc_connect
conn->flushing = false;
}
-void nn_rpc_add_socket(struct nn_rpc *rpc, struct nn_socket *sock)
+void nn_rpc_add_stream(struct nn_rpc *rpc, struct nn_packet_stream *stream)
{
struct nn_rpc_connection *conn = al_alloc_object(struct nn_rpc_connection);
init_rpc_connection(rpc, conn);
- conn->stream.connection_callback = stream_connection_callback;
- conn->stream.connection_closed_callback = stream_connection_closed_callback;
- conn->stream.userdata = conn;
-
- nn_packet_stream_from_socket(&conn->stream, rpc->loop, sock);
+ conn->stream = stream;
+ conn->stream->connection_callback = stream_connection_callback;
+ conn->stream->connection_closed_callback = stream_connection_closed_callback;
+ conn->stream->userdata = conn;
}
-bool nn_rpc_prepare_client(struct nn_rpc *rpc, u8 type, u8 id)
+void nn_rpc_prepare_client(struct nn_rpc *rpc)
{
al_assert(!rpc->conn);
struct nn_rpc_connection *conn = al_alloc_object(struct nn_rpc_connection);
init_rpc_connection(rpc, conn);
- if (!nn_packet_stream_init(&conn->stream, type, id,
- stream_connection_callback, stream_connection_closed_callback, conn)) {
- al_free(conn);
- return false;
- }
+ conn->stream = al_alloc_object(struct nn_packet_stream);
+ nn_packet_stream_init(conn->stream, stream_connection_callback, stream_connection_closed_callback, conn);
rpc->conn = conn;
al_array_push(rpc->connections, conn);
-
- return true;
}
-void nn_rpc_connect(struct nn_rpc *rpc, str *addr, u16 port)
+void nn_rpc_connect(struct nn_rpc *rpc, u8 id, u8 type, str *addr, u16 port)
{
al_assert(rpc->conn);
- nn_packet_stream_connect(&rpc->conn->stream, rpc->loop, addr, port);
+ nn_packet_stream_connect(rpc->conn->stream, rpc->loop, id, type, addr, port);
}
void nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port)
{
al_assert(rpc->conn);
- nn_packet_stream_reconnect(&rpc->conn->stream, addr, port);
+ nn_packet_stream_reconnect(rpc->conn->stream, addr, port);
}
struct nn_packet *nn_rpc_get_packet(struct nn_rpc *rpc, s8 op)
@@ -157,8 +152,9 @@ void nn_rpc_free(struct nn_rpc *rpc)
{
struct nn_rpc_connection *conn;
al_array_foreach(rpc->connections, i, conn) {
- nn_packet_stream_free(&conn->stream);
+ nn_packet_stream_free(conn->stream);
al_array_free(conn->callbacks);
+ al_free(conn->stream);
al_free(conn);
}
al_array_free(rpc->connections);
@@ -166,7 +162,7 @@ void nn_rpc_free(struct nn_rpc *rpc)
}
void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet *packet,
- void (*callback)(void *, struct nn_packet *), void *userdata)
+ void (*callback)(void *, struct nn_rpc_connection *conn, struct nn_packet *), void *userdata)
{
if (callback) {
al_array_push(conn->callbacks, ((struct nn_rpc_callback){
@@ -176,7 +172,7 @@ void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet
}));
}
conn->outgoing++;
- nn_packet_stream_send_packet(&conn->stream, packet);
+ nn_packet_stream_send_packet(conn->stream, packet);
}
void nn_rpc_conn_flush(struct nn_rpc_connection *conn)
@@ -184,11 +180,11 @@ void nn_rpc_conn_flush(struct nn_rpc_connection *conn)
if (conn->outgoing != 0) {
conn->flushing = true;
} else {
- nn_packet_stream_disconnect(&conn->stream);
+ nn_packet_stream_disconnect(conn->stream);
}
}
void nn_rpc_conn_disconnect(struct nn_rpc_connection *conn)
{
- nn_packet_stream_disconnect(&conn->stream);
+ nn_packet_stream_disconnect(conn->stream);
}
diff --git a/src/rpc2.h b/src/rpc2.h
index ab89319..ecfb91a 100644
--- a/src/rpc2.h
+++ b/src/rpc2.h
@@ -6,14 +6,15 @@
#include "packet_stream.h"
+struct nn_rpc_connection;
struct nn_rpc_callback {
u32 id;
- void (*callback)(void *, struct nn_packet *);
+ void (*callback)(void *, struct nn_rpc_connection *, struct nn_packet *);
void *userdata;
};
struct nn_rpc_connection {
- struct nn_packet_stream stream;
+ struct nn_packet_stream *stream;
array(struct nn_rpc_callback) callbacks;
u32 outgoing;
bool flushing;
@@ -41,15 +42,15 @@ bool nn_rpc_init(struct nn_rpc *rpc, struct nn_event_loop *loop,
void (*connection_callback)(void *, struct nn_rpc_connection *),
void (*connection_closed_callback)(void *, struct nn_rpc_connection *), void *userdata);
void nn_rpc_add_command(struct nn_rpc *rpc, struct nn_rpc_command *command);
-void nn_rpc_add_socket(struct nn_rpc *rpc, struct nn_socket *sock);
-bool nn_rpc_prepare_client(struct nn_rpc *rpc, u8 type, u8 id);
-void nn_rpc_connect(struct nn_rpc *rpc, str *addr, u16 port);
+void nn_rpc_add_stream(struct nn_rpc *rpc, struct nn_packet_stream *stream);
+void nn_rpc_prepare_client(struct nn_rpc *rpc);
+void nn_rpc_connect(struct nn_rpc *rpc, u8 id, u8 type, str *addr, u16 port);
void nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port);
struct nn_packet *nn_rpc_get_packet(struct nn_rpc *rpc, s8 op);
void nn_rpc_free(struct nn_rpc *rpc);
// The order of commands per connection will be respected.
void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet *packet,
- void (*callback)(void *, struct nn_packet *), void *userdata);
+ void (*callback)(void *, struct nn_rpc_connection *conn, struct nn_packet *), void *userdata);
void nn_rpc_conn_flush(struct nn_rpc_connection *conn);
void nn_rpc_conn_disconnect(struct nn_rpc_connection *conn);
diff --git a/src/socket/socket.h b/src/socket/socket.h
index 6b53812..bb1fa3f 100644
--- a/src/socket/socket.h
+++ b/src/socket/socket.h
@@ -3,7 +3,9 @@
#include <al/types.h>
#include <al/str.h>
-#ifndef _WIN32
+#ifdef NAUNET_ON_WINDOWS
+#include "../winwrap.h"
+#else
#include <unistd.h>
#include <sys/un.h>
#include <sys/eventfd.h>
@@ -12,8 +14,6 @@
#include <netinet/tcp.h>
#include <netdb.h>
#include <poll.h>
-#else
-#include "../winwrap.h"
#endif
enum {
@@ -29,23 +29,23 @@ enum {
struct nn_socket {
u8 type;
-#ifndef _WIN32
- s32 fd;
- struct addrinfo *addrinfo;
- struct sockaddr_un addr_un;
-#else
+#ifdef NAUNET_ON_WINDOWS
SOCKET fd;
s32 internal_fd;
SOCKADDR_IN addr_in;
+#else
+ s32 fd;
+ struct addrinfo *addrinfo;
+ struct sockaddr_un addr_un;
#endif
};
-#ifndef _WIN32
-#define nn_read read
-#define nn_write write
-#else
+#ifdef NAUNET_ON_WINDOWS
#define nn_read _read
#define nn_write _write
+#else
+#define nn_read read
+#define nn_write write
#endif
#define nn_htons htons
@@ -53,7 +53,7 @@ struct nn_socket {
#define nn_ntohs ntohs
#define nn_ntohl ntohl
-#ifndef _WIN32 // Posix-only helpers.
+#ifndef NAUNET_ON_WINDOWS // Posix-only helpers.
#define nn_pollfd pollfd
#define nn_nfds nfds_t
void nn_fd_set_blocking(s32 fd, bool blocking);
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);
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 22:14:33 +0000