From 8b5c569b53f9287e8b6c84e4402ec6f796e4cb6b Mon Sep 17 00:00:00 2001 From: Andrew Opalach Date: Wed, 1 Jan 2025 15:44:14 -0500 Subject: Add direct (same process) mode for packet stream Signed-off-by: Andrew Opalach --- src/common.c | 51 +++++++++++--- src/curl/http.c | 79 ++++++++++++---------- src/curl/http.h | 8 +-- src/ev_embed_compat.c | 2 +- src/fs_event/fs_event.h | 8 +-- src/multiplex.c | 49 +++++++++++++- src/multiplex.h | 17 ++++- src/packet_cache.c | 14 ++-- src/packet_cache.h | 3 +- src/packet_pool.c | 68 +++++++++---------- src/packet_pool.h | 10 ++- src/packet_stream.c | 140 ++++++++++++++++++++++++++++----------- src/packet_stream.h | 14 ++-- src/rpc.c | 48 ++++++-------- src/rpc2.h | 13 ++-- src/socket/socket.h | 26 ++++---- src/util/error.h | 6 +- src/util/file/file.h | 21 +++--- src/util/file/file_linux.c | 27 ++++---- src/util/file/file_stdio.c | 16 ++--- src/util/file/util.c | 4 +- src/util/packet.c | 11 +++ src/util/packet.h | 5 ++ src/util/thread/thread.h | 49 +++++++++----- src/util/thread/thread_linux.c | 8 +++ src/util/thread/thread_windows.c | 6 ++ 26 files changed, 466 insertions(+), 237 deletions(-) (limited to 'src') 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 +#include #include #include -#ifndef _WIN32 -#include -#else +#ifdef NAUNET_ON_WINDOWS #include "winwrap.h" +#else +#include #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 -#ifndef _WIN32 -#include -#define NNWT_CLOSE_WRITE IN_CLOSE_WRITE -#else +#ifdef NAUNET_ON_WINDOWS struct inotify_event {}; #define NNWT_CLOSE_WRITE 0 +#else +#include +#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 #include -#ifndef _WIN32 +#ifdef NAUNET_ON_WINDOWS +#include "../winwrap.h" +#else #include #include #include @@ -12,8 +14,6 @@ #include #include #include -#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 #else -#ifndef _WIN32 +#ifdef NAUNET_ON_WINDOWS +// TODO +#else #include #include #include @@ -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 -#ifndef _WIN32 -#include -#else +#ifdef NAUNET_ON_WINDOWS #include "../../winwrap.h" +#else +#include #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, ¶ms); +} + 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); -- cgit v1.2.3-101-g0448