diff options
Diffstat (limited to 'src')
53 files changed, 1405 insertions, 1235 deletions
diff --git a/src/common.c b/src/common.c index ec2c2a9..a24592a 100644 --- a/src/common.c +++ b/src/common.c @@ -1,7 +1,6 @@ #include <al/lib.h> #include <al/random.h> #include <locale.h> -#include <signal.h> #ifndef _WIN32 #include <unistd.h> @@ -11,18 +10,26 @@ #include "common.h" -bool aki_common_init(void) +#ifdef AL_MEMORY_TRACKING +#include "util/thread/thread.h" + +static struct nn_mutex malloc_mutex; + +static void malloc_lock(void) { - if (!setlocale(LC_ALL, "")) return false; + nn_mutex_lock(&malloc_mutex); +} -#ifdef AL_MEMORY_TRACKING - al_malloc_init(); -#endif - al_rand_init(); -#ifdef _WIN32 - aki_windows_init(); +static void malloc_unlock(void) +{ + nn_mutex_unlock(&malloc_mutex); +} #endif +bool nn_common_init(void) +{ + if (!setlocale(LC_ALL, "")) return false; + #ifndef _WIN32 al_set_page_size(sysconf(_SC_PAGESIZE)); #else @@ -31,19 +38,36 @@ bool aki_common_init(void) al_set_page_size(info.dwPageSize); #endif -#ifndef _WIN32 - signal(SIGPIPE, SIG_IGN); +#ifdef AL_MEMORY_TRACKING + al_malloc_init(); + nn_mutex_init(&malloc_mutex); + al_set_alloc(malloc, calloc, realloc, +#ifdef AL_HAVE_POSIX_MEMALIGN + posix_memalign, +#endif + free, malloc_lock, malloc_unlock); +#endif + al_rand_init(); +#ifdef _WIN32 + nn_windows_init(); #endif return true; } -void aki_common_close(void) +void nn_common_close(void) { #ifdef _WIN32 - aki_windows_close(); + nn_windows_close(); #endif #ifdef AL_MEMORY_TRACKING al_malloc_close(); + size_t current; + size_t peak; + size_t total; + al_malloc_stats(¤t, &peak, &total); + al_printf("malloc stats, current: %.2fKB, peak: %.2fKB, total: %.2fKB.\n", + current/1024.f, peak/1024.f, total/1024.f); + nn_mutex_destroy(&malloc_mutex); #endif } diff --git a/src/common.h b/src/common.h index c70cc24..6b02bcf 100644 --- a/src/common.h +++ b/src/common.h @@ -2,5 +2,5 @@ #include <al/types.h> -bool aki_common_init(void); -void aki_common_close(void); +bool nn_common_init(void); +void nn_common_close(void); diff --git a/src/curl/curl.c b/src/curl/curl.c index b96d46f..c996011 100644 --- a/src/curl/curl.c +++ b/src/curl/curl.c @@ -3,12 +3,13 @@ #include "curl.h" // TODO: https://curl.se/libcurl/c/externalsocket.html +// TODO: Use timer again/repeat. static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_curl *curl = (struct aki_curl *)w->data; - s32 action = ((revents & EV_READ) ? CURL_CSELECT_IN : 0) | ((revents & EV_WRITE) ? CURL_CSELECT_OUT : 0); + struct nn_curl *curl = (struct nn_curl *)w->data; + s32 action = (revents & EV_READ ? CURL_CSELECT_IN : 0) | (revents & EV_WRITE ? CURL_CSELECT_OUT : 0); s32 running; CURLMcode mc = curl_multi_socket_action(curl->multi_handle, curl->sock, action, &running); if (mc != CURLM_OK) { @@ -18,10 +19,10 @@ static void curl_socket_action_callback(struct ev_loop *loop, ev_io *w, s32 reve curl->handle_events(curl->userdata, curl); } -static void set_sock(struct aki_curl *curl, s32 what) +static void set_sock(struct nn_curl *curl, s32 what) { - s32 action = ((what & CURL_POLL_IN) ? EV_READ : 0) | ((what & CURL_POLL_OUT) ? EV_WRITE : 0); if (curl->event.active) ev_io_stop(curl->loop->ev, &curl->event); + s32 action = (what & CURL_POLL_IN ? EV_READ : 0) | (what & CURL_POLL_OUT ? EV_WRITE : 0); ev_io_init(&curl->event, curl_socket_action_callback, curl->sock, action); ev_io_start(curl->loop->ev, &curl->event); } @@ -29,7 +30,7 @@ static void set_sock(struct aki_curl *curl, s32 what) static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *sockp) { (void)e; - struct aki_curl *curl = (struct aki_curl *)cbp; + struct nn_curl *curl = (struct nn_curl *)cbp; if (what == CURL_POLL_REMOVE) { ev_io_stop(curl->loop->ev, &curl->event); curl_multi_assign(curl->multi_handle, curl->sock, NULL); @@ -47,10 +48,14 @@ static s32 sock_callback(CURL *e, curl_socket_t s, s32 what, void *cbp, void *so static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents) { (void)loop; - struct aki_curl *curl = (struct aki_curl *)w->data; + struct nn_curl *curl = (struct nn_curl *)w->data; (void)revents; - ev_timer_stop(curl->loop->ev, w); - curl->timer_started = false; + + if (curl->timer_started) { + ev_timer_stop(curl->loop->ev, w); + curl->timer_started = false; + } + s32 running; curl_multi_socket_action(curl->multi_handle, CURL_SOCKET_TIMEOUT, 0, &running); curl->handle_events(curl->userdata, curl); @@ -59,58 +64,64 @@ static void timeout_callback(struct ev_loop *loop, ev_timer *w, s32 revents) static s32 timer_callback(CURLM *multi, s64 timeout_ms, void *userp) { (void)multi; - struct aki_curl *curl = (struct aki_curl *)userp; - if (timeout_ms == -1) { - ev_timer_stop(curl->loop->ev, &curl->timer); - return 0; - } + struct nn_curl *curl = (struct nn_curl *)userp; + if (curl->timer_started) { ev_timer_stop(curl->loop->ev, &curl->timer); + curl->timer_started = false; } + + if (timeout_ms == -1) + return 0; + curl->timer.data = curl; ev_timer_init(&curl->timer, timeout_callback, timeout_ms / 1000.0, 0.0); + curl->timer_started = true; ev_timer_start(curl->loop->ev, &curl->timer); + return 0; } -bool aki_curl_init(struct aki_curl *curl) +bool nn_curl_init(struct nn_curl *curl) { curl->handle = curl_easy_init(); if (!curl->handle) goto err; -#ifdef _DEBUG_ //curl_easy_setopt(curl->handle, CURLOPT_VERBOSE, 1L); -#endif curl_easy_setopt(curl->handle, CURLOPT_NOPROGRESS, 1L); + curl->multi_handle = curl_multi_init(); if (!curl->multi_handle) goto err; curl_multi_setopt(curl->multi_handle, CURLMOPT_SOCKETFUNCTION, sock_callback); curl_multi_setopt(curl->multi_handle, CURLMOPT_SOCKETDATA, curl); curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERFUNCTION, timer_callback); curl_multi_setopt(curl->multi_handle, CURLMOPT_TIMERDATA, curl); - curl->added = false; - curl->timer_started = false; + curl->loop = NULL; curl->event.data = curl; curl->event.active = 0; + curl->timer_started = false; + curl->added = false; + return true; err: - aki_curl_close(curl); + nn_curl_close(curl); + return false; } -void aki_curl_set_url(struct aki_curl *curl, str *url) +void nn_curl_set_url(struct nn_curl *curl, str *url) { char *c_str = al_str_to_c_str(url); curl_easy_setopt(curl->handle, CURLOPT_URL, c_str); al_free(c_str); } -bool aki_curl_add_handle(struct aki_curl *curl) +bool nn_curl_add_handle(struct nn_curl *curl) { al_assert(!curl->added); - // CURLM_RECURSIVE_API_CALL (8) also means we can't even call this - // while a callback is running on another thread. + // CURLM_RECURSIVE_API_CALL (8) also means we can't call this while + // a callback is running on another thread. CURLMcode mc = curl_multi_add_handle(curl->multi_handle, curl->handle); if (mc != CURLM_OK) { al_log_error("curl", "curl_multi_add_handle() failed (%s).", curl_multi_strerror(mc)); @@ -120,7 +131,7 @@ bool aki_curl_add_handle(struct aki_curl *curl) return true; } -bool aki_curl_remove_handle(struct aki_curl *curl) +bool nn_curl_remove_handle(struct nn_curl *curl) { al_assert(curl->added); CURLMcode mc = curl_multi_remove_handle(curl->multi_handle, curl->handle); @@ -132,11 +143,10 @@ bool aki_curl_remove_handle(struct aki_curl *curl) return true; } -void aki_curl_close(struct aki_curl *curl) +void nn_curl_close(struct nn_curl *curl) { - if (curl->handle && curl->multi_handle && curl->added) { - aki_curl_remove_handle(curl); - } + if (curl->handle && curl->multi_handle && curl->added) + nn_curl_remove_handle(curl); if (curl->handle) curl_easy_cleanup(curl->handle); if (curl->multi_handle) curl_multi_cleanup(curl->multi_handle); } diff --git a/src/curl/curl.h b/src/curl/curl.h index f3a1bfc..7a5e31f 100644 --- a/src/curl/curl.h +++ b/src/curl/curl.h @@ -5,21 +5,21 @@ #include "../loop.h" -struct aki_curl { +struct nn_curl { + struct nn_event_loop *loop; ev_io event; ev_timer timer; bool timer_started; - struct aki_event_loop *loop; CURL *handle; CURLM *multi_handle; bool added; curl_socket_t sock; - void (*handle_events)(void *, struct aki_curl *); + void (*handle_events)(void *, struct nn_curl *); void *userdata; }; -bool aki_curl_init(struct aki_curl *curl); -void aki_curl_set_url(struct aki_curl *curl, str *url); -bool aki_curl_add_handle(struct aki_curl *curl); -bool aki_curl_remove_handle(struct aki_curl *curl); -void aki_curl_close(struct aki_curl *curl); +bool nn_curl_init(struct nn_curl *curl); +void nn_curl_set_url(struct nn_curl *curl, str *url); +bool nn_curl_add_handle(struct nn_curl *curl); +bool nn_curl_remove_handle(struct nn_curl *curl); +void nn_curl_close(struct nn_curl *curl); diff --git a/src/curl/http.c b/src/curl/http.c index b3c13a2..00d5cd9 100644 --- a/src/curl/http.c +++ b/src/curl/http.c @@ -2,59 +2,69 @@ #include "http.h" -static void check_status_codes(struct aki_curl *curl) +static void check_status_codes(struct nn_curl *curl) { - struct aki_http *http = (struct aki_http *)curl; - if (http->response_code <= 0) { - curl_easy_getinfo(curl->handle, CURLINFO_RESPONSE_CODE, &http->response_code); - if (http->response_code > 0) { - http->callback(http->userdata, AKI_HTTP_RESPONSE_CODE, NULL, http->response_code); - } + struct nn_http *http = (struct nn_http *)curl; + + if (http->status_code <= 0) { + 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->response_code == 200 && http->content_length < 0) { + + if (http->content_length < 0 && http->status_code == 200) { curl_easy_getinfo(curl->handle, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &http->content_length); - if (http->content_length >= 0) { - http->callback(http->userdata, AKI_HTTP_CONTENT_LENGTH, NULL, (s64)http->content_length); - } + if (http->content_length >= 0) + http->callback(http->userdata, NNWT_HTTP_CONTENT_LENGTH, NULL, (s64)http->content_length); } } -static void http_handle_events(void *userdata, struct aki_curl *curl) +static void http_handle_events(void *userdata, struct nn_curl *curl) { (void)userdata; s32 pending; CURLMsg *msg; - struct aki_http *http = (struct aki_http *)curl; + struct nn_http *http = (struct nn_http *)curl; while ((msg = curl_multi_info_read(curl->multi_handle, &pending))) { switch (msg->msg) { - case CURLMSG_DONE: + case CURLMSG_DONE: { al_assert(msg->easy_handle == curl->handle); + CURLcode code = msg->data.result; if (code != CURLE_OK) { al_log_error("http", "Curl error: %s.", curl_easy_strerror(code)); - http->callback(http->userdata, AKI_HTTP_ERROR, NULL, -1); + http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, -1); return; } - if (!aki_curl_remove_handle(curl)) return; + + if (!nn_curl_remove_handle(curl)) + return; + check_status_codes(curl); - if (http->response_code == 200) { - http->callback(http->userdata, AKI_HTTP_FINISHED, NULL, http->response_code); - } else if (http->response_code / 100 == 3) { + + 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); + char *redirect_url = NULL; curl_easy_getinfo(curl->handle, CURLINFO_REDIRECT_URL, &redirect_url); if (!redirect_url) { - http->callback(http->userdata, AKI_HTTP_ERROR, NULL, http->response_code); + http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code); return; } - aki_curl_set_url(curl, al_str_cr(redirect_url)); - http->response_code = -1; + nn_curl_set_url(curl, al_str_cr(redirect_url)); + + http->status_code = -1; http->content_length = -1; - if (!aki_curl_add_handle(curl)) return; - http->callback(http->userdata, AKI_HTTP_REDIRECT, NULL, http->response_code); + if (!nn_curl_add_handle(curl)) + return; } else { - http->callback(http->userdata, AKI_HTTP_ERROR, NULL, http->response_code); + http->callback(http->userdata, NNWT_HTTP_ERROR, NULL, http->status_code); } + return; + } case CURLMSG_LAST: al_log_debug("http", "Unhandled CURLMSG_LAST."); break; @@ -65,34 +75,34 @@ static void http_handle_events(void *userdata, struct aki_curl *curl) } } -bool aki_http_init(struct aki_http *http) +bool nn_http_init(struct nn_http *http) { - http->headers = NULL; - http->response_code = -1; + http->status_code = -1; http->content_length = -1; - return aki_curl_init(&http->curl); + http->headers = NULL; + return nn_curl_init(&http->curl); } -void aki_http_set_url(struct aki_http *http, str *url) +void nn_http_set_url(struct nn_http *http, str *url) { - aki_curl_set_url(&http->curl, url); + nn_curl_set_url(&http->curl, url); } -void aki_http_set_user_agent(struct aki_http *http, str *user_agent) +void nn_http_set_user_agent(struct nn_http *http, str *user_agent) { char *c_str = al_str_to_c_str(user_agent); curl_easy_setopt(http->curl.handle, CURLOPT_USERAGENT, c_str); al_free(c_str); } -void aki_http_add_header(struct aki_http *http, str *header) +void nn_http_add_header(struct nn_http *http, str *header) { char *c_str = al_str_to_c_str(header); http->headers = curl_slist_append(http->headers, c_str); al_free(c_str); } -void aki_http_set_range(struct aki_http *http, size_t start, size_t end) +void nn_http_set_range(struct nn_http *http, size_t start, size_t end) { char range[44]; if (start == 0u) al_snprintf(range, sizeof(range), "-%zu", end); @@ -101,128 +111,132 @@ void aki_http_set_range(struct aki_http *http, size_t start, size_t end) curl_easy_setopt(http->curl.handle, CURLOPT_RANGE, range); } -void aki_http_close(struct aki_http *http) +void nn_http_close(struct nn_http *http) { - aki_curl_close(&http->curl); + nn_curl_close(&http->curl); } static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, void *userdata) { - struct aki_http *http = (struct aki_http *)userdata; + struct nn_http *http = (struct nn_http *)userdata; check_status_codes(&http->curl); - return http->callback(http->userdata, AKI_HTTP_WRITE, (u8 *)buffer, size * nmemb); + return http->callback(http->userdata, NNWT_HTTP_WRITE, (u8 *)buffer, size * nmemb); } static size_t stream_read_callback(char *buffer, size_t size, size_t nmemb, void *userdata) { - struct aki_http *http = (struct aki_http *)userdata; - return http->callback(http->userdata, AKI_HTTP_READ, (u8 *)buffer, size * nmemb); + struct nn_http *http = (struct nn_http *)userdata; + return http->callback(http->userdata, NNWT_HTTP_READ, (u8 *)buffer, size * nmemb); } static size_t request_callback(void *userdata, u8 op, u8 *buf, s64 int0) { - struct aki_http_request *request = (struct aki_http_request *)userdata; + struct nn_http_request *request = (struct nn_http_request *)userdata; switch (op) { - case AKI_HTTP_FINISHED: { - request->callback(request->userdata, request, true); + case NNWT_HTTP_READ: { + al_assert(request->pointer >= 0); + + size_t size = nn_buffer_get_size(&request->payload); + if (request->pointer + int0 > (off_t)size) + int0 = size - request->pointer; + + nn_buffer_read(&request->payload, buf, request->pointer, int0); + request->pointer += int0; + break; } - case AKI_HTTP_ERROR: { - request->callback(request->userdata, request, false); + case NNWT_HTTP_WRITE: + nn_buffer_append(&request->response, buf, int0); break; - } - case AKI_HTTP_RESPONSE_CODE: { + case NNWT_HTTP_RESPONSE_CODE: break; - } - case AKI_HTTP_CONTENT_LENGTH: { - aki_buffer_ensure_space(&request->response, int0); + case NNWT_HTTP_CONTENT_LENGTH: + nn_buffer_ensure_space(&request->response, int0); break; - } - case AKI_HTTP_WRITE: { - aki_buffer_append(&request->response, buf, int0); + case NNWT_HTTP_REDIRECT: break; - } - case AKI_HTTP_READ: { - al_assert(request->pointer >= 0); - size_t size = aki_buffer_get_size(&request->payload); - if (request->pointer + int0 > (off_t)size) { - int0 = size - request->pointer; - } - aki_buffer_read(&request->payload, buf, request->pointer, int0); - request->pointer += int0; + case NNWT_HTTP_FINISHED: + request->callback(request->userdata, request, true); + break; + case NNWT_HTTP_ERROR: + request->callback(request->userdata, request, false); break; - } } return int0; } -static bool init_request_internal(struct aki_http *http, u8 method) +static bool init_request_internal(struct nn_http *http, u8 method) { - struct aki_curl *curl = &http->curl; + struct nn_curl *curl = &http->curl; + switch (method) { - case AKI_HTTP_HEAD: + case NNWT_HTTP_HEAD: curl_easy_setopt(curl->handle, CURLOPT_NOBODY, 1L); break; - case AKI_HTTP_GET: + case NNWT_HTTP_GET: curl_easy_setopt(curl->handle, CURLOPT_HTTPGET, 1L); break; - case AKI_HTTP_POST: + case NNWT_HTTP_POST: curl_easy_setopt(curl->handle, CURLOPT_POST, 1L); break; } + curl_easy_setopt(curl->handle, CURLOPT_WRITEFUNCTION, stream_write_callback); curl_easy_setopt(curl->handle, CURLOPT_WRITEDATA, http); curl_easy_setopt(curl->handle, CURLOPT_READFUNCTION, stream_read_callback); curl_easy_setopt(curl->handle, CURLOPT_READDATA, http); - if (http->headers) { + + if (http->headers) curl_easy_setopt(curl->handle, CURLOPT_HTTPHEADER, http->headers); - } + curl->handle_events = http_handle_events; - return aki_curl_add_handle(curl); + + return nn_curl_add_handle(curl); } -bool aki_http_request_stream(struct aki_http *http, u8 method, struct aki_event_loop *loop, +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) { - struct aki_curl *curl = &http->curl; + struct nn_curl *curl = &http->curl; al_assert(curl->loop == NULL); - al_assert(method != AKI_HTTP_POST); // Not implemented. + al_assert(method != NNWT_HTTP_POST); // Not implemented. curl->loop = loop; http->callback = callback; http->userdata = userdata; return init_request_internal(http, method); } -bool aki_http_request_init(struct aki_http_request *request) +bool nn_http_request_init(struct nn_http_request *request) { request->pointer = 0; - aki_buffer_init(&request->payload); - aki_buffer_init(&request->response); - return aki_http_init(&request->http); + nn_buffer_init(&request->payload); + nn_buffer_init(&request->response); + return nn_http_init(&request->http); } -bool aki_http_request(struct aki_http_request *request, u8 method, struct aki_event_loop *loop, - void (*callback)(void *, struct aki_http_request *, bool), void *userdata) +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) { - struct aki_curl *curl = &request->http.curl; + struct nn_curl *curl = &request->http.curl; al_assert(curl->loop == NULL); curl->loop = loop; + request->http.callback = request_callback; request->http.userdata = request; - size_t payload = aki_buffer_get_size(&request->payload); - if (payload > 0) { + + size_t payload = nn_buffer_get_size(&request->payload); + if (payload > 0) curl_easy_setopt(curl->handle, CURLOPT_POSTFIELDSIZE, (s64)payload); - } else { - al_assert(method != AKI_HTTP_POST); - } + request->callback = callback; request->userdata = userdata; + return init_request_internal(&request->http, method); } -void aki_http_request_close(struct aki_http_request *request) +void nn_http_request_close(struct nn_http_request *request) { - aki_buffer_free(&request->payload); - aki_buffer_free(&request->response); - aki_http_close(&request->http); + nn_buffer_free(&request->payload); + nn_buffer_free(&request->response); + nn_http_close(&request->http); } diff --git a/src/curl/http.h b/src/curl/http.h index 3d0b796..d5efcf9 100644 --- a/src/curl/http.h +++ b/src/curl/http.h @@ -9,48 +9,48 @@ #include "curl.h" enum { - AKI_HTTP_HEAD = 0, - AKI_HTTP_GET, - AKI_HTTP_POST + NNWT_HTTP_HEAD = 0, + NNWT_HTTP_GET, + NNWT_HTTP_POST }; enum { - AKI_HTTP_WRITE = 0, - AKI_HTTP_READ, - AKI_HTTP_RESPONSE_CODE, - AKI_HTTP_CONTENT_LENGTH, - AKI_HTTP_FINISHED, - AKI_HTTP_REDIRECT, - AKI_HTTP_ERROR + NNWT_HTTP_READ = 0, + NNWT_HTTP_WRITE, + NNWT_HTTP_RESPONSE_CODE, + NNWT_HTTP_CONTENT_LENGTH, + NNWT_HTTP_REDIRECT, + NNWT_HTTP_FINISHED, + NNWT_HTTP_ERROR }; -struct aki_http { - struct aki_curl curl; - s64 response_code; +struct nn_http { + struct nn_curl curl; + s64 status_code; curl_off_t content_length; struct curl_slist *headers; size_t (*callback)(void *, u8, u8 *, s64); void *userdata; }; -struct aki_http_request { - struct aki_http http; +struct nn_http_request { + struct nn_http http; off_t pointer; - struct aki_buffer payload; - struct aki_buffer response; - void (*callback)(void *, struct aki_http_request *, bool); + struct nn_buffer payload; + struct nn_buffer response; + void (*callback)(void *, struct nn_http_request *, bool); void *userdata; }; -bool aki_http_init(struct aki_http *http); -void aki_http_set_url(struct aki_http *http, str *url); -void aki_http_set_user_agent(struct aki_http *http, str *user_agent); -void aki_http_add_header(struct aki_http *http, str *header); -void aki_http_set_range(struct aki_http *http, size_t start, size_t end); -void aki_http_close(struct aki_http *http); -bool aki_http_request_stream(struct aki_http *http, u8 method, struct aki_event_loop *loop, +bool nn_http_init(struct nn_http *http); +void nn_http_set_url(struct nn_http *http, str *url); +void nn_http_set_user_agent(struct nn_http *http, str *user_agent); +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); -bool aki_http_request_init(struct aki_http_request *request); -bool aki_http_request(struct aki_http_request *request, u8 method, struct aki_event_loop *loop, - void (*callback)(void *, struct aki_http_request *, bool), void *userdata); -void aki_http_request_close(struct aki_http_request *request); +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); +void nn_http_request_close(struct nn_http_request *request); diff --git a/src/curl/websocket.c b/src/curl/websocket.c index 8016981..25c9996 100644 --- a/src/curl/websocket.c +++ b/src/curl/websocket.c @@ -1,19 +1,20 @@ #include <al/log.h> -#include <aki/socket.h> + +#include "../socket/socket.h" #include "websocket.h" -static void websocket_handle_events(void *userdata, struct aki_curl *curl) +static void websocket_handle_events(void *userdata, struct nn_curl *curl) { - struct aki_websocket *ws = (struct aki_websocket *)userdata; + struct nn_websocket *ws = (struct nn_websocket *)userdata; s32 pending; CURLMsg *msg; while ((msg = curl_multi_info_read(curl->multi_handle, &pending))) { switch (msg->msg) { case CURLMSG_DONE: - aki_curl_remove_handle(curl); - aki_curl_close(curl); - aki_buffer_free(&ws->frame); + nn_curl_remove_handle(curl); + nn_curl_close(curl); + nn_buffer_free(&ws->frame); ws->callback(ws->userdata, NULL, NULL, 0); return; case CURLMSG_LAST: @@ -28,52 +29,61 @@ static size_t stream_write_callback(char *buffer, size_t size, size_t nmemb, voi { (void)size; (void)nmemb; - struct aki_websocket *ws = (struct aki_websocket *)userdata; + struct nn_websocket *ws = (struct nn_websocket *)userdata; + const struct curl_ws_frame *m = curl_ws_meta(ws->curl.handle); size_t frame_size = m->offset + m->len + m->bytesleft; al_log_debug("ws", "frame_size: %zu, bytesleft: %zu.", frame_size, m->bytesleft); - aki_buffer_ensure_space(&ws->frame, frame_size); - al_memcpy(aki_buffer_get_ptr(&ws->frame, m->offset), buffer, m->len); - if (m->bytesleft == 0) { - ws->callback(ws->userdata, m, aki_buffer_get_ptr(&ws->frame, 0), frame_size); - } + + nn_buffer_ensure_space(&ws->frame, frame_size); + al_memcpy(nn_buffer_get_ptr(&ws->frame, m->offset), buffer, m->len); + + if (m->bytesleft == 0) + ws->callback(ws->userdata, m, nn_buffer_get_ptr(&ws->frame, 0), frame_size); + return m->len; } -bool aki_websocket_init(struct aki_websocket *ws) +bool nn_websocket_init(struct nn_websocket *ws) { - aki_buffer_init(&ws->frame); - return aki_curl_init(&ws->curl); + nn_buffer_init(&ws->frame); + return nn_curl_init(&ws->curl); } -void aki_websocket_set_url(struct aki_websocket *ws, str *url) +void nn_websocket_set_url(struct nn_websocket *ws, str *url) { - aki_curl_set_url(&ws->curl, url); + nn_curl_set_url(&ws->curl, url); } -bool aki_websocket_connect(struct aki_websocket *ws, struct aki_event_loop *loop, +bool nn_websocket_connect(struct nn_websocket *ws, struct nn_event_loop *loop, void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t), void *userdata) { - curl_easy_setopt(ws->curl.handle, CURLOPT_CONNECT_ONLY, 0L); - ws->curl.loop = loop; + struct nn_curl *curl = &ws->curl; + al_assert(curl->loop == NULL); + curl->loop = loop; + + curl_easy_setopt(curl->handle, CURLOPT_CONNECT_ONLY, 0L); ws->callback = callback; ws->userdata = userdata; - curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEFUNCTION, stream_write_callback); - curl_easy_setopt(ws->curl.handle, CURLOPT_WRITEDATA, ws); - ws->curl.handle_events = websocket_handle_events; - ws->curl.userdata = ws; - return aki_curl_add_handle(&ws->curl); + + curl_easy_setopt(curl->handle, CURLOPT_WRITEFUNCTION, stream_write_callback); + curl_easy_setopt(curl->handle, CURLOPT_WRITEDATA, ws); + + curl->handle_events = websocket_handle_events; + curl->userdata = ws; + + return nn_curl_add_handle(&ws->curl); } -void aki_websocket_disconnect(struct aki_websocket *ws, u16 status) +void nn_websocket_disconnect(struct nn_websocket *ws, u16 status) { size_t sent; - status = aki_htons(status); + status = nn_htons(status); curl_ws_send(ws->curl.handle, &status, sizeof(u16), &sent, 0, CURLWS_CLOSE); al_assert(sent == sizeof(u16)); } -bool aki_websocket_send_frame(struct aki_websocket *ws, u8 *data, size_t size) +bool nn_websocket_send_frame(struct nn_websocket *ws, u8 *data, size_t size) { size_t sent; CURLcode code = curl_ws_send(ws->curl.handle, data, size, &sent, 0, CURLWS_TEXT); diff --git a/src/curl/websocket.h b/src/curl/websocket.h index 276f6bc..5eee73f 100644 --- a/src/curl/websocket.h +++ b/src/curl/websocket.h @@ -4,16 +4,16 @@ #include "curl.h" -struct aki_websocket { - struct aki_curl curl; - struct aki_buffer frame; +struct nn_websocket { + struct nn_curl curl; + struct nn_buffer frame; void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t); void *userdata; }; -bool aki_websocket_init(struct aki_websocket *ws); -void aki_websocket_set_url(struct aki_websocket *ws, str *url); -bool aki_websocket_connect(struct aki_websocket *ws, struct aki_event_loop *loop, +bool nn_websocket_init(struct nn_websocket *ws); +void nn_websocket_set_url(struct nn_websocket *ws, str *url); +bool nn_websocket_connect(struct nn_websocket *ws, struct nn_event_loop *loop, void (*callback)(void *, const struct curl_ws_frame *, u8 *, size_t), void *userdata); -void aki_websocket_disconnect(struct aki_websocket *ws, u16 status); -bool aki_websocket_send_frame(struct aki_websocket *ws, u8 *data, size_t size); +void nn_websocket_disconnect(struct nn_websocket *ws, u16 status); +bool nn_websocket_send_frame(struct nn_websocket *ws, u8 *data, size_t size); diff --git a/src/fs_event/fs_event.h b/src/fs_event/fs_event.h index d4983d4..1e99b62 100644 --- a/src/fs_event/fs_event.h +++ b/src/fs_event/fs_event.h @@ -4,27 +4,27 @@ #ifndef _WIN32 #include <sys/inotify.h> -#define AKI_CLOSE_WRITE IN_CLOSE_WRITE +#define NNWT_CLOSE_WRITE IN_CLOSE_WRITE #else struct inotify_event {}; -#define AKI_CLOSE_WRITE 0 +#define NNWT_CLOSE_WRITE 0 #endif #include "../loop.h" -struct aki_fs_event { +struct nn_fs_event { s32 fd; u32 mask; char *path; s32 wd; ev_io event; - struct aki_event_loop *loop; + struct nn_event_loop *loop; void (*callback)(void *, struct inotify_event *); void *userdata; }; -void aki_fs_event_init(struct aki_fs_event *fs, str *path, u32 mask, +void nn_fs_event_init(struct nn_fs_event *fs, str *path, u32 mask, void (*callback)(void *, struct inotify_event *), void *userdata); -bool aki_fs_event_start(struct aki_fs_event *fs, struct aki_event_loop *loop); -void aki_fs_event_stop(struct aki_fs_event *fs); -void aki_fs_event_free(struct aki_fs_event *fs); +bool nn_fs_event_start(struct nn_fs_event *fs, struct nn_event_loop *loop); +void nn_fs_event_stop(struct nn_fs_event *fs); +void nn_fs_event_free(struct nn_fs_event *fs); diff --git a/src/fs_event/fs_event_inotify.c b/src/fs_event/fs_event_inotify.c index 0c0dd01..c0862c7 100644 --- a/src/fs_event/fs_event_inotify.c +++ b/src/fs_event/fs_event_inotify.c @@ -8,22 +8,22 @@ static void event_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_fs_event *fs = (struct aki_fs_event *)w->data; + struct nn_fs_event *fs = (struct nn_fs_event *)w->data; (void)revents; + struct inotify_event event = { 0 }; - ssize_t ret = aki_read(fs->fd, &event, sizeof(struct inotify_event)); + ssize_t ret = nn_read(fs->fd, &event, sizeof(struct inotify_event)); if (ret == sizeof(struct inotify_event)) { - al_assert(event.wd == fs->wd); - al_assert(event.len == 0); + al_assert(event.wd == fs->wd && event.len == 0); fs->callback(fs->userdata, &event); } } -void aki_fs_event_init(struct aki_fs_event *fs, str *path, u32 mask, +void nn_fs_event_init(struct nn_fs_event *fs, str *path, u32 mask, void (*callback)(void *, struct inotify_event *), void *userdata) { fs->fd = inotify_init(); - aki_fd_set_blocking(fs->fd, true); + nn_fd_set_blocking(fs->fd, true); fs->path = al_str_to_c_str(path); fs->mask = mask; fs->callback = callback; @@ -32,25 +32,25 @@ void aki_fs_event_init(struct aki_fs_event *fs, str *path, u32 mask, ev_io_init(&fs->event, event_callback, fs->fd, EV_READ); } -bool aki_fs_event_start(struct aki_fs_event *fs, struct aki_event_loop *loop) +bool nn_fs_event_start(struct nn_fs_event *fs, struct nn_event_loop *loop) { fs->loop = loop; fs->wd = inotify_add_watch(fs->fd, fs->path, fs->mask); if (fs->wd == -1) { - al_log_error("fs_event", "inotify_add_watch(%s) failed (%s)", fs->path, aki_strerror(errno)); + al_log_error("fs_event", "inotify_add_watch(%s) failed (%s)", fs->path, nn_strerror(errno)); return false; } ev_io_start(fs->loop->ev, &fs->event); return true; } -void aki_fs_event_stop(struct aki_fs_event *fs) +void nn_fs_event_stop(struct nn_fs_event *fs) { inotify_rm_watch(fs->fd, fs->wd); ev_io_stop(fs->loop->ev, &fs->event); } -void aki_fs_event_free(struct aki_fs_event *fs) +void nn_fs_event_free(struct nn_fs_event *fs) { close(fs->fd); al_free(fs->path); diff --git a/src/http_server.h b/src/http_server.h index 3f59c93..6f70f09 100644 --- a/src/http_server.h +++ b/src/http_server.h @@ -1,2 +1 @@ #pragma once - diff --git a/src/line_processor.c b/src/line_processor.c index b7d4ef0..f76197b 100644 --- a/src/line_processor.c +++ b/src/line_processor.c @@ -2,151 +2,161 @@ #include "line_processor.h" -// Read from a socket/fd and callback for each line. -// This isn't meant to be used seriously. In it's current state it's just to make testing easier. +// Read from a socket/fd and callback for each "line" determined by a delimiter string. +// This isn't meant to be used seriously and is currently just for testing. -#define CHUNK_SIZE 512 -#define END al_str_c("X") +#define LINE_PROCESSOR_CHUNK_SIZE 128 +#define LINE_PROCESSOR_END al_str_c("X") -static void remove_client(struct aki_line_processor *pro, struct aki_line_processor_client *client) +static void remove_client(struct nn_line_processor *pro, struct nn_line_processor_client *client) { ev_io_stop(pro->loop->ev, &client->event); - if (client->pro->type == AKI_LINE_PROCESSOR_SOCKET) { - aki_socket_close(&client->sock); - } - aki_buffer_free(&client->buf); - struct aki_line_processor_client *rclient; - al_array_foreach(pro->clients, i, rclient) { - if (rclient == client) { - al_array_remove_at_iter(pro->clients, i); - } - } + if (pro->type == NNWT_LINE_PROCESSOR_SOCKET) + nn_socket_close(&client->sock); + nn_buffer_free(&client->buf); + al_array_remove(pro->clients, client); al_free(client); } static void read_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_line_processor_client *client = (struct aki_line_processor_client *)w->data; + struct nn_line_processor_client *cl = (struct nn_line_processor_client *)w->data; (void)revents; + struct nn_line_processor *pro = cl->pro; + u8 type = pro->type; + + nn_buffer_ensure_space(&cl->buf, cl->index + LINE_PROCESSOR_CHUNK_SIZE); + u8 *ptr = nn_buffer_get_ptr(&cl->buf, cl->index); - aki_buffer_ensure_space(&client->buf, client->index + CHUNK_SIZE); - u8 *ptr = aki_buffer_get_ptr(&client->buf, client->index); ssize_t ret; - if (client->pro->type == AKI_LINE_PROCESSOR_SOCKET) { - ret = aki_socket_read(&client->sock, ptr, CHUNK_SIZE); - } else { - ret = aki_read(client->pro->fd, ptr, CHUNK_SIZE); + switch (type) { + case NNWT_LINE_PROCESSOR_SOCKET: + ret = nn_socket_read(&cl->sock, ptr, LINE_PROCESSOR_CHUNK_SIZE); + break; + case NNWT_LINE_PROCESSOR_FD: + ret = nn_read(pro->fd, ptr, LINE_PROCESSOR_CHUNK_SIZE); + break; + default: + al_assert_and_return(); } - if (ret <= 0) return; - client->index += ret; + if (ret <= 0) + return; + + cl->index += ret; + ptr = nn_buffer_get_ptr(&cl->buf, 0); + str *s = al_str_w((char *)ptr, 0, cl->index); + + str *dl = &pro->delim; + u32 offset = dl->len - 1; + for (size_t i = cl->mark; i < cl->index - offset; i++) { + if (al_str_cmp(s, dl, i, dl->len) == 0) { + s->len = i; - str *delim = &client->pro->delim; - u32 offset = delim->len - 1; - ptr = aki_buffer_get_ptr(&client->buf, 0); - for (size_t i = client->mark; i < client->index - offset; i++) { - if (al_str_cmp(al_str_w((char *)ptr, i, (u32)(client->index - i)), delim, 0, delim->len) == 0) { - str *s = al_str_w((char *)ptr, 0, i); - if (al_str_eq(s, END)) { - remove_client(client->pro, client); + if (al_str_eq(s, LINE_PROCESSOR_END)) { + remove_client(pro, cl); return; } - switch (client->pro->callback(client->pro->userdata, s)) { - case AKI_LINE_PROCESSOR_CONTINUE: - break; - case AKI_LINE_PROCESSOR_STOP: - aki_line_processor_stop(client->pro); + + switch (pro->callback(pro->userdata, s)) { + case NNWT_LINE_PROCESSOR_STOP: + nn_line_processor_stop(pro); return; + case NNWT_LINE_PROCESSOR_CONTINUE: + break; } + size_t skip = i + offset + 1; - client->index = client->index - skip; - al_memmove(ptr, &ptr[skip], client->index); - client->mark = i = 0; + cl->index = cl->index - skip; + al_memmove(ptr, &ptr[skip], cl->index); + s = al_str_w((char *)ptr, 0, cl->index); + + cl->mark = i = 0; } } - client->mark = client->index; + cl->mark = cl->index; } static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_line_processor *pro = (struct aki_line_processor *)w->data; + struct nn_line_processor *pro = (struct nn_line_processor *)w->data; (void)revents; - struct aki_line_processor_client *client = al_alloc_object(struct aki_line_processor_client); - if (!aki_socket_accept(pro->sock, &client->sock)) { - al_free(client); + + struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client); + if (!nn_socket_accept(pro->sock, &cl->sock, NNWT_SOCKET_NONBLOCKING)) { + al_free(cl); return; } - aki_socket_set_blocking(&client->sock, false); - client->index = 0; - aki_buffer_init(&client->buf); - client->mark = 0; - client->pro = pro; - client->event.data = client; - al_array_push(pro->clients, client); - ev_io_init(&client->event, read_callback, aki_socket_get_fd(&client->sock), EV_READ); - ev_io_start(pro->loop->ev, &client->event); + cl->index = 0; + nn_buffer_init(&cl->buf); + cl->mark = 0; + cl->pro = pro; + cl->event.data = cl; + al_array_push(pro->clients, cl); + + ev_io_init(&cl->event, read_callback, nn_socket_get_fd(&cl->sock), EV_READ); + ev_io_start(pro->loop->ev, &cl->event); } -void aki_line_processor_init(struct aki_line_processor *pro, str *delim) +void nn_line_processor_init(struct nn_line_processor *pro, str *delim) { pro->loop = NULL; al_str_clone(&pro->delim, delim); al_array_init(pro->clients); } -void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *sock) +void nn_line_processor_open_socket(struct nn_line_processor *pro, struct nn_socket *sock) { - pro->type = AKI_LINE_PROCESSOR_SOCKET; + pro->type = NNWT_LINE_PROCESSOR_SOCKET; pro->sock = sock; } -void aki_line_processor_open_fd(struct aki_line_processor *pro, s32 fd) +void nn_line_processor_open_fd(struct nn_line_processor *pro, s32 fd) { - pro->type = AKI_LINE_PROCESSOR_FD; + pro->type = NNWT_LINE_PROCESSOR_FD; pro->fd = fd; } -void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loop *loop) +void nn_line_processor_run(struct nn_line_processor *pro, struct nn_event_loop *loop) { pro->loop = loop; switch (pro->type) { - case AKI_LINE_PROCESSOR_SOCKET: + case NNWT_LINE_PROCESSOR_SOCKET: pro->event.data = pro; - ev_io_init(&pro->event, socket_connection_callback, aki_socket_get_fd(pro->sock), EV_READ); + ev_io_init(&pro->event, socket_connection_callback, nn_socket_get_fd(pro->sock), EV_READ); ev_io_start(pro->loop->ev, &pro->event); break; - case AKI_LINE_PROCESSOR_FD: { // Not supported on windows. - struct aki_line_processor_client *client = al_alloc_object(struct aki_line_processor_client); - client->index = 0; - aki_buffer_init(&client->buf); - client->mark = 0; - client->pro = pro; - client->event.data = client; - al_array_push(pro->clients, client); - ev_io_init(&client->event, read_callback, pro->fd, EV_READ); - ev_io_start(pro->loop->ev, &client->event); + case NNWT_LINE_PROCESSOR_FD: { // Not supported on windows. + struct nn_line_processor_client *cl = al_alloc_object(struct nn_line_processor_client); + cl->index = 0; + nn_buffer_init(&cl->buf); + cl->mark = 0; + cl->pro = pro; + cl->event.data = cl; + al_array_push(pro->clients, cl); + ev_io_init(&cl->event, read_callback, pro->fd, EV_READ); + ev_io_start(pro->loop->ev, &cl->event); break; } } } -void aki_line_processor_stop(struct aki_line_processor *pro) +void nn_line_processor_stop(struct nn_line_processor *pro) { if (pro->loop) { switch (pro->type) { - case AKI_LINE_PROCESSOR_SOCKET: + case NNWT_LINE_PROCESSOR_SOCKET: ev_io_stop(pro->loop->ev, &pro->event); break; - case AKI_LINE_PROCESSOR_FD: + case NNWT_LINE_PROCESSOR_FD: break; } - struct aki_line_processor_client *client; - al_array_foreach(pro->clients, i, client) { - remove_client(client->pro, client); - } + struct nn_line_processor_client *cl; + al_array_foreach(pro->clients, i, cl) + remove_client(cl->pro, cl); } al_array_free(pro->clients); al_str_free(&pro->delim); diff --git a/src/line_processor.h b/src/line_processor.h index 20f1101..5da7385 100644 --- a/src/line_processor.h +++ b/src/line_processor.h @@ -5,46 +5,43 @@ #include <al/array.h> #include "util/buffer.h" -#include "util/thread/thread.h" -#include "util/file/file.h" #include "socket/socket.h" #include "loop.h" -#include "signal.h" enum { - AKI_LINE_PROCESSOR_SOCKET = 0, - AKI_LINE_PROCESSOR_FD + NNWT_LINE_PROCESSOR_SOCKET = 0, + NNWT_LINE_PROCESSOR_FD }; enum { - AKI_LINE_PROCESSOR_CONTINUE = 0, - AKI_LINE_PROCESSOR_STOP + NNWT_LINE_PROCESSOR_CONTINUE = 0, + NNWT_LINE_PROCESSOR_STOP }; -struct aki_line_processor_client { +struct nn_line_processor_client { ev_io event; - struct aki_socket sock; + struct nn_socket sock; size_t index; - struct aki_buffer buf; + struct nn_buffer buf; size_t mark; - struct aki_line_processor *pro; + struct nn_line_processor *pro; }; -struct aki_line_processor { - struct aki_event_loop *loop; +struct nn_line_processor { + struct nn_event_loop *loop; u8 type; ev_io event; - struct aki_socket *sock; + struct nn_socket *sock; s32 fd; str delim; - array(struct aki_line_processor_client *) clients; + array(struct nn_line_processor_client *) clients; u8 (*callback)(void *, str *line); void *userdata; }; -void aki_line_processor_init(struct aki_line_processor *pro, str *delim); -void aki_line_processor_open_socket(struct aki_line_processor *pro, struct aki_socket *sock); -void aki_line_processor_open_fd(struct aki_line_processor *pro, s32 fd); -void aki_line_processor_run(struct aki_line_processor *pro, struct aki_event_loop *loop); -void aki_line_processor_stop(struct aki_line_processor *pro); +void nn_line_processor_init(struct nn_line_processor *pro, str *delim); +void nn_line_processor_open_socket(struct nn_line_processor *pro, struct nn_socket *sock); +void nn_line_processor_open_fd(struct nn_line_processor *pro, s32 fd); +void nn_line_processor_run(struct nn_line_processor *pro, struct nn_event_loop *loop); +void nn_line_processor_stop(struct nn_line_processor *pro); @@ -2,7 +2,7 @@ #include "loop.h" -bool aki_event_loop_init(struct aki_event_loop *loop) +bool nn_event_loop_init(struct nn_event_loop *loop) { loop->ev = ev_default_loop(EVFLAG_AUTO); //loop->ev = ev_loop_new(EVFLAG_AUTO); @@ -11,40 +11,40 @@ bool aki_event_loop_init(struct aki_event_loop *loop) return true; } -s32 aki_event_loop_run(struct aki_event_loop *loop) +s32 nn_event_loop_run(struct nn_event_loop *loop) { return ev_run(loop->ev, 0); } -s32 aki_event_loop_run_once(struct aki_event_loop *loop) +s32 nn_event_loop_run_once(struct nn_event_loop *loop) { return ev_run(loop->ev, EVRUN_ONCE | EVRUN_NOWAIT); } -void aki_event_loop_sleep(struct aki_event_loop *loop, aki_os_tstamp delay) +void nn_event_loop_sleep(struct nn_event_loop *loop, nn_os_tstamp delay) { - al_assert(delay >= AKI_TS_FROM_USEC(1000)); - u64 prev = aki_get_timestamp(), now; + al_assert(delay >= NNWT_TS_FROM_USEC(1000)); + u64 prev = nn_get_timestamp(), now; while (delay >= 0.0) { - aki_event_loop_run_once(loop); - aki_thread_sleep(AKI_TS_FROM_USEC(1000)); - now = aki_get_timestamp(); - delay -= AKI_TS_FROM_USEC(now - prev); + nn_event_loop_run_once(loop); + nn_thread_sleep(NNWT_TS_FROM_USEC(1000)); + now = nn_get_timestamp(); + delay -= NNWT_TS_FROM_USEC(now - prev); prev = now; } } -void aki_event_loop_break_one(struct aki_event_loop *loop) +void nn_event_loop_break_one(struct nn_event_loop *loop) { ev_break(loop->ev, EVBREAK_ONE); } -void aki_event_loop_break(struct aki_event_loop *loop) +void nn_event_loop_break(struct nn_event_loop *loop) { ev_break(loop->ev, EVBREAK_ALL); } -void aki_event_loop_destroy(struct aki_event_loop *loop) +void nn_event_loop_destroy(struct nn_event_loop *loop) { ev_loop_destroy(loop->ev); loop->ev = NULL; @@ -6,15 +6,15 @@ #include "evwrap.h" -struct aki_event_loop { +struct nn_event_loop { struct ev_loop *ev; }; -bool aki_event_loop_init(struct aki_event_loop *loop); -s32 aki_event_loop_run(struct aki_event_loop *loop); -s32 aki_event_loop_run_once(struct aki_event_loop *loop); -void aki_event_loop_sleep(struct aki_event_loop *loop, aki_os_tstamp delay); -void aki_event_loop_break_one(struct aki_event_loop *loop); -void aki_event_loop_break(struct aki_event_loop *loop); -bool aki_event_loop_is_broken(struct aki_event_loop *loop); -void aki_event_loop_destroy(struct aki_event_loop *loop); +bool nn_event_loop_init(struct nn_event_loop *loop); +s32 nn_event_loop_run(struct nn_event_loop *loop); +s32 nn_event_loop_run_once(struct nn_event_loop *loop); +void nn_event_loop_sleep(struct nn_event_loop *loop, nn_os_tstamp delay); +void nn_event_loop_break_one(struct nn_event_loop *loop); +void nn_event_loop_break(struct nn_event_loop *loop); +bool nn_event_loop_is_broken(struct nn_event_loop *loop); +void nn_event_loop_destroy(struct nn_event_loop *loop); diff --git a/src/multiplex.c b/src/multiplex.c index 4f5330f..7007cfc 100644 --- a/src/multiplex.c +++ b/src/multiplex.c @@ -1,48 +1,58 @@ #include "multiplex.h" -bool aki_multiplex_socket_init(struct aki_multiplex_socket *multi, u8 type, - bool (*connection_callback)(void *, u8, struct aki_socket *), void *userdata) +bool nn_multiplex_socket_init(struct nn_multiplex_socket *multi, u8 type, + bool (*connection_callback)(void *, u8, struct nn_socket *), void *userdata) { multi->sock.type = type; - if (!aki_socket_init(&multi->sock)) return false; - aki_socket_set_blocking(&multi->sock, false); + if (!nn_socket_init(&multi->sock, NNWT_SOCKET_NONBLOCKING)) + return false; + multi->connection_callback = connection_callback; multi->userdata = userdata; + return true; } static void socket_connection_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_multiplex_socket *multi = (struct aki_multiplex_socket *)w->data; + struct nn_multiplex_socket *multi = (struct nn_multiplex_socket *)w->data; (void)revents; - struct aki_socket sock; - if (!aki_socket_accept(&multi->sock, &sock)) return; + + struct nn_socket sock; + if (!nn_socket_accept(&multi->sock, &sock, 0)) + return; + u8 id; - ssize_t ret = aki_socket_read(&sock, &id, sizeof(u8)); + ssize_t ret = nn_socket_read(&sock, &id, sizeof(u8)); if (ret > 0) { - al_assert(ret == sizeof(u8)); - if (multi->connection_callback(multi->userdata, id, &sock)) { + nn_socket_set_blocking(&sock, false); + if (multi->connection_callback(multi->userdata, id, &sock)) return; - } } - aki_socket_close(&sock); + + // Failure case. + nn_socket_close(&sock); } -bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, u16 port) +bool nn_multiplex_socket_listen(struct nn_multiplex_socket *multi, struct nn_event_loop *loop, str *addr, u16 port) { - if (!aki_socket_bind(&multi->sock, addr, port)) return false; - if (!aki_socket_listen(&multi->sock)) return false; + if (!nn_socket_bind(&multi->sock, addr, port) || !nn_socket_listen(&multi->sock)) + return false; + multi->loop = loop; + multi->event.data = multi; - ev_io_init(&multi->event, socket_connection_callback, aki_socket_get_fd(&multi->sock), EV_READ); + ev_io_init(&multi->event, socket_connection_callback, nn_socket_get_fd(&multi->sock), EV_READ); + ev_io_start(multi->loop->ev, &multi->event); + return true; } -void aki_multiplex_socket_close(struct aki_multiplex_socket *multi) +void nn_multiplex_socket_close(struct nn_multiplex_socket *multi) { - aki_socket_shutdown(&multi->sock); - aki_socket_close(&multi->sock); + nn_socket_shutdown(&multi->sock); + nn_socket_close(&multi->sock); ev_io_stop(multi->loop->ev, &multi->event); } diff --git a/src/multiplex.h b/src/multiplex.h index 68d24a4..6339798 100644 --- a/src/multiplex.h +++ b/src/multiplex.h @@ -6,15 +6,15 @@ #include "loop.h" -struct aki_multiplex_socket { - struct aki_socket sock; - struct aki_event_loop *loop; +struct nn_multiplex_socket { + struct nn_socket sock; + struct nn_event_loop *loop; ev_io event; - bool (*connection_callback)(void *, u8 id, struct aki_socket *); + bool (*connection_callback)(void *, u8 id, struct nn_socket *); void *userdata; }; -bool aki_multiplex_socket_init(struct aki_multiplex_socket *multi, u8 type, - bool (*connection_callback)(void *, u8, struct aki_socket *), void *userdata); -bool aki_multiplex_socket_listen(struct aki_multiplex_socket *multi, struct aki_event_loop *loop, str *addr, u16 port); -void aki_multiplex_socket_close(struct aki_multiplex_socket *multi); +bool nn_multiplex_socket_init(struct nn_multiplex_socket *multi, u8 type, + bool (*connection_callback)(void *, u8, struct nn_socket *), 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); diff --git a/src/packet_cache.c b/src/packet_cache.c index b234259..a4e2354 100644 --- a/src/packet_cache.c +++ b/src/packet_cache.c @@ -1,90 +1,94 @@ #include "packet_cache.h" -void aki_packet_cache_init(struct aki_packet_cache *cache, u32 size) +void nn_packet_cache_init(struct nn_packet_cache *cache, u32 size) { al_array_init(cache->cache); al_array_reserve(cache->cache, size); cache->flush = size > 0 ? size / 2 : 0; - aki_cond_init(&cache->cond); - aki_mutex_init(&cache->mutex); cache->disabled = false; + nn_cond_init(&cache->cond); + nn_mutex_init(&cache->mutex); } -bool aki_packet_cache_send_packet(struct aki_packet_cache *cache, struct aki_packet *packet) +bool nn_packet_cache_send_packet(struct nn_packet_cache *cache, struct nn_packet *packet) { - aki_mutex_lock(&cache->mutex); + nn_mutex_lock(&cache->mutex); + if (cache->disabled) { - aki_mutex_unlock(&cache->mutex); + nn_mutex_unlock(&cache->mutex); return false; } + al_array_push(cache->cache, packet); + bool flush = !packet || cache->cache.size >= cache->flush; - if (flush && aki_cond_is_waiting(&cache->cond)) { - aki_cond_signal(&cache->cond); - } - aki_mutex_unlock(&cache->mutex); + if (flush && nn_cond_is_waiting(&cache->cond)) + nn_cond_signal(&cache->cond); + + nn_mutex_unlock(&cache->mutex); + return true; } -void aki_packet_cache_flush(struct aki_packet_cache *cache) +void nn_packet_cache_flush(struct nn_packet_cache *cache) { - aki_mutex_lock(&cache->mutex); - if (aki_cond_is_waiting(&cache->cond)) { - aki_cond_signal(&cache->cond); - } - aki_mutex_unlock(&cache->mutex); + nn_mutex_lock(&cache->mutex); + if (nn_cond_is_waiting(&cache->cond)) + nn_cond_signal(&cache->cond); + nn_mutex_unlock(&cache->mutex); } -bool aki_packet_cache_wait(struct aki_packet_cache *cache, u32 *count) +bool nn_packet_cache_wait(struct nn_packet_cache *cache, u32 *count) { - aki_mutex_lock(&cache->mutex); - if (!cache->disabled && cache->cache.size == 0) { - aki_cond_wait(&cache->cond, &cache->mutex); - } + nn_mutex_lock(&cache->mutex); + + if (!cache->disabled && cache->cache.size == 0) + nn_cond_wait(&cache->cond, &cache->mutex); + // If there was an active wait before calling disable(), expected // behavior would be that wait() returns false. if (cache->disabled) { - aki_mutex_unlock(&cache->mutex); + nn_mutex_unlock(&cache->mutex); return false; } + *count = cache->cache.size; + return true; } -struct aki_packet *aki_packet_cache_pop(struct aki_packet_cache *cache) +struct nn_packet *nn_packet_cache_pop(struct nn_packet_cache *cache) { - struct aki_packet *packet = NULL; - if (cache->cache.size > 0) { + struct nn_packet *packet = NULL; + if (cache->cache.size > 0) al_array_pop_at(cache->cache, 0, packet); - } return packet; } -void aki_packet_cache_unlock(struct aki_packet_cache *cache) +void nn_packet_cache_unlock(struct nn_packet_cache *cache) { - aki_mutex_unlock(&cache->mutex); + nn_mutex_unlock(&cache->mutex); } -void aki_packet_cache_disable(struct aki_packet_cache *cache) +void nn_packet_cache_disable(struct nn_packet_cache *cache) { - aki_mutex_lock(&cache->mutex); + nn_mutex_lock(&cache->mutex); cache->disabled = true; - if (aki_cond_is_waiting(&cache->cond)) { - aki_cond_signal(&cache->cond); - } - aki_mutex_unlock(&cache->mutex); + if (nn_cond_is_waiting(&cache->cond)) + nn_cond_signal(&cache->cond); + nn_mutex_unlock(&cache->mutex); } -void aki_packet_cache_enable(struct aki_packet_cache *cache) +void nn_packet_cache_enable(struct nn_packet_cache *cache) { - aki_mutex_lock(&cache->mutex); + nn_mutex_lock(&cache->mutex); cache->disabled = false; - aki_mutex_unlock(&cache->mutex); + nn_mutex_unlock(&cache->mutex); } -void aki_packet_cache_free(struct aki_packet_cache *cache) +void nn_packet_cache_free(struct nn_packet_cache *cache) { al_array_free(cache->cache); - aki_mutex_destroy(&cache->mutex); - aki_cond_destroy(&cache->cond); + nn_mutex_destroy(&cache->mutex); + nn_cond_destroy(&cache->cond); } diff --git a/src/packet_cache.h b/src/packet_cache.h index 57133d7..1d1b83a 100644 --- a/src/packet_cache.h +++ b/src/packet_cache.h @@ -5,20 +5,20 @@ #include "util/packet.h" #include "util/thread/thread.h" -struct aki_packet_cache { - array(struct aki_packet *) cache; +struct nn_packet_cache { + array(struct nn_packet *) cache; u32 flush; - struct aki_cond cond; - struct aki_mutex mutex; bool disabled; + struct nn_cond cond; + struct nn_mutex mutex; }; -void aki_packet_cache_init(struct aki_packet_cache *cache, u32 size); -bool aki_packet_cache_send_packet(struct aki_packet_cache *cache, struct aki_packet *packet); -void aki_packet_cache_flush(struct aki_packet_cache *cache); -bool aki_packet_cache_wait(struct aki_packet_cache *cache, u32 *count); -struct aki_packet *aki_packet_cache_pop(struct aki_packet_cache *cache); -void aki_packet_cache_unlock(struct aki_packet_cache *cache); -void aki_packet_cache_disable(struct aki_packet_cache *cache); -void aki_packet_cache_enable(struct aki_packet_cache *cache); -void aki_packet_cache_free(struct aki_packet_cache *cache); +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); +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); +void nn_packet_cache_free(struct nn_packet_cache *cache); diff --git a/src/packet_pool.c b/src/packet_pool.c index 1e3117f..b613253 100644 --- a/src/packet_pool.c +++ b/src/packet_pool.c @@ -1,154 +1,177 @@ #include "packet_pool.h" -static void return_internal(struct aki_packet_pool *pool, struct aki_packet *packet) +static void return_internal(struct nn_packet_pool *pool, struct nn_packet *packet) { - aki_packet_reset(packet); + nn_packet_reset(packet); al_array_push(pool->empty, packet); - if (aki_cond_is_waiting(&pool->cond)) { - aki_cond_signal(&pool->cond); - } + 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) { (void)loop; - struct aki_packet_pool *pool = (struct aki_packet_pool *)w->data; + struct nn_packet_pool *pool = (struct nn_packet_pool *)w->data; (void)revents; - aki_mutex_lock(&pool->mutex); + + nn_mutex_lock(&pool->mutex); + if (pool->disabled) { - aki_mutex_unlock(&pool->mutex); + nn_mutex_unlock(&pool->mutex); return; } + while (pool->ready.size > 0) { - struct aki_packet *packet; + struct nn_packet *packet; al_array_pop_at(pool->ready, 0, packet); u8 ret = pool->callback(pool->userdata, packet); switch (ret) { - case AKI_PACKET_POOL_KEEP: + case NNWT_PACKET_POOL_KEEP: break; - case AKI_PACKET_POOL_RETURN: + case NNWT_PACKET_POOL_RETURN: return_internal(pool, packet); break; } } - aki_mutex_unlock(&pool->mutex); + + nn_mutex_unlock(&pool->mutex); } -void aki_packet_pool_init(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop, - u8 (*callback)(void *, struct aki_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) { pool->loop = loop; + al_assert(size > 0); pool->size = size; + al_array_init(pool->ready); al_array_reserve(pool->ready, size); + al_array_init(pool->empty); al_array_reserve(pool->empty, size); - for (u32 i = 0; i < size; i++) { - al_array_push(pool->empty, aki_packet_create()); - } + for (u32 i = 0; i < size; i++) + al_array_push(pool->empty, nn_packet_create()); + pool->disabled = false; - aki_cond_init(&pool->cond); - aki_mutex_init(&pool->mutex); + nn_cond_init(&pool->cond); + nn_mutex_init(&pool->mutex); + pool->signal.data = pool; ev_async_init(&pool->signal, signal_callback); ev_async_start(pool->loop->ev, &pool->signal); + pool->callback = callback; pool->userdata = userdata; } -static const bool grow = false; +// This defeats the entire purpose of a packet pool +// but is useful for testing. +//#define PACKET_POOL_GROW -struct aki_packet *aki_packet_pool_get(struct aki_packet_pool *pool) +struct nn_packet *nn_packet_pool_get(struct nn_packet_pool *pool) { - aki_mutex_lock(&pool->mutex); + nn_mutex_lock(&pool->mutex); + if (!pool->disabled && !pool->empty.size) { - if (grow) { - al_array_push(pool->empty, aki_packet_create()); - } else { - aki_cond_wait(&pool->cond, &pool->mutex); - } +#ifdef PACKET_POOL_GROW + al_array_push(pool->empty, nn_packet_create()); +#else + nn_cond_wait(&pool->cond, &pool->mutex); +#endif } + if (pool->disabled) { - aki_mutex_unlock(&pool->mutex); + nn_mutex_unlock(&pool->mutex); return NULL; } - struct aki_packet *packet = al_array_pop(pool->empty); - aki_mutex_unlock(&pool->mutex); + + struct nn_packet *packet = al_array_pop(pool->empty); + + nn_mutex_unlock(&pool->mutex); + return packet; } -void aki_packet_pool_submit(struct aki_packet_pool *pool, struct aki_packet *packet) +void nn_packet_pool_submit(struct nn_packet_pool *pool, struct nn_packet *packet) { - aki_mutex_lock(&pool->mutex); + nn_mutex_lock(&pool->mutex); + if (pool->disabled) { - aki_mutex_unlock(&pool->mutex); + nn_mutex_unlock(&pool->mutex); return; } + al_array_push(pool->ready, packet); + bool flush = pool->ready.size >= pool->size / 2; - aki_mutex_unlock(&pool->mutex); - if (flush) { - ev_async_send(pool->loop->ev, &pool->signal); - } + + nn_mutex_unlock(&pool->mutex); + + if (flush) ev_async_send(pool->loop->ev, &pool->signal); } -void aki_packet_pool_flush(struct aki_packet_pool *pool) +void nn_packet_pool_flush(struct nn_packet_pool *pool) { ev_async_send(pool->loop->ev, &pool->signal); } -void aki_packet_pool_return(struct aki_packet_pool *pool, struct aki_packet *packet) +void nn_packet_pool_return(struct nn_packet_pool *pool, struct nn_packet *packet) { - aki_mutex_lock(&pool->mutex); + nn_mutex_lock(&pool->mutex); return_internal(pool, packet); - aki_mutex_unlock(&pool->mutex); + nn_mutex_unlock(&pool->mutex); } -void aki_packet_pool_disable(struct aki_packet_pool *pool) +void nn_packet_pool_disable(struct nn_packet_pool *pool) { - aki_mutex_lock(&pool->mutex); + nn_mutex_lock(&pool->mutex); + pool->disabled = true; - struct aki_packet *packet; + + struct nn_packet *packet; al_array_foreach(pool->ready, i, packet) { - aki_packet_reset(packet); + nn_packet_reset(packet); al_array_push(pool->empty, packet); } pool->ready.size = 0; - if (aki_cond_is_waiting(&pool->cond)) { - aki_cond_signal(&pool->cond); - } - aki_mutex_unlock(&pool->mutex); + + if (nn_cond_is_waiting(&pool->cond)) + nn_cond_signal(&pool->cond); + + nn_mutex_unlock(&pool->mutex); } -void aki_packet_pool_enable(struct aki_packet_pool *pool) +void nn_packet_pool_enable(struct nn_packet_pool *pool) { - aki_mutex_lock(&pool->mutex); + nn_mutex_lock(&pool->mutex); pool->disabled = false; - aki_mutex_unlock(&pool->mutex); + nn_mutex_unlock(&pool->mutex); } -struct aki_packet *aki_packet_pool_pop(struct aki_packet_pool *pool) +struct nn_packet *nn_packet_pool_pop(struct nn_packet_pool *pool) { - struct aki_packet *packet = NULL; - if (pool->ready.size > 0) { + struct nn_packet *packet = NULL; + if (pool->ready.size > 0) al_array_pop_at(pool->ready, 0, packet); - } return packet; } -void aki_packet_pool_free(struct aki_packet_pool *pool) +void nn_packet_pool_free(struct nn_packet_pool *pool) { ev_async_stop(pool->loop->ev, &pool->signal); - aki_mutex_destroy(&pool->mutex); - aki_cond_destroy(&pool->cond); - struct aki_packet *packet; + + nn_mutex_destroy(&pool->mutex); + nn_cond_destroy(&pool->cond); + + struct nn_packet *packet; al_array_foreach(pool->empty, i, packet) { - aki_packet_free(packet); + nn_packet_free(packet); } al_array_free(pool->empty); + al_array_foreach(pool->ready, i, packet) { - aki_packet_free(packet); + nn_packet_free(packet); } al_array_free(pool->ready); } diff --git a/src/packet_pool.h b/src/packet_pool.h index c31ac9e..e20e29a 100644 --- a/src/packet_pool.h +++ b/src/packet_pool.h @@ -17,30 +17,30 @@ // pool_free() enum { - AKI_PACKET_POOL_KEEP = 0, - AKI_PACKET_POOL_RETURN + NNWT_PACKET_POOL_KEEP = 0, + NNWT_PACKET_POOL_RETURN }; -struct aki_packet_pool { - struct aki_event_loop *loop; +struct nn_packet_pool { + struct nn_event_loop *loop; u32 size; - array(struct aki_packet *) empty; - array(struct aki_packet *) ready; - struct aki_cond cond; - struct aki_mutex mutex; - ev_async signal; + array(struct nn_packet *) empty; + array(struct nn_packet *) ready; bool disabled; - u8 (*callback)(void *, struct aki_packet *); + struct nn_cond cond; + struct nn_mutex mutex; + ev_async signal; + u8 (*callback)(void *, struct nn_packet *); void *userdata; }; -void aki_packet_pool_init(struct aki_packet_pool *pool, u32 size, struct aki_event_loop *loop, - u8 (*callback)(void *, struct aki_packet *), void *userdata); -struct aki_packet *aki_packet_pool_get(struct aki_packet_pool *pool); -void aki_packet_pool_submit(struct aki_packet_pool *pool, struct aki_packet *packet); -void aki_packet_pool_flush(struct aki_packet_pool *pool); -void aki_packet_pool_return(struct aki_packet_pool *pool, struct aki_packet *packet); -void aki_packet_pool_disable(struct aki_packet_pool *pool); -void aki_packet_pool_enable(struct aki_packet_pool *pool); -struct aki_packet *aki_packet_pool_pop(struct aki_packet_pool *pool); -void aki_packet_pool_free(struct aki_packet_pool *pool); +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); +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_return(struct nn_packet_pool *pool, struct nn_packet *packet); +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); +void nn_packet_pool_free(struct nn_packet_pool *pool); diff --git a/src/packet_stream.c b/src/packet_stream.c index bd6d476..9a820dd 100644 --- a/src/packet_stream.c +++ b/src/packet_stream.c @@ -7,113 +7,124 @@ enum { PACKET_STREAM_DISCONNECTED }; -static void init_internal(struct aki_packet_stream *stream) +static inline void init_io_state(struct nn_packet_stream *stream) { + stream->corked = true; + stream->in.packet = nn_packet_create(); + stream->in.have_header = false; + stream->in.index = 0; al_array_init(stream->out.queue); stream->out.packet = NULL; stream->out.active = false; - stream->in.packet = aki_packet_create(); - stream->in.have_size = false; - stream->in.index = 0; } -bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type, - void (*connection_callback)(void *, struct aki_packet_stream *), - void (*connection_closed_callback)(void *, struct aki_packet_stream *), void *userdata) +bool nn_packet_stream_init(struct nn_packet_stream *stream, u8 type, + bool (*connection_callback)(void *, struct nn_packet_stream *), + void (*connection_closed_callback)(void *, struct nn_packet_stream *), void *userdata) { stream->sock.type = type; - if (!aki_socket_init(&stream->sock)) { + if (!nn_socket_init(&stream->sock, NNWT_SOCKET_NONBLOCKING)) return false; - } - aki_socket_set_blocking(&stream->sock, false); - stream->corked = false; - stream->connect = PACKET_STREAM_DISCONNECTED; + + init_io_state(stream); + stream->connection_callback = connection_callback; stream->connection_closed_callback = connection_closed_callback; stream->packet_callback = NULL; stream->packet_sent_callback = NULL; stream->userdata = userdata; - init_internal(stream); + + stream->connect = PACKET_STREAM_DISCONNECTED; + return true; } -void aki_packet_stream_set_nodelay(struct aki_packet_stream *stream, s32 nodelay) +void nn_packet_stream_set_nodelay(struct nn_packet_stream *stream, s32 nodelay) { - aki_socket_set_nodelay(&stream->sock, nodelay); + nn_socket_set_nodelay(&stream->sock, nodelay); } -void aki_packet_stream_set_multiplex(struct aki_packet_stream *stream, u8 id) +void nn_packet_stream_set_multiplex(struct nn_packet_stream *stream, u8 id) { stream->id = id; } -static void start_write_internal(struct aki_packet_stream *stream) +static void start_write_internal(struct nn_packet_stream *stream) { stream->out.active = true; ev_io_start(stream->loop->ev, &stream->wevent); } -static void stop_write_internal(struct aki_packet_stream *stream) +static void stop_write_internal(struct nn_packet_stream *stream) { stream->out.active = false; ev_io_stop(stream->loop->ev, &stream->wevent); } -static void shutdown_internal(struct aki_packet_stream *stream) +static void shutdown_internal(struct nn_packet_stream *stream) { stream->connect = PACKET_STREAM_DISCONNECTING; // This assumes select()/poll() will return after shutdown(). - aki_socket_shutdown(&stream->sock); + nn_socket_shutdown(&stream->sock); } -static void stop_internal(struct aki_packet_stream *stream) +static void stop_internal(struct nn_packet_stream *stream) { - stream->corked = false; stream->connect = PACKET_STREAM_DISCONNECTED; - aki_socket_close(&stream->sock); + + nn_socket_close(&stream->sock); + + // Mark the stream corked so reconnect() can be consistent + // with connect() and from_socket(). + stream->corked = true; + + nn_packet_reset(stream->in.packet); + stream->in.have_header = false; + stream->in.index = 0; + if (stream->out.packet) { stream->packet_sent_callback(stream->userdata, stream->out.packet); stream->out.packet = NULL; } - struct aki_packet *packet; - al_array_foreach(stream->out.queue, i, packet) { + + struct nn_packet *packet; + al_array_foreach(stream->out.queue, i, packet) stream->packet_sent_callback(stream->userdata, packet); - } stream->out.queue.size = 0; - aki_packet_reset(stream->in.packet); - stream->in.have_size = false; - stream->in.index = 0; + stream->connection_closed_callback(stream->userdata, stream); } static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_packet_stream *stream = (struct aki_packet_stream *)w->data; - al_assert(revents & EV_READ); // Assert on EV_ERROR. + struct nn_packet_stream *stream = (struct nn_packet_stream *)w->data; // EV_READ can mean any of POLLIN, POLLERR, or POLLHUP. - u8 *ptr = aki_buffer_get_ptr(&stream->in.packet->buffer, stream->in.index); - u32 size = stream->in.have_size ? aki_packet_get_size(stream->in.packet) : AKI_PACKET_HEADER_LENGTH; - ssize_t ret = aki_socket_read(&stream->sock, ptr, size - stream->in.index); + al_assert(revents & EV_READ); // Assert on EV_ERROR. + + u8 *ptr = nn_buffer_get_ptr(&stream->in.packet->buffer, stream->in.index); + u32 size = stream->in.have_header ? nn_packet_get_size(stream->in.packet) : NNWT_PACKET_HEADER_LENGTH; + 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) { - stop_write_internal(stream); - } + if (stream->out.active) stop_write_internal(stream); stop_internal(stream); return; } + stream->in.index += ret; - if (!stream->in.have_size && stream->in.index >= AKI_PACKET_HEADER_LENGTH) { - stream->in.have_size = true; - size = aki_packet_get_size(stream->in.packet); - aki_buffer_ensure_space(&stream->in.packet->buffer, size); + + if (!stream->in.have_header && stream->in.index >= NNWT_PACKET_HEADER_LENGTH) { + stream->in.have_header = true; + size = nn_packet_get_size(stream->in.packet); + nn_buffer_ensure_space(&stream->in.packet->buffer, size); } - if (stream->in.have_size && stream->in.index >= size) { - struct aki_packet *packet = stream->in.packet; - stream->in.packet = aki_packet_create(); + + if (stream->in.have_header && stream->in.index >= size) { + struct nn_packet *packet = stream->in.packet; + stream->in.packet = nn_packet_create(); stream->in.index = 0; - stream->in.have_size = false; + stream->in.have_header = false; stream->packet_callback(stream->userdata, stream, packet); // Stream could be invalid at this point. } @@ -122,11 +133,12 @@ static void stream_read_callback(struct ev_loop *loop, ev_io *w, s32 revents) static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_packet_stream *stream = (struct aki_packet_stream *)w->data; - al_assert(revents & EV_WRITE); // Assert on EV_ERROR. + struct nn_packet_stream *stream = (struct nn_packet_stream *)w->data; // EV_WRITE can mean any of POLLOUT, POLLERR, or POLLHUP. + al_assert(revents & EV_WRITE); // Assert on EV_ERROR. + if (UNLIKELY(stream->connect == PACKET_STREAM_CONNECTING)) { - ssize_t ret = aki_socket_write(&stream->sock, &stream->id, sizeof(u8)); + ssize_t ret = nn_socket_write(&stream->sock, &stream->id, sizeof(u8)); if (ret < 0) { // It seems POLLOUT can be signaled even if any write() will error. stop_write_internal(stream); @@ -136,16 +148,17 @@ 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; } - al_assert(ret == sizeof(u8)); + stream->connect = PACKET_STREAM_CONNECTED; - // Callbacks must be set in connection_callback. - stream->connection_callback(stream->userdata, stream); - al_assert(stream->packet_callback); - al_assert(stream->packet_sent_callback); - if (!stream->corked) { + al_assert(stream->corked); + if (stream->connection_callback(stream->userdata, stream)) { + stream->corked = false; ev_io_start(stream->loop->ev, &stream->revent); } + // Callbacks must be set in connection_callback. + al_assert(stream->packet_callback && stream->packet_sent_callback); } + if (!stream->out.packet) { if (stream->out.queue.size > 0) { al_array_pop_at(stream->out.queue, 0, stream->out.packet); @@ -155,109 +168,118 @@ static void stream_write_callback(struct ev_loop *loop, ev_io *w, s32 revents) return; } } - u8 *ptr = aki_buffer_get_ptr(&stream->out.packet->buffer, stream->out.index); - u32 size = aki_packet_get_size(stream->out.packet); - ssize_t ret = aki_socket_write(&stream->sock, ptr, size - stream->out.index); + + u8 *ptr = nn_buffer_get_ptr(&stream->out.packet->buffer, stream->out.index); + u32 size = nn_packet_get_size(stream->out.packet); + ssize_t ret = nn_socket_write(&stream->sock, ptr, size - stream->out.index); if (ret <= 0) { stop_write_internal(stream); return; } + stream->out.index += ret; + if (stream->out.index >= size) { stream->packet_sent_callback(stream->userdata, stream->out.packet); stream->out.packet = NULL; } } -void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_event_loop *loop, struct aki_socket *sock) +void nn_packet_stream_from_socket(struct nn_packet_stream *stream, struct nn_event_loop *loop, struct nn_socket *sock) { stream->loop = loop; - stream->corked = false; - stream->connect = PACKET_STREAM_CONNECTED; stream->sock = *sock; - aki_socket_set_blocking(&stream->sock, false); - init_internal(stream); - stream->connection_callback(stream->userdata, stream); + + init_io_state(stream); + stream->wevent.data = stream; - ev_io_init(&stream->wevent, stream_write_callback, - aki_socket_get_fd(&stream->sock), EV_WRITE); + ev_io_init(&stream->wevent, stream_write_callback, nn_socket_get_fd(&stream->sock), EV_WRITE); + stream->revent.data = stream; - ev_io_init(&stream->revent, stream_read_callback, - aki_socket_get_fd(&stream->sock), EV_READ); - ev_io_start(stream->loop->ev, &stream->revent); + 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; + ev_io_start(stream->loop->ev, &stream->revent); + } + al_assert(stream->connection_callback && stream->connection_closed_callback && + stream->packet_callback && stream->packet_sent_callback); } -static void connect_internal(struct aki_packet_stream *stream, str *addr, u16 port) +static void do_connect_internal(struct nn_packet_stream *stream, str *addr, u16 port) { - if (!aki_socket_connect(&stream->sock, addr, port)) { - aki_socket_close(&stream->sock); + if (!nn_socket_connect(&stream->sock, addr, port)) { + nn_socket_close(&stream->sock); stream->connection_closed_callback(stream->userdata, stream); return; } - stream->connect = PACKET_STREAM_CONNECTING; + stream->revent.data = stream; - ev_io_init(&stream->revent, stream_read_callback, - aki_socket_get_fd(&stream->sock), EV_READ); + ev_io_init(&stream->revent, stream_read_callback, nn_socket_get_fd(&stream->sock), EV_READ); + stream->wevent.data = stream; - ev_io_init(&stream->wevent, stream_write_callback, - aki_socket_get_fd(&stream->sock), EV_WRITE); + ev_io_init(&stream->wevent, stream_write_callback, nn_socket_get_fd(&stream->sock), EV_WRITE); + + // Start non-blocking connection. + stream->connect = PACKET_STREAM_CONNECTING; start_write_internal(stream); } -void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_event_loop *loop, str *addr, u16 port) +void nn_packet_stream_connect(struct nn_packet_stream *stream, struct nn_event_loop *loop, str *addr, u16 port) { stream->loop = loop; - connect_internal(stream, addr, port); + do_connect_internal(stream, addr, port); } -void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, u16 port) +void nn_packet_stream_reconnect(struct nn_packet_stream *stream, str *addr, u16 port) { al_assert(stream->connect == PACKET_STREAM_DISCONNECTED); - if (!aki_socket_init(&stream->sock)) { + + if (!nn_socket_init(&stream->sock, NNWT_SOCKET_NONBLOCKING)) { stream->connection_closed_callback(stream->userdata, stream); return; } - aki_socket_set_blocking(&stream->sock, false); - connect_internal(stream, addr, port); + + do_connect_internal(stream, addr, port); } -void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork) +void nn_packet_stream_cork(struct nn_packet_stream *stream, bool cork) { if (stream->connect == PACKET_STREAM_CONNECTED) { - if (cork && !stream->corked) { + if (cork && !stream->corked) ev_io_stop(stream->loop->ev, &stream->revent); - } else if (!cork && stream->corked) { + else if (!cork && stream->corked) ev_io_start(stream->loop->ev, &stream->revent); - } stream->corked = cork; } } -bool aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet) +bool nn_packet_stream_send_packet(struct nn_packet_stream *stream, struct nn_packet *packet) { - if (stream->connect == PACKET_STREAM_DISCONNECTING) { + if (stream->connect == PACKET_STREAM_DISCONNECTING) return false; - } + al_assert(stream->connect == PACKET_STREAM_CONNECTED); - aki_packet_write_size(packet); + + nn_packet_write_size(packet); al_array_push(stream->out.queue, packet); - if (!stream->out.active) { - start_write_internal(stream); - } + + if (!stream->out.active) start_write_internal(stream); + return true; } -void aki_packet_stream_disconnect(struct aki_packet_stream *stream) +void nn_packet_stream_disconnect(struct nn_packet_stream *stream) { al_assert(stream->connect != PACKET_STREAM_DISCONNECTING); - // To reuse this stream is may need to be uncorked. That should - // be tracked externally. - if (stream->connect == PACKET_STREAM_DISCONNECTED) { + + // If reusing this stream, keep in mind it will have been set + // back to a default state in stop_internal(). + if (stream->connect == PACKET_STREAM_DISCONNECTED) return; - } - // If we are not corked, this relies on stream_read_callback() to resolve and call stop_internal(). - // NOTE: read() can still return data after shutdown(). This is why we have to check if - // connect = DISCONNECTING in send_packet(). + if (stream->connect == PACKET_STREAM_CONNECTING) { // wevent is always active when connect = CONNECTING. stop_write_internal(stream); @@ -265,15 +287,18 @@ void aki_packet_stream_disconnect(struct aki_packet_stream *stream) } else if (stream->corked) { stop_internal(stream); } else { + // This relies on stream_read_callback() to resolve and call stop_internal(). + // Note that read() can still return data after shutdown(). This is why we have to + // check if connect = DISCONNECTING in send_packet(). shutdown_internal(stream); } } -void aki_packet_stream_free(struct aki_packet_stream *stream) +void nn_packet_stream_free(struct nn_packet_stream *stream) { - // Loosely assert that stop_internal has run. - al_assert(!stream->out.active); al_assert(stream->in.packet); - aki_packet_free(stream->in.packet); + // Loosely assert that stop_internal() has run. + al_assert(!stream->out.active); + 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 a6ca708..0121d3f 100644 --- a/src/packet_stream.h +++ b/src/packet_stream.h @@ -7,41 +7,41 @@ #include "loop.h" -struct aki_packet_stream { - struct aki_socket sock; - struct aki_event_loop *loop; +struct nn_packet_stream { + struct nn_event_loop *loop; + struct nn_socket sock; ev_io revent; ev_io wevent; - bool corked; u8 id; // multiplex u8 connect; - void (*connection_callback)(void *, struct aki_packet_stream *); - void (*connection_closed_callback)(void *, struct aki_packet_stream *); - void (*packet_callback)(void *, struct aki_packet_stream *, struct aki_packet *); - void (*packet_sent_callback)(void *, struct aki_packet *); - void *userdata; + bool corked; struct { - array(struct aki_packet *) queue; - struct aki_packet *packet; - bool active; + struct nn_packet *packet; + bool have_header; size_t index; - } out; + } in; struct { - struct aki_packet *packet; - bool have_size; + array(struct nn_packet *) queue; + struct nn_packet *packet; + bool active; size_t index; - } in; + } out; + 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 *userdata; }; -bool aki_packet_stream_init(struct aki_packet_stream *stream, u8 type, - void (*connection_callback)(void *, struct aki_packet_stream *), - void (*connection_closed_callback)(void *, struct aki_packet_stream *), void *userdata); -void aki_packet_stream_set_nodelay(struct aki_packet_stream *stream, s32 no_delay); -void aki_packet_stream_set_multiplex(struct aki_packet_stream *stream, u8 id); -void aki_packet_stream_from_socket(struct aki_packet_stream *stream, struct aki_event_loop *loop, struct aki_socket *sock); -void aki_packet_stream_connect(struct aki_packet_stream *stream, struct aki_event_loop *loop, str *addr, u16 port); -void aki_packet_stream_reconnect(struct aki_packet_stream *stream, str *addr, u16 port); -void aki_packet_stream_cork(struct aki_packet_stream *stream, bool cork); -bool aki_packet_stream_send_packet(struct aki_packet_stream *stream, struct aki_packet *packet); -void aki_packet_stream_disconnect(struct aki_packet_stream *stream); -void aki_packet_stream_free(struct aki_packet_stream *stream); +bool nn_packet_stream_init(struct nn_packet_stream *stream, u8 type, + 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_set_multiplex(struct nn_packet_stream *stream, u8 id); +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_reconnect(struct nn_packet_stream *stream, str *addr, u16 port); +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_disconnect(struct nn_packet_stream *stream); +void nn_packet_stream_free(struct nn_packet_stream *stream); @@ -3,11 +3,11 @@ static void event_callback(struct ev_loop *loop, ev_io *w, s32 revents) { (void)loop; - struct aki_poll *poll = (struct aki_poll *)w->data; + struct nn_poll *poll = (struct nn_poll *)w->data; poll->callback(poll->userdata, revents); } -void aki_poll_init(struct aki_poll *poll, void (*callback)(void *, s32 revents), void *userdata) +void nn_poll_init(struct nn_poll *poll, void (*callback)(void *, s32 revents), void *userdata) { poll->callback = callback; poll->userdata = userdata; @@ -15,18 +15,18 @@ void aki_poll_init(struct aki_poll *poll, void (*callback)(void *, s32 revents), ev_init(&poll->event, event_callback); } -void aki_poll_set(struct aki_poll *poll, s32 fd, s32 events) +void nn_poll_set(struct nn_poll *poll, s32 fd, s32 events) { ev_io_set(&poll->event, fd, events); } -void aki_poll_start(struct aki_poll *poll, struct aki_event_loop *loop) +void nn_poll_start(struct nn_poll *poll, struct nn_event_loop *loop) { poll->loop = loop; ev_io_start(poll->loop->ev, &poll->event); } -void aki_poll_stop(struct aki_poll *poll) +void nn_poll_stop(struct nn_poll *poll) { ev_io_stop(poll->loop->ev, &poll->event); } @@ -2,17 +2,17 @@ #include "loop.h" -#define AKI_POLL_READ EV_READ -#define AKI_POLL_WRITE EV_WRITE +#define NNWT_POLL_READ EV_READ +#define NNWT_POLL_WRITE EV_WRITE -struct aki_poll { +struct nn_poll { ev_io event; - struct aki_event_loop *loop; + struct nn_event_loop *loop; void (*callback)(void *, s32 revents); void *userdata; }; -void aki_poll_init(struct aki_poll *poll, void (*callback)(void *, s32 revents), void *userdata); -void aki_poll_start(struct aki_poll *poll, struct aki_event_loop *loop); -void aki_poll_set(struct aki_poll *poll, s32 fd, s32 events); -void aki_poll_stop(struct aki_poll *poll); +void nn_poll_init(struct nn_poll *poll, void (*callback)(void *, s32 revents), void *userdata); +void nn_poll_start(struct nn_poll *poll, struct nn_event_loop *loop); +void nn_poll_set(struct nn_poll *poll, s32 fd, s32 events); +void nn_poll_stop(struct nn_poll *poll); @@ -1,8 +1,8 @@ #include "rpc2.h" -bool aki_rpc_init(struct aki_rpc *rpc, struct aki_event_loop *loop, - void (*connection_callback)(void *, struct aki_rpc_connection *), - void (*connection_closed_callback)(void *, struct aki_rpc_connection *), void *userdata) +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) { rpc->loop = loop; rpc->increment = 0; @@ -15,141 +15,150 @@ bool aki_rpc_init(struct aki_rpc *rpc, struct aki_event_loop *loop, return true; } -void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command) +void nn_rpc_add_command(struct nn_rpc *rpc, struct nn_rpc_command *command) { al_array_push(rpc->commands, *command); } -static void packet_callback(void *userdata, struct aki_packet_stream *stream, struct aki_packet *packet) +static void packet_callback(void *userdata, struct nn_packet_stream *stream, struct nn_packet *packet) { - struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata; - s8 op = aki_packet_read_s8(packet); - u32 id = aki_packet_read_u32(packet); + struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata; + s8 op = nn_packet_read_s8(packet); + u32 id = nn_packet_read_u32(packet); if (op == -1) { // Response - struct aki_rpc_callback *callback; + struct nn_rpc_callback *callback; al_array_foreach_ptr(conn->callbacks, i, callback) { if (callback->id == id) { callback->callback(callback->userdata, packet); - al_array_remove_at_iter(conn->callbacks, i); + al_array_remove_at(conn->callbacks, i); return; } } - } else { // Command - struct aki_rpc_command *command; + } + else { // Command + struct nn_rpc_command *command; al_array_foreach_ptr(conn->rpc->commands, i, command) { if (command->op == op) { - struct aki_packet *rpacket = aki_packet_create(); - aki_packet_write_s8(rpacket, -1); - aki_packet_write_u32(rpacket, id); + struct nn_packet *rpacket = nn_packet_create(); + nn_packet_write_s8(rpacket, -1); + nn_packet_write_u32(rpacket, id); if (command->callback(command->userdata, conn, packet, rpacket)) { conn->outgoing++; - aki_packet_stream_send_packet(stream, rpacket); - } else { - aki_packet_free(rpacket); + nn_packet_stream_send_packet(stream, rpacket); + } + else { + nn_packet_free(rpacket); } return; } } } - aki_packet_free(packet); + nn_packet_free(packet); } -static void packet_sent_callback(void *userdata, struct aki_packet *packet) +static void packet_sent_callback(void *userdata, struct nn_packet *packet) { - struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata; - aki_packet_free(packet); + struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata; + nn_packet_free(packet); al_assert(conn->outgoing > 0); conn->outgoing--; - if (conn->flushing && conn->outgoing == 0) { - aki_packet_stream_disconnect(&conn->stream); - } + if (conn->flushing && conn->outgoing == 0) + nn_packet_stream_disconnect(&conn->stream); } -static void stream_connection_callback(void *userdata, struct aki_packet_stream *stream) +static bool stream_connection_callback(void *userdata, struct nn_packet_stream *stream) { - struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata; - struct aki_rpc *rpc = conn->rpc; - if (stream->sock.type == AKI_SOCKET_TCP) { - aki_packet_stream_set_nodelay(stream, 1); - } + struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata; + struct nn_rpc *rpc = conn->rpc; + + if (stream->sock.type == NNWT_SOCKET_TCP) + nn_packet_stream_set_nodelay(stream, 1); + stream->packet_callback = packet_callback; stream->packet_sent_callback = packet_sent_callback; stream->userdata = conn; + rpc->connection_callback(rpc->userdata, conn); - // Only add connections in server mode. + + // Only add connections when acting as a server. if (!rpc->conn) al_array_push(rpc->connections, conn); + + return true; } -static void stream_connection_closed_callback(void *userdata, struct aki_packet_stream *stream) +static void stream_connection_closed_callback(void *userdata, struct nn_packet_stream *stream) { - struct aki_rpc_connection *conn = (struct aki_rpc_connection *)userdata; + struct nn_rpc_connection *conn = (struct nn_rpc_connection *)userdata; al_assert(stream == &conn->stream); conn->rpc->connection_closed_callback(conn->rpc->userdata, conn); } -static void init_connection(struct aki_rpc_connection *conn) +static inline void init_rpc_connection(struct nn_rpc *rpc, struct nn_rpc_connection *conn) { + conn->rpc = rpc; al_array_init(conn->callbacks); conn->outgoing = 0; conn->flushing = false; } -void aki_rpc_add_socket(struct aki_rpc *rpc, struct aki_socket *sock) +void nn_rpc_add_socket(struct nn_rpc *rpc, struct nn_socket *sock) { - struct aki_rpc_connection *conn = al_alloc_object(struct aki_rpc_connection); - conn->rpc = rpc; - init_connection(conn); + 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; - aki_packet_stream_from_socket(&conn->stream, rpc->loop, sock); + + nn_packet_stream_from_socket(&conn->stream, rpc->loop, sock); } -bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id) +bool nn_rpc_prepare_client(struct nn_rpc *rpc, u8 type, u8 id) { al_assert(!rpc->conn); - struct aki_rpc_connection *conn = al_alloc_object(struct aki_rpc_connection); - rpc->conn = conn; - al_array_push(rpc->connections, conn); - if (!aki_packet_stream_init(&conn->stream, type, stream_connection_callback, - stream_connection_closed_callback, 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, + stream_connection_callback, stream_connection_closed_callback, conn)) { + al_free(conn); return false; } - conn->rpc = rpc; - init_connection(conn); - if (conn->stream.sock.type == AKI_SOCKET_TCP) { - aki_packet_stream_set_nodelay(&conn->stream, 1); - } - aki_packet_stream_set_multiplex(&conn->stream, id); + nn_packet_stream_set_multiplex(&conn->stream, id); + + rpc->conn = conn; + al_array_push(rpc->connections, conn); + return true; } -void aki_rpc_connect(struct aki_rpc *rpc, str *addr, u16 port) +void nn_rpc_connect(struct nn_rpc *rpc, str *addr, u16 port) { al_assert(rpc->conn); - aki_packet_stream_connect(&rpc->conn->stream, rpc->loop, addr, port); + nn_packet_stream_connect(&rpc->conn->stream, rpc->loop, addr, port); } -void aki_rpc_reconnect(struct aki_rpc *rpc, str *addr, u16 port) +void nn_rpc_reconnect(struct nn_rpc *rpc, str *addr, u16 port) { al_assert(rpc->conn); - aki_packet_stream_reconnect(&rpc->conn->stream, addr, port); + nn_packet_stream_reconnect(&rpc->conn->stream, addr, port); } -struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op) +struct nn_packet *nn_rpc_get_packet(struct nn_rpc *rpc, s8 op) { - struct aki_packet *packet = aki_packet_create(); - aki_packet_write_s8(packet, op); - aki_packet_write_u32(packet, rpc->increment); + struct nn_packet *packet = nn_packet_create(); + nn_packet_write_s8(packet, op); + nn_packet_write_u32(packet, rpc->increment); rpc->increment = al_u32_inc_wrap(rpc->increment); return packet; } -void aki_rpc_free(struct aki_rpc *rpc) +void nn_rpc_free(struct nn_rpc *rpc) { - struct aki_rpc_connection *conn; + struct nn_rpc_connection *conn; al_array_foreach(rpc->connections, i, conn) { - aki_packet_stream_free(&conn->stream); + nn_packet_stream_free(&conn->stream); al_array_free(conn->callbacks); al_free(conn); } @@ -157,30 +166,30 @@ void aki_rpc_free(struct aki_rpc *rpc) al_array_free(rpc->commands); } -void aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_packet *packet, - void (*callback)(void *, struct aki_packet *), void *userdata) +void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet *packet, + void (*callback)(void *, struct nn_packet *), void *userdata) { if (callback) { - al_array_push(conn->callbacks, ((struct aki_rpc_callback){ - .id = AKI_PACKET_GET_ID(packet), + al_array_push(conn->callbacks, ((struct nn_rpc_callback){ + .id = NNWT_PACKET_GET_ID(packet), .callback = callback, .userdata = userdata })); } conn->outgoing++; - aki_packet_stream_send_packet(&conn->stream, packet); + nn_packet_stream_send_packet(&conn->stream, packet); } -void aki_rpc_conn_flush(struct aki_rpc_connection *conn) +void nn_rpc_conn_flush(struct nn_rpc_connection *conn) { - if (conn->outgoing == 0) { - aki_packet_stream_disconnect(&conn->stream); - } else { + if (conn->outgoing != 0) { conn->flushing = true; + } else { + nn_packet_stream_disconnect(&conn->stream); } } -void aki_rpc_conn_disconnect(struct aki_rpc_connection *conn) +void nn_rpc_conn_disconnect(struct nn_rpc_connection *conn) { - aki_packet_stream_disconnect(&conn->stream); + nn_packet_stream_disconnect(&conn->stream); } @@ -6,50 +6,50 @@ #include "packet_stream.h" -struct aki_rpc_callback { +struct nn_rpc_callback { u32 id; - void (*callback)(void *, struct aki_packet *); + void (*callback)(void *, struct nn_packet *); void *userdata; }; -struct aki_rpc_connection { - struct aki_packet_stream stream; - array(struct aki_rpc_callback) callbacks; +struct nn_rpc_connection { + struct nn_packet_stream stream; + array(struct nn_rpc_callback) callbacks; u32 outgoing; bool flushing; - struct aki_rpc *rpc; + struct nn_rpc *rpc; }; -struct aki_rpc_command { +struct nn_rpc_command { s8 op; - bool (*callback)(void *, struct aki_rpc_connection *, struct aki_packet *, struct aki_packet *); + bool (*callback)(void *, struct nn_rpc_connection *, struct nn_packet *, struct nn_packet *); void *userdata; }; -struct aki_rpc { - struct aki_event_loop *loop; +struct nn_rpc { + struct nn_event_loop *loop; u32 increment; - array(struct aki_rpc_command) commands; - struct aki_rpc_connection *conn; // client - array(struct aki_rpc_connection *) connections; - void (*connection_callback)(void *, struct aki_rpc_connection *); - void (*connection_closed_callback)(void *, struct aki_rpc_connection *); + array(struct nn_rpc_command) commands; + struct nn_rpc_connection *conn; // client + array(struct nn_rpc_connection *) connections; + void (*connection_callback)(void *, struct nn_rpc_connection *); + void (*connection_closed_callback)(void *, struct nn_rpc_connection *); void *userdata; }; -bool aki_rpc_init(struct aki_rpc *rpc, struct aki_event_loop *loop, - void (*connection_callback)(void *, struct aki_rpc_connection *), - void (*connection_closed_callback)(void *, struct aki_rpc_connection *), void *userdata); -void aki_rpc_add_command(struct aki_rpc *rpc, struct aki_rpc_command *command); -void aki_rpc_add_socket(struct aki_rpc *rpc, struct aki_socket *sock); -bool aki_rpc_prepare_client(struct aki_rpc *rpc, u8 type, u8 id); -void aki_rpc_connect(struct aki_rpc *rpc, str *addr, u16 port); -void aki_rpc_reconnect(struct aki_rpc *rpc, str *addr, u16 port); -struct aki_packet *aki_rpc_get_packet(struct aki_rpc *rpc, s8 op); -void aki_rpc_free(struct aki_rpc *rpc); +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_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 aki_rpc_connection_command(struct aki_rpc_connection *conn, struct aki_packet *packet, - void (*callback)(void *, struct aki_packet *), void *userdata); -void aki_rpc_conn_flush(struct aki_rpc_connection *conn); -void aki_rpc_conn_disconnect(struct aki_rpc_connection *conn); +void nn_rpc_connection_command(struct nn_rpc_connection *conn, struct nn_packet *packet, + void (*callback)(void *, 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/signal.c b/src/signal.c index f7ebd6d..a5ad885 100644 --- a/src/signal.c +++ b/src/signal.c @@ -3,12 +3,12 @@ static void signal_callback(struct ev_loop *loop, ev_async *w, s32 revents) { (void)loop; - struct aki_signal *signal = (struct aki_signal *)w->data; + struct nn_signal *signal = (struct nn_signal *)w->data; (void)revents; signal->callback(signal->userdata); } -void aki_signal_init(struct aki_signal *signal, struct aki_event_loop *loop, +void nn_signal_init(struct nn_signal *signal, struct nn_event_loop *loop, void (*callback)(void *), void *userdata) { signal->loop = loop; @@ -18,17 +18,17 @@ void aki_signal_init(struct aki_signal *signal, struct aki_event_loop *loop, ev_async_init(&signal->signal, signal_callback); } -void aki_signal_start(struct aki_signal *signal) +void nn_signal_start(struct nn_signal *signal) { ev_async_start(signal->loop->ev, &signal->signal); } -void aki_signal_send(struct aki_signal *signal) +void nn_signal_send(struct nn_signal *signal) { ev_async_send(signal->loop->ev, &signal->signal); } -void aki_signal_stop(struct aki_signal *signal) +void nn_signal_stop(struct nn_signal *signal) { ev_async_stop(signal->loop->ev, &signal->signal); } diff --git a/src/signal.h b/src/signal.h index 9423432..b9f0edb 100644 --- a/src/signal.h +++ b/src/signal.h @@ -2,15 +2,15 @@ #include "loop.h" -struct aki_signal { +struct nn_signal { ev_async signal; - struct aki_event_loop *loop; + struct nn_event_loop *loop; void (*callback)(void *); void *userdata; }; -void aki_signal_init(struct aki_signal *signal, struct aki_event_loop *loop, +void nn_signal_init(struct nn_signal *signal, struct nn_event_loop *loop, void (*callback)(void *), void *userdata); -void aki_signal_start(struct aki_signal *signal); -void aki_signal_send(struct aki_signal *signal); -void aki_signal_stop(struct aki_signal *signal); +void nn_signal_start(struct nn_signal *signal); +void nn_signal_send(struct nn_signal *signal); +void nn_signal_stop(struct nn_signal *signal); diff --git a/src/socket/socket.h b/src/socket/socket.h index cf8953c..6b53812 100644 --- a/src/socket/socket.h +++ b/src/socket/socket.h @@ -17,12 +17,17 @@ #endif enum { - AKI_SOCKET_TCP = 0, - AKI_SOCKET_UDP, - AKI_SOCKET_UNIX + NNWT_SOCKET_TCP = 0, + NNWT_SOCKET_UDP, + NNWT_SOCKET_UNIX }; -struct aki_socket { +enum { + NNWT_SOCKET_NONBLOCKING = 1, + NNWT_SOCKET_NODELAY = 1 << 1 +}; + +struct nn_socket { u8 type; #ifndef _WIN32 s32 fd; @@ -36,47 +41,47 @@ struct aki_socket { }; #ifndef _WIN32 -#define aki_read read -#define aki_write write +#define nn_read read +#define nn_write write #else -#define aki_read _read -#define aki_write _write +#define nn_read _read +#define nn_write _write #endif -#define aki_htons htons -#define aki_htonl htonl -#define aki_ntohs ntohs -#define aki_ntohl ntohl +#define nn_htons htons +#define nn_htonl htonl +#define nn_ntohs ntohs +#define nn_ntohl ntohl #ifndef _WIN32 // Posix-only helpers. -#define aki_pollfd pollfd -#define aki_nfds nfds_t -void aki_fd_set_blocking(s32 fd, bool blocking); -s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns); +#define nn_pollfd pollfd +#define nn_nfds nfds_t +void nn_fd_set_blocking(s32 fd, bool blocking); +s32 nn_poll_fds(struct nn_pollfd *fds, nn_nfds nfds, s64 timeout_ns); #endif -bool aki_socket_init(struct aki_socket *sock); +bool nn_socket_init(struct nn_socket *sock, s32 flags); -void aki_socket_set_blocking(struct aki_socket *sock, bool blocking); -void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay); +void nn_socket_set_blocking(struct nn_socket *sock, bool blocking); +void nn_socket_set_nodelay(struct nn_socket *sock, s32 nodelay); -u32 aki_socket_get_send_buf(struct aki_socket *sock); -void aki_socket_set_send_buf(struct aki_socket *sock, u32 sndbuf); -u32 aki_socket_get_recv_buf(struct aki_socket *sock); -void aki_socket_set_recv_buf(struct aki_socket *sock, u32 rcvbuf); +u32 nn_socket_get_send_buf(struct nn_socket *sock); +void nn_socket_set_send_buf(struct nn_socket *sock, u32 sndbuf); +u32 nn_socket_get_recv_buf(struct nn_socket *sock); +void nn_socket_set_recv_buf(struct nn_socket *sock, u32 rcvbuf); -bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port); -bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port); -bool aki_socket_listen(struct aki_socket *sock); -bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *c); -bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port); +bool nn_socket_set(struct nn_socket *sock, str *addr, u16 port); +bool nn_socket_bind(struct nn_socket *sock, str *addr, u16 port); +bool nn_socket_listen(struct nn_socket *sock); +bool nn_socket_accept(struct nn_socket *sock, struct nn_socket *c, s32 flags); +bool nn_socket_connect(struct nn_socket *sock, str *addr, u16 port); -s32 aki_socket_get_fd(struct aki_socket *sock); +s32 nn_socket_get_fd(struct nn_socket *sock); -ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size); -ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size); -ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size); -ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size); +ssize_t nn_socket_read(struct nn_socket *sock, void *buf, size_t size); +ssize_t nn_socket_write(struct nn_socket *sock, void *buf, size_t size); +ssize_t nn_socket_sendto(struct nn_socket *sock, void *buf, size_t size); +ssize_t nn_socket_recvfrom(struct nn_socket *sock, void *buf, size_t size); -void aki_socket_shutdown(struct aki_socket *sock); -void aki_socket_close(struct aki_socket *sock); +void nn_socket_shutdown(struct nn_socket *sock); +void nn_socket_close(struct nn_socket *sock); diff --git a/src/socket/socket_internal.h b/src/socket/socket_internal.h index 33f6a7b..fd9a001 100644 --- a/src/socket/socket_internal.h +++ b/src/socket/socket_internal.h @@ -2,10 +2,21 @@ #include <al/str.h> +#include "socket.h" + static str *ipany = al_str_c("0.0.0.0"); -static inline str *aki_socket_display_addr(str *addr) +static inline str *nn_socket_display_addr(str *addr) { return addr ? addr : ipany; } -#define aki_socket_printf_addr(addr) AL_STR_PRINTF(aki_socket_display_addr(addr)) +#define nn_socket_printf_addr(addr) AL_STR_PRINTF(nn_socket_display_addr(addr)) + +static inline void nn_socket_apply_flags(struct nn_socket *sock, s32 flags) +{ + if (flags & NNWT_SOCKET_NONBLOCKING) + nn_socket_set_blocking(sock, false); + + if (flags & NNWT_SOCKET_NODELAY) + nn_socket_set_nodelay(sock, 1); +} diff --git a/src/socket/socket_linux.c b/src/socket/socket_linux.c index 6d8b3fe..c3467dd 100644 --- a/src/socket/socket_linux.c +++ b/src/socket/socket_linux.c @@ -5,7 +5,7 @@ #include "socket.h" #include "socket_internal.h" -void aki_fd_set_blocking(s32 fd, bool blocking) +void nn_fd_set_blocking(s32 fd, bool blocking) { s32 ret = fcntl(fd, F_GETFL, 0); // ret = flags al_assert(ret != -1); @@ -14,26 +14,27 @@ void aki_fd_set_blocking(s32 fd, bool blocking) } // https://github.com/mpv-player/mpv/blob/e575ec4fc3654387c7358bd3640877ef32628d2c/osdep/poll_wrapper.c#L29 -#define AKI_TIME_S_TO_NS(s) ((s) * INT64_C(1000000000)) -s32 aki_poll_fds(struct aki_pollfd *fds, aki_nfds nfds, s64 timeout_ns) +#define NNWT_TIME_S_TO_NS(s) ((s) * INT64_C(1000000000)) +s32 nn_poll_fds(struct nn_pollfd *fds, nn_nfds nfds, s64 timeout_ns) { struct timespec ts; - ts.tv_sec = timeout_ns / AKI_TIME_S_TO_NS(1); - ts.tv_nsec = timeout_ns % AKI_TIME_S_TO_NS(1); + ts.tv_sec = timeout_ns / NNWT_TIME_S_TO_NS(1); + ts.tv_nsec = timeout_ns % NNWT_TIME_S_TO_NS(1); struct timespec *tsp = timeout_ns >= 0 ? &ts : NULL; return ppoll(fds, nfds, tsp, NULL); } -bool aki_socket_init(struct aki_socket *sock) + +bool nn_socket_init(struct nn_socket *sock, s32 flags) { switch (sock->type) { - case AKI_SOCKET_TCP: + case NNWT_SOCKET_TCP: sock->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); break; - case AKI_SOCKET_UDP: + case NNWT_SOCKET_UDP: sock->fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); break; - case AKI_SOCKET_UNIX: + case NNWT_SOCKET_UNIX: sock->fd = socket(AF_UNIX, SOCK_STREAM, 0); break; default: @@ -41,27 +42,29 @@ bool aki_socket_init(struct aki_socket *sock) } if (sock->fd < 0) { - al_log_error("socket", "socket() failed: %s (%d).", aki_strerror(errno), errno); + al_log_error("socket", "socket() failed: %s (%d).", nn_strerror(errno), errno); return false; } + nn_socket_apply_flags(sock, flags); + sock->addrinfo = NULL; return true; } -void aki_socket_set_blocking(struct aki_socket *sock, bool blocking) +void nn_socket_set_blocking(struct nn_socket *sock, bool blocking) { - aki_fd_set_blocking(sock->fd, blocking); + nn_fd_set_blocking(sock->fd, blocking); } -void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay) +void nn_socket_set_nodelay(struct nn_socket *sock, s32 nodelay) { - al_assert(sock->type == AKI_SOCKET_TCP); + al_assert(sock->type == NNWT_SOCKET_TCP); setsockopt(sock->fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay)); } -u32 aki_socket_get_send_buf(struct aki_socket *sock) +u32 nn_socket_get_send_buf(struct nn_socket *sock) { u32 send_queue_size; socklen_t optlen = sizeof(send_queue_size); @@ -69,14 +72,14 @@ u32 aki_socket_get_send_buf(struct aki_socket *sock) return send_queue_size / 2; } -void aki_socket_set_send_buf(struct aki_socket *sock, u32 sndbuf) +void nn_socket_set_send_buf(struct nn_socket *sock, u32 sndbuf) { u32 send_queue_size = sndbuf; setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &send_queue_size, sizeof(send_queue_size)); - al_assert(aki_socket_get_send_buf(sock) == send_queue_size); + al_assert(nn_socket_get_send_buf(sock) == send_queue_size); } -u32 aki_socket_get_recv_buf(struct aki_socket *sock) +u32 nn_socket_get_recv_buf(struct nn_socket *sock) { u32 receive_queue_size; socklen_t optlen = sizeof(receive_queue_size); @@ -84,23 +87,23 @@ u32 aki_socket_get_recv_buf(struct aki_socket *sock) return receive_queue_size / 2; } -void aki_socket_set_recv_buf(struct aki_socket *sock, u32 rcvbuf) +void nn_socket_set_recv_buf(struct nn_socket *sock, u32 rcvbuf) { u32 receive_queue_size = rcvbuf; setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &receive_queue_size, sizeof(receive_queue_size)); - al_assert(aki_socket_get_recv_buf(sock) == receive_queue_size); + al_assert(nn_socket_get_recv_buf(sock) == receive_queue_size); } -static bool parse_address(struct aki_socket *sock, str *addr, u16 port) +static bool parse_address(struct nn_socket *sock, str *addr, u16 port) { struct addrinfo hints = { 0 }; hints.ai_family = AF_INET; switch (sock->type) { - case AKI_SOCKET_TCP: + case NNWT_SOCKET_TCP: hints.ai_socktype = SOCK_STREAM; hints.ai_protocol = IPPROTO_TCP; break; - case AKI_SOCKET_UDP: + case NNWT_SOCKET_UDP: hints.ai_socktype = SOCK_DGRAM; hints.ai_protocol = IPPROTO_UDP; break; @@ -120,26 +123,26 @@ static bool parse_address(struct aki_socket *sock, str *addr, u16 port) return true; } -bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_set(struct nn_socket *sock, str *addr, u16 port) { - al_assert(sock->type == AKI_SOCKET_UDP); + al_assert(sock->type == NNWT_SOCKET_UDP); return parse_address(sock, addr, port); } -bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_bind(struct nn_socket *sock, str *addr, u16 port) { struct sockaddr *saddr = NULL; socklen_t addrlen = 0; switch (sock->type) { - case AKI_SOCKET_TCP: - case AKI_SOCKET_UDP: { - if (!parse_address(sock, addr, port)) return false; + case NNWT_SOCKET_TCP: + case NNWT_SOCKET_UDP: + if (!parse_address(sock, addr, port)) + return false; saddr = sock->addrinfo->ai_addr; addrlen = sock->addrinfo->ai_addrlen; break; - } - case AKI_SOCKET_UNIX: { + case NNWT_SOCKET_UNIX: { al_assert(addr); sock->addr_un.sun_family = AF_UNIX; char *c_str = al_str_to_c_str(addr); @@ -156,21 +159,21 @@ bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port) if (bind(sock->fd, saddr, addrlen) < 0) { al_log_error("socket", "bind(%.*s:%hu) failed: %s (%d).", - aki_socket_printf_addr(addr), port, aki_strerror(errno), errno); + nn_socket_printf_addr(addr), port, nn_strerror(errno), errno); return false; } - al_log_info("socket", "Socket bound to %.*s:%hu.", aki_socket_printf_addr(addr), port); + al_log_info("socket", "Socket bound to %.*s:%hu.", nn_socket_printf_addr(addr), port); return true; } -bool aki_socket_listen(struct aki_socket *sock) +bool nn_socket_listen(struct nn_socket *sock) { - al_assert(sock->type == AKI_SOCKET_TCP || sock->type == AKI_SOCKET_UNIX); + al_assert(sock->type == NNWT_SOCKET_TCP || sock->type == NNWT_SOCKET_UNIX); if (listen(sock->fd, SOMAXCONN) < 0) { - al_log_error("socket", "listen() failed: %s (%d).", aki_strerror(errno), errno); + al_log_error("socket", "listen() failed: %s (%d).", nn_strerror(errno), errno); return false; } @@ -179,22 +182,24 @@ bool aki_socket_listen(struct aki_socket *sock) return true; } -bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl) +bool nn_socket_accept(struct nn_socket *sock, struct nn_socket *cl, s32 flags) { - al_assert(sock->type == AKI_SOCKET_TCP || sock->type == AKI_SOCKET_UNIX); - + al_assert(sock->type == NNWT_SOCKET_TCP || sock->type == NNWT_SOCKET_UNIX); cl->type = sock->type; + struct sockaddr_storage addr = { 0 }; socklen_t addrlen = sizeof(addr); if ((cl->fd = accept(sock->fd, (struct sockaddr *)&addr, &addrlen)) == -1) { - al_log_error("socket", "accept() failed: %s (%d).", aki_strerror(errno), errno); + al_log_error("socket", "accept() failed: %s (%d).", nn_strerror(errno), errno); return false; } + nn_socket_apply_flags(cl, flags); + cl->addrinfo = NULL; switch (cl->type) { - case AKI_SOCKET_TCP: { + case NNWT_SOCKET_TCP: { u16 port = 0; char addr_str[INET6_ADDRSTRLEN] = { 0 }; if (addr.ss_family == AF_INET) { @@ -209,7 +214,7 @@ bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl) al_log_info("socket", "Connection from %s:%hu.", addr_str, port); break; } - case AKI_SOCKET_UNIX: + case NNWT_SOCKET_UNIX: al_log_info("socket", "Connection on UNIX socket."); break; } @@ -217,16 +222,17 @@ bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl) return true; } -bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_connect(struct nn_socket *sock, str *addr, u16 port) { s32 ret; switch (sock->type) { - case AKI_SOCKET_TCP: - case AKI_SOCKET_UDP: - if (!parse_address(sock, addr, port)) return false; + case NNWT_SOCKET_TCP: + case NNWT_SOCKET_UDP: + if (!parse_address(sock, addr, port)) + return false; ret = connect(sock->fd, sock->addrinfo->ai_addr, sock->addrinfo->ai_addrlen); break; - case AKI_SOCKET_UNIX: + case NNWT_SOCKET_UNIX: sock->addr_un.sun_family = AF_UNIX; al_memcpy(sock->addr_un.sun_path, addr->data, addr->len); sock->addr_un.sun_path[addr->len] = '\0'; @@ -238,44 +244,44 @@ bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port) if (ret != 0 && errno != EINPROGRESS) { al_log_error("socket", "connect(%.*s:%hu) failed: %s (%d).", - aki_socket_printf_addr(addr), port, aki_strerror(errno), errno); + nn_socket_printf_addr(addr), port, nn_strerror(errno), errno); return false; } return true; } -ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_read(struct nn_socket *sock, void *buf, size_t size) { return recv(sock->fd, buf, size, 0); } -ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_write(struct nn_socket *sock, void *buf, size_t size) { return send(sock->fd, buf, size, MSG_NOSIGNAL); } -ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_sendto(struct nn_socket *sock, void *buf, size_t size) { return sendto(sock->fd, buf, size, 0, sock->addrinfo->ai_addr, sock->addrinfo->ai_addrlen); } -ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_recvfrom(struct nn_socket *sock, void *buf, size_t size) { return recvfrom(sock->fd, buf, size, 0, sock->addrinfo->ai_addr, &sock->addrinfo->ai_addrlen); } -s32 aki_socket_get_fd(struct aki_socket *sock) +s32 nn_socket_get_fd(struct nn_socket *sock) { return sock->fd; } -void aki_socket_shutdown(struct aki_socket *sock) +void nn_socket_shutdown(struct nn_socket *sock) { shutdown(sock->fd, SHUT_RDWR); } -void aki_socket_close(struct aki_socket *sock) +void nn_socket_close(struct nn_socket *sock) { close(sock->fd); if (sock->addrinfo) { diff --git a/src/socket/socket_windows.c b/src/socket/socket_windows.c index 013a5bd..28e6dc5 100644 --- a/src/socket/socket_windows.c +++ b/src/socket/socket_windows.c @@ -3,12 +3,12 @@ #include "socket.h" #include "socket_internal.h" -bool aki_socket_init(struct aki_socket *sock) +bool nn_socket_init(struct nn_socket *sock, s32 flags) { - al_assert(sock->type == AKI_SOCKET_TCP); + al_assert(sock->type == NNWT_SOCKET_TCP); switch (sock->type) { - case AKI_SOCKET_TCP: + case NNWT_SOCKET_TCP: sock->fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); break; } @@ -18,24 +18,26 @@ bool aki_socket_init(struct aki_socket *sock) return false; } + nn_socket_apply_flags(sock, flags); + sock->internal_fd = _open_osfhandle(sock->fd, 0); return true; } -void aki_socket_set_blocking(struct aki_socket *sock, bool blocking) +void nn_socket_set_blocking(struct nn_socket *sock, bool blocking) { u_long mode = blocking ? 0 : 1; ioctlsocket(sock->fd, FIONBIO, &mode); } -void aki_socket_set_nodelay(struct aki_socket *sock, s32 nodelay) +void nn_socket_set_nodelay(struct nn_socket *sock, s32 nodelay) { (void)sock; (void)nodelay; } -bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_set(struct nn_socket *sock, str *addr, u16 port) { (void)sock; (void)addr; @@ -43,24 +45,27 @@ bool aki_socket_set(struct aki_socket *sock, str *addr, u16 port) return false; } -bool aki_socket_bind(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_bind(struct nn_socket *sock, str *addr, u16 port) { sock->addr_in.sin_family = AF_INET; - sock->addr_in.sin_addr.s_addr = INADDR_ANY; - sock->addr_in.sin_port = aki_htons(port); al_memset(sock->addr_in.sin_zero, '\0', sizeof(sock->addr_in.sin_zero)); - if (bind(sock->fd, (SOCKADDR *)&sock->addr_in, sizeof(sock->addr_in)) == SOCKET_ERROR) { + + sock->addr_in.sin_addr.s_addr = INADDR_ANY; + sock->addr_in.sin_port = nn_htons(port); + + s32 ret = bind(sock->fd, (SOCKADDR *)&sock->addr_in, sizeof(sock->addr_in)); + if (ret == SOCKET_ERROR) { al_log_error("socket", "bind(%.*s:%hu) failed: %d.", - aki_socket_printf_addr(addr), port, WSAGetLastError()); + nn_socket_printf_addr(addr), port, WSAGetLastError()); return false; } - al_log_info("socket", "Socket bound to %.*s:%hu.", aki_socket_printf_addr(addr), port); + al_log_info("socket", "Socket bound to %.*s:%hu.", nn_socket_printf_addr(addr), port); return true; } -bool aki_socket_listen(struct aki_socket *sock) +bool nn_socket_listen(struct nn_socket *sock) { if (listen(sock->fd, SOMAXCONN) == SOCKET_ERROR) { al_log_error("socket", "listen() failed: %d.", WSAGetLastError()); @@ -72,49 +77,56 @@ bool aki_socket_listen(struct aki_socket *sock) return true; } -bool aki_socket_accept(struct aki_socket *sock, struct aki_socket *cl) +bool nn_socket_accept(struct nn_socket *sock, struct nn_socket *cl, s32 flags) { cl->type = sock->type; if ((cl->fd = accept(sock->fd, NULL, NULL)) == INVALID_SOCKET) { al_log_error("socket", "accept() failed: %d.", WSAGetLastError()); return false; } + + nn_socket_apply_flags(sock, flags); + cl->internal_fd = _open_osfhandle(cl->fd, 0); + return true; } -bool aki_socket_connect(struct aki_socket *sock, str *addr, u16 port) +bool nn_socket_connect(struct nn_socket *sock, str *addr, u16 port) { sock->addr_in.sin_family = AF_INET; + al_memset(sock->addr_in.sin_zero, '\0', sizeof(sock->addr_in.sin_zero)); + char *c_str = al_str_to_c_str(addr); sock->addr_in.sin_addr.s_addr = inet_addr(c_str); al_free(c_str); - sock->addr_in.sin_port = aki_htons(port); - al_memset(sock->addr_in.sin_zero, '\0', sizeof(sock->addr_in.sin_zero)); + sock->addr_in.sin_port = nn_htons(port); + s32 ret = connect(sock->fd, (SOCKADDR *)&sock->addr_in, sizeof(sock->addr_in)); if (ret == SOCKET_ERROR && WSAGetLastError() != WSAEWOULDBLOCK) { al_log_error("socket", "connect() failed: %d.", WSAGetLastError()); return false; } + return true; } -s32 aki_socket_get_fd(struct aki_socket *sock) +s32 nn_socket_get_fd(struct nn_socket *sock) { return sock->internal_fd; } -ssize_t aki_socket_read(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_read(struct nn_socket *sock, void *buf, size_t size) { return recv(sock->fd, buf, (s32)size, 0); } -ssize_t aki_socket_write(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_write(struct nn_socket *sock, void *buf, size_t size) { return send(sock->fd, buf, (s32)size, 0); } -ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_sendto(struct nn_socket *sock, void *buf, size_t size) { (void)sock; (void)buf; @@ -122,7 +134,7 @@ ssize_t aki_socket_sendto(struct aki_socket *sock, void *buf, size_t size) return 0; } -ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size) +ssize_t nn_socket_recvfrom(struct nn_socket *sock, void *buf, size_t size) { (void)sock; (void)buf; @@ -130,12 +142,12 @@ ssize_t aki_socket_recvfrom(struct aki_socket *sock, void *buf, size_t size) return 0; } -void aki_socket_shutdown(struct aki_socket *sock) +void nn_socket_shutdown(struct nn_socket *sock) { if (shutdown(sock->fd, SD_BOTH) == SOCKET_ERROR) {} } -void aki_socket_close(struct aki_socket *sock) +void nn_socket_close(struct nn_socket *sock) { _close(sock->internal_fd); //closesocket(sock->fd); diff --git a/src/timer.c b/src/timer.c index 0057fd6..73902ea 100644 --- a/src/timer.c +++ b/src/timer.c @@ -3,13 +3,13 @@ static void timer_callback(struct ev_loop *loop, ev_timer *w, s32 revents) { (void)loop; - struct aki_timer *timer = (struct aki_timer *)w->data; + struct nn_timer *timer = (struct nn_timer *)w->data; (void)revents; if (!timer->disabled) timer->callback(timer->userdata, timer); } -void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop, - void (*callback)(void *, struct aki_timer *), void *userdata) +void nn_timer_init(struct nn_timer *timer, struct nn_event_loop *loop, + void (*callback)(void *, struct nn_timer *), void *userdata) { timer->loop = loop; timer->callback = callback; @@ -19,27 +19,27 @@ void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop, ev_init(&timer->timer, timer_callback); } -void aki_timer_set_repeat(struct aki_timer *timer, aki_os_tstamp repeat) +void nn_timer_set_repeat(struct nn_timer *timer, nn_os_tstamp repeat) { timer->timer.repeat = repeat; } -void aki_timer_again(struct aki_timer *timer) +void nn_timer_again(struct nn_timer *timer) { if (!timer->disabled) ev_timer_again(timer->loop->ev, &timer->timer); } -void aki_timer_disable(struct aki_timer *timer) +void nn_timer_disable(struct nn_timer *timer) { timer->disabled = true; } -void aki_timer_enable(struct aki_timer *timer) +void nn_timer_enable(struct nn_timer *timer) { timer->disabled = false; } -void aki_timer_stop(struct aki_timer *timer) +void nn_timer_stop(struct nn_timer *timer) { ev_timer_stop(timer->loop->ev, &timer->timer); } diff --git a/src/timer.h b/src/timer.h index 364c82f..15b467f 100644 --- a/src/timer.h +++ b/src/timer.h @@ -4,18 +4,18 @@ #include "loop.h" -struct aki_timer { +struct nn_timer { ev_timer timer; - struct aki_event_loop *loop; + struct nn_event_loop *loop; bool disabled; - void (*callback)(void *, struct aki_timer *); + void (*callback)(void *, struct nn_timer *); void *userdata; }; -void aki_timer_init(struct aki_timer *timer, struct aki_event_loop *loop, - void (*callback)(void *, struct aki_timer *), void *userdata); -void aki_timer_set_repeat(struct aki_timer *timer, aki_os_tstamp repeat); -void aki_timer_again(struct aki_timer *timer); -void aki_timer_disable(struct aki_timer *timer); -void aki_timer_enable(struct aki_timer *timer); -void aki_timer_stop(struct aki_timer *timer); +void nn_timer_init(struct nn_timer *timer, struct nn_event_loop *loop, + void (*callback)(void *, struct nn_timer *), void *userdata); +void nn_timer_set_repeat(struct nn_timer *timer, nn_os_tstamp repeat); +void nn_timer_again(struct nn_timer *timer); +void nn_timer_disable(struct nn_timer *timer); +void nn_timer_enable(struct nn_timer *timer); +void nn_timer_stop(struct nn_timer *timer); diff --git a/src/util/buffer.c b/src/util/buffer.c index 32a8230..1d343e3 100644 --- a/src/util/buffer.c +++ b/src/util/buffer.c @@ -1,58 +1,62 @@ +#include <al/macros.h> + #include "buffer.h" -#define BUFFER_INITIAL_SIZE 128 +#define NNWT_BUFFER_SIZE KB(1) +#define NNWT_BUFFER_GROW ((size_t)al_grow_step) -void aki_buffer_ensure_space(struct aki_buffer *buf, size_t size) +void nn_buffer_ensure_space(struct nn_buffer *buf, size_t size) { if (size > buf->alloc) { - if (size < BUFFER_INITIAL_SIZE) buf->alloc = BUFFER_INITIAL_SIZE; - else buf->alloc = al_next_power_of_two(size); - buf->data = buf->data ? al_realloc(buf->data, buf->alloc) : al_malloc(buf->alloc); + buf->alloc = size < NNWT_BUFFER_SIZE ? NNWT_BUFFER_SIZE : + (size + NNWT_BUFFER_GROW) & ~NNWT_BUFFER_GROW; + buf->data = (u8 *)(buf->data ? al_realloc(buf->data, buf->alloc) : al_malloc(buf->alloc)); } } -void aki_buffer_init(struct aki_buffer *buf) +void nn_buffer_init(struct nn_buffer *buf) { - buf->data = NULL; buf->size = 0; buf->alloc = 0; + buf->data = NULL; } -size_t aki_buffer_get_size(struct aki_buffer *buf) +size_t nn_buffer_get_size(struct nn_buffer *buf) { return buf->size; } -void aki_buffer_set_size(struct aki_buffer *buf, size_t size) +void nn_buffer_set_size(struct nn_buffer *buf, size_t size) { buf->size = size; } -void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size) +void nn_buffer_write(struct nn_buffer *buf, void *data, size_t index, size_t size) { size_t reach = index + size; - aki_buffer_ensure_space(buf, reach); - if (reach > buf->size) buf->size = reach; + nn_buffer_ensure_space(buf, reach); + if (reach > buf->size) + buf->size = reach; al_memcpy(&buf->data[index], data, size); } -void aki_buffer_append(struct aki_buffer *buf, void *data, size_t size) +void nn_buffer_append(struct nn_buffer *buf, void *data, size_t size) { - aki_buffer_write(buf, data, buf->size, size); + nn_buffer_write(buf, data, buf->size, size); } -u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index) +u8 *nn_buffer_get_ptr(struct nn_buffer *buf, size_t index) { return &buf->data[index]; } -void aki_buffer_read(struct aki_buffer *buf, void *ptr, size_t index, size_t size) +void nn_buffer_read(struct nn_buffer *buf, void *ptr, size_t index, size_t size) { al_assert(index + size <= buf->size); al_memcpy(ptr, &buf->data[index], size); } -void aki_buffer_free(struct aki_buffer *buf) +void nn_buffer_free(struct nn_buffer *buf) { if (buf->data && buf->alloc > 0) al_free(buf->data); } diff --git a/src/util/buffer.h b/src/util/buffer.h index 78d32d5..612aae4 100644 --- a/src/util/buffer.h +++ b/src/util/buffer.h @@ -3,18 +3,18 @@ #include <al/types.h> #include <al/lib.h> -struct aki_buffer { - u8 *data; +struct nn_buffer { size_t size; size_t alloc; + u8 *data; }; -void aki_buffer_init(struct aki_buffer *buf); -void aki_buffer_ensure_space(struct aki_buffer *buf, size_t size); -size_t aki_buffer_get_size(struct aki_buffer *buf); -void aki_buffer_set_size(struct aki_buffer *buf, size_t size); -void aki_buffer_write(struct aki_buffer *buf, void *data, size_t index, size_t size); -void aki_buffer_append(struct aki_buffer *buf, void *data, size_t size); -u8 *aki_buffer_get_ptr(struct aki_buffer *buf, size_t index); -void aki_buffer_read(struct aki_buffer *buf, void *ptr, size_t index, size_t size); -void aki_buffer_free(struct aki_buffer *buf); +void nn_buffer_init(struct nn_buffer *buf); +void nn_buffer_ensure_space(struct nn_buffer *buf, size_t size); +size_t nn_buffer_get_size(struct nn_buffer *buf); +void nn_buffer_set_size(struct nn_buffer *buf, size_t size); +void nn_buffer_write(struct nn_buffer *buf, void *data, size_t index, size_t size); +void nn_buffer_append(struct nn_buffer *buf, void *data, size_t size); +u8 *nn_buffer_get_ptr(struct nn_buffer *buf, size_t index); +void nn_buffer_read(struct nn_buffer *buf, void *ptr, size_t index, size_t size); +void nn_buffer_free(struct nn_buffer *buf); diff --git a/src/util/error.c b/src/util/error.c index c75696d..928e083 100644 --- a/src/util/error.c +++ b/src/util/error.c @@ -1,3 +1,3 @@ #include "error.h" -__thread char errorbuf[AKI_ERROR_STR_MAXLEN]; +__thread char errorbuf[NNWT_ERROR_STR_MAXLEN]; diff --git a/src/util/error.h b/src/util/error.h index 7f63aee..898119c 100644 --- a/src/util/error.h +++ b/src/util/error.h @@ -4,20 +4,20 @@ #include <errno.h> -#define AKI_ERROR_STR_MAXLEN 94 +#define NNWT_ERROR_STR_MAXLEN 94 -extern __thread char errorbuf[AKI_ERROR_STR_MAXLEN]; +extern __thread char errorbuf[NNWT_ERROR_STR_MAXLEN]; -static inline char *aki_strerror(s32 errnum) +static inline char *nn_strerror(s32 errnum) { #ifndef _WIN32 #if (_POSIX_C_SOURCE >= 200112L) && ! _GNU_SOURCE - if (strerror_r(errnum, errorbuf, AKI_ERROR_STR_MAXLEN) != 0) errorbuf[0] = '\0'; + if (strerror_r(errnum, errorbuf, NNWT_ERROR_STR_MAXLEN) != 0) errorbuf[0] = '\0'; #else - return strerror_r(errnum, errorbuf, AKI_ERROR_STR_MAXLEN); + return strerror_r(errnum, errorbuf, NNWT_ERROR_STR_MAXLEN); #endif #else - strerror_s(errorbuf, AKI_ERROR_STR_MAXLEN, errnum); + 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 0544d3c..30b44b6 100644 --- a/src/util/file/file.h +++ b/src/util/file/file.h @@ -3,7 +3,7 @@ #include <al/str.h> #include <al/lib.h> -#ifdef AKIYO_FILE_STDIO_COMPAT +#ifdef NAUNET_NEEDS_STDIO_ASSIST #include <stdio.h> #else #ifndef _WIN32 @@ -16,76 +16,73 @@ #endif enum { - AKI_FILE_READONLY = 1, - AKI_FILE_CREATE = 1 << 1, - AKI_FILE_LOCK = 1 << 2 + NNWT_FILE_READONLY = 1, + NNWT_FILE_CREATE = 1 << 1, + NNWT_FILE_LOCK = 1 << 2 }; enum { - AKI_ENTRY_FILE = 0, - AKI_ENTRY_DIR, - AKI_ENTRY_OTHER + NNWT_ENTRY_FILE = 0, + NNWT_ENTRY_DIR, + NNWT_ENTRY_OTHER }; -struct aki_file { -#ifdef AKIYO_FILE_STDIO_COMPAT +struct nn_file { +#ifdef NAUNET_NEEDS_STDIO_ASSIST FILE *file; #else #ifndef _WIN32 s32 fd; #else - // WINDOWS #endif bool locked; #endif size_t filesize; }; -struct aki_dir { -#ifdef AKIYO_FILE_STDIO_COMPAT +struct nn_dir { +#ifdef NAUNET_NEEDS_STDIO_ASSIST #else #ifndef _WIN32 DIR *dir; #else - // WINDOWS #endif #endif str path; }; -struct aki_dir_entry { +struct nn_dir_entry { u8 type; -#ifdef AKIYO_FILE_STDIO_COMPAT +#ifdef NAUNET_NEEDS_STDIO_ASSIST #else #ifndef _WIN32 struct dirent *entry; #else - // WINDOWS #endif #endif str name; str path; }; -bool aki_file_open(struct aki_file *file, str *path, s32 flags); -bool aki_file_exists(str *path); -bool aki_file_create(str *path); -bool aki_file_truncate(struct aki_file *file, off_t size); -off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence); -size_t aki_file_get_filesize(struct aki_file *file); -bool aki_file_write(struct aki_file *file, void *buf, size_t size); -bool aki_file_read(struct aki_file *file, void *buf, size_t size); -s32 aki_file_read_as_str(struct aki_file *file, str *out); -s32 aki_file_read_as_c_str(struct aki_file *file, char **out); -s32 aki_file_replace(struct aki_file *file, str *buf); -void *aki_file_mmap(struct aki_file *file); -void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map); -void aki_file_munmap(struct aki_file *file, void *map); -void aki_file_close(struct aki_file *file); +bool nn_file_open(struct nn_file *file, str *path, s32 flags); +bool nn_file_exists(str *path); +bool nn_file_create(str *path); +bool nn_file_truncate(struct nn_file *file, off_t size); +off_t nn_file_seek(struct nn_file *file, off_t offset, s32 whence); +size_t nn_file_get_filesize(struct nn_file *file); +bool nn_file_write(struct nn_file *file, void *buf, size_t size); +bool nn_file_read(struct nn_file *file, void *buf, size_t size); +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); +s32 nn_file_replace(struct nn_file *file, str *buf); +void *nn_file_mmap(struct nn_file *file); +void *nn_file_mremap(struct nn_file *file, size_t size, void *old_map); +void nn_file_munmap(struct nn_file *file, void *map); +void nn_file_close(struct nn_file *file); -bool aki_dir_open(struct aki_dir *dir, str *path); -void aki_dir_close(struct aki_dir *dir); -bool aki_dir_exists(str *path); -bool aki_dir_create(str *path); -bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry); -void aki_dir_entry_free(struct aki_dir_entry *entry); +bool nn_dir_open(struct nn_dir *dir, str *path); +void nn_dir_close(struct nn_dir *dir); +bool nn_dir_exists(str *path); +bool nn_dir_create(str *path); +bool nn_dir_read(struct nn_dir *dir, struct nn_dir_entry *entry); +void nn_dir_entry_free(struct nn_dir_entry *entry); diff --git a/src/util/file/file_linux.c b/src/util/file/file_linux.c index 977853c..9c6b1ce 100644 --- a/src/util/file/file_linux.c +++ b/src/util/file/file_linux.c @@ -9,7 +9,7 @@ static bool lock_file_internal(s32 fd, size_t size) { struct flock l = { .l_type = F_WRLCK, .l_whence = SEEK_SET, .l_start = 0, .l_len = size }; if (fcntl(fd, F_SETLKW, &l) == -1) { - al_log_error("file", "fcntl(F_SETLKW, F_WRLCK) failed (%s).", aki_strerror(errno)); + al_log_error("file", "fcntl(F_SETLKW, F_WRLCK) failed (%s).", nn_strerror(errno)); close(fd); return false; } @@ -20,41 +20,46 @@ static bool unlock_file_internal(s32 fd, size_t size) { struct flock l = { .l_type = F_ULOCK, .l_whence = SEEK_SET, .l_start = 0, .l_len = size }; if (fcntl(fd, F_SETLKW, &l) == -1) { - al_log_error("file", "fcntl(F_SETLKW, F_ULOCK) failed (%s).", aki_strerror(errno)); + al_log_error("file", "fcntl(F_SETLKW, F_ULOCK) failed (%s).", nn_strerror(errno)); return false; } return true; } -bool aki_file_open(struct aki_file *file, str *path, s32 flags) +bool nn_file_open(struct nn_file *file, str *path, s32 flags) { - al_assert(!((flags & AKI_FILE_READONLY) && (flags & AKI_FILE_LOCK))); + al_assert(!((flags & NNWT_FILE_READONLY) && (flags & NNWT_FILE_LOCK))); + s32 oflags = flags & NNWT_FILE_READONLY ? O_RDONLY : O_RDWR; + if (flags & NNWT_FILE_CREATE) + oflags |= O_CREAT; + char *c_str = al_str_to_c_str(path); - s32 oflags = (flags & AKI_FILE_READONLY) ? O_RDONLY : O_RDWR; - if (flags & AKI_FILE_CREATE) oflags |= O_CREAT; file->fd = open(c_str, oflags, 0644); al_free(c_str); if (file->fd == -1) { - al_log_error("file", "open(%.*s) failed (%d: %s).", AL_STR_PRINTF(path), errno, aki_strerror(errno)); + al_log_error("file", "open(%.*s) failed (%d: %s).", AL_STR_PRINTF(path), errno, nn_strerror(errno)); return false; } + struct stat sb; s32 res = fstat(file->fd, &sb); if (res == -1) { - al_log_error("file", "fstat() failed (%s).", aki_strerror(errno)); + al_log_error("file", "fstat() failed (%s).", nn_strerror(errno)); close(file->fd); return false; } file->filesize = sb.st_size; - if (flags & AKI_FILE_LOCK) { - file->locked = lock_file_internal(file->fd, file->filesize); - } else { - file->locked = false; + + file->locked = flags & NNWT_FILE_LOCK; + if (file->locked && !lock_file_internal(file->fd, file->filesize)) { + close(file->fd); + return false; } + return true; } -bool aki_file_exists(str *path) +bool nn_file_exists(str *path) { char *c_str = al_str_to_c_str(path); s32 ret = access(c_str, F_OK); @@ -62,37 +67,36 @@ bool aki_file_exists(str *path) return ret == 0; } -bool aki_file_create(str *path) +bool nn_file_create(str *path) { - struct aki_file file; - if (aki_file_open(&file, path, AKI_FILE_CREATE)) { - aki_file_close(&file); + struct nn_file file; + if (nn_file_open(&file, path, NNWT_FILE_CREATE)) { + nn_file_close(&file); return true; } return false; } -off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence) +off_t nn_file_seek(struct nn_file *file, off_t offset, s32 whence) { off_t res = lseek(file->fd, offset, whence); - if (res == (off_t)-1) { - al_log_error("file", "lseek(%jd, %d) failed (%s).", (intmax_t)offset, whence, aki_strerror(errno)); - } + if (res == (off_t)-1) + al_log_error("file", "lseek(%jd, %d) failed (%s).", (intmax_t)offset, whence, nn_strerror(errno)); return res; } -bool aki_file_truncate(struct aki_file *file, off_t size) +bool nn_file_truncate(struct nn_file *file, off_t size) { s32 ret = ftruncate(file->fd, size); if (ret != 0) { - al_log_error("file", "ftruncate(%jd) failed (%s).", (intmax_t)size, aki_strerror(errno)); + al_log_error("file", "ftruncate(%jd) failed (%s).", (intmax_t)size, nn_strerror(errno)); return false; } file->filesize = size; return true; } -size_t aki_file_get_filesize(struct aki_file *file) +size_t nn_file_get_filesize(struct nn_file *file) { return file->filesize; } @@ -100,140 +104,136 @@ size_t aki_file_get_filesize(struct aki_file *file) #define file_io_loop(call, fail_case) \ size_t bc = 0; \ ssize_t res; \ - do { \ + for (;;) { \ res = call(file->fd, buf, size - bc); \ if (fail_case) { \ - al_log_error("file", ""#call"() failed (%s).", aki_strerror(errno)); \ + al_log_error("file", ""#call"() failed (%s).", nn_strerror(errno)); \ return false; \ } \ - bc += res; \ - } while (bc < size); \ + if ((bc += res) >= size) break; \ + } \ return true -bool aki_file_write(struct aki_file *file, void *buf, size_t size) +bool nn_file_write(struct nn_file *file, void *buf, size_t size) { file_io_loop(write, (res == -1)); } -bool aki_file_read(struct aki_file *file, void *buf, size_t size) +bool nn_file_read(struct nn_file *file, void *buf, size_t size) { file_io_loop(read, (res == -1)); } -void *aki_file_mmap(struct aki_file *file) +void *nn_file_mmap(struct nn_file *file) { void *map = mmap(NULL, file->filesize, PROT_READ | PROT_WRITE, MAP_SHARED, file->fd, 0L); if (map == MAP_FAILED) { - al_log_error("file", "mmap() failed (%s).", aki_strerror(errno)); + al_log_error("file", "mmap() failed (%s).", nn_strerror(errno)); return NULL; } return map; } -void *aki_file_mremap(struct aki_file *file, size_t size, void *old_map) +void *nn_file_mremap(struct nn_file *file, size_t size, void *old_map) { void *map = mremap(old_map, file->filesize, size, MREMAP_MAYMOVE); if (map == MAP_FAILED) { - al_log_error("file", "mremap(%p, %zu, %zu) failed (%s).", old_map, - file->filesize, size, aki_strerror(errno)); + al_log_error("file", "mremap(%p, %zu, %zu) failed (%s).", + old_map, file->filesize, size, nn_strerror(errno)); return NULL; } return map; } -void aki_file_munmap(struct aki_file *file, void *map) +void nn_file_munmap(struct nn_file *file, void *map) { s32 ret = munmap(map, file->filesize); - if (ret != 0) { - al_log_error("file", "munmap(%p, %zu) failed (%s).", map, file->filesize, aki_strerror(errno)); - } + if (ret != 0) + al_log_error("file", "munmap(%p, %zu) failed (%s).", map, file->filesize, nn_strerror(errno)); } -void aki_file_close(struct aki_file *file) +void nn_file_close(struct nn_file *file) { - if (file->locked) { - unlock_file_internal(file->fd, file->filesize); - } + if (file->locked) unlock_file_internal(file->fd, file->filesize); close(file->fd); } -bool aki_dir_open(struct aki_dir *dir, str *path) +bool nn_dir_open(struct nn_dir *dir, str *path) { char *c_str = al_str_to_c_str(path); dir->dir = opendir(c_str); al_free(c_str); if (!dir->dir) { - al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno)); + al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), nn_strerror(errno)); return false; } al_str_clone(&dir->path, path); return true; } -void aki_dir_close(struct aki_dir *dir) +void nn_dir_close(struct nn_dir *dir) { al_str_free(&dir->path); closedir(dir->dir); } -bool aki_dir_exists(str *path) +bool nn_dir_exists(str *path) { - struct aki_dir dir; + struct nn_dir dir; char *c_str = al_str_to_c_str(path); dir.dir = opendir(c_str); al_free(c_str); if (!dir.dir) { - if (errno != ENOENT) { - al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno)); - } + if (errno != ENOENT) + al_log_error("file", "opendir(%.*s) failed (%s).", AL_STR_PRINTF(path), nn_strerror(errno)); return false; } closedir(dir.dir); return true; } -bool aki_dir_create(str *path) +bool nn_dir_create(str *path) { char *c_str = al_str_to_c_str(path); s32 ret = mkdir(c_str, 0755); al_free(c_str); if (ret == -1) { - al_log_error("file", "mkdir(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno)); + al_log_error("file", "mkdir(%.*s) failed (%s).", AL_STR_PRINTF(path), nn_strerror(errno)); return false; } return true; } -bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry) +bool nn_dir_read(struct nn_dir *dir, struct nn_dir_entry *entry) { - errno = 0; - entry->entry = readdir(dir->dir); - if (!entry->entry) { - if (errno) { - al_log_error("file", "readdir() failed (%s).", aki_strerror(errno)); - } + if (!(entry->entry = readdir(dir->dir))) { + if (errno) + al_log_error("file", "readdir() failed (%s).", nn_strerror(errno)); return false; } + switch (entry->entry->d_type) { - case DT_REG: case DT_UNKNOWN: - entry->type = AKI_ENTRY_FILE; + entry->type = NNWT_ENTRY_FILE; break; + case DT_REG: case DT_DIR: - entry->type = AKI_ENTRY_DIR; + entry->type = NNWT_ENTRY_DIR; break; default: - entry->type = AKI_ENTRY_OTHER; + entry->type = NNWT_ENTRY_OTHER; break; } + al_str_clone(&entry->name, al_str_cr(entry->entry->d_name)); al_str_clone(&entry->path, &dir->path); al_str_cat(&entry->path, al_str_c("/")); al_str_cat(&entry->path, &entry->name); + return true; } -void aki_dir_entry_free(struct aki_dir_entry *entry) +void nn_dir_entry_free(struct nn_dir_entry *entry) { al_str_free(&entry->name); al_str_free(&entry->path); diff --git a/src/util/file/file_stdio.c b/src/util/file/file_stdio.c index fa1e4ff..8f6d7ae 100644 --- a/src/util/file/file_stdio.c +++ b/src/util/file/file_stdio.c @@ -4,10 +4,10 @@ #include "file.h" -bool aki_file_open(struct aki_file *file, str *path, s32 flags) +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); - const char *mode = (flags & AKI_FILE_CREATE) ? "wb+" : "rb+"; #ifndef _WIN32 file->file = fopen(c_str, mode); #else @@ -19,82 +19,84 @@ bool aki_file_open(struct aki_file *file, str *path, s32 flags) #else if (ret != 0) { #endif - al_log_error("file_stdio", "fopen(%.*s) failed (%s).", AL_STR_PRINTF(path), aki_strerror(errno)); + al_log_error("file_stdio", "fopen(%.*s) failed (%s).", AL_STR_PRINTF(path), nn_strerror(errno)); return false; } - off_t pos = aki_file_seek(file, 0, SEEK_END); + + off_t pos = nn_file_seek(file, 0, SEEK_END); if (pos == -1) { - aki_file_close(file); + nn_file_close(file); return false; } file->filesize = pos; rewind(file->file); + return true; } -bool aki_file_exists(str *path) +bool nn_file_exists(str *path) { (void)path; return false; } -off_t aki_file_seek(struct aki_file *file, off_t offset, s32 whence) +off_t nn_file_seek(struct nn_file *file, off_t offset, s32 whence) { if (fseek(file->file, offset, whence) != 0) { - al_log_error("file_stdio", "fseek(%jd, %d) failed (%s).", (intmax_t)offset, - whence, aki_strerror(errno)); + al_log_error("file_stdio", "fseek(%jd, %d) failed (%s).", + (intmax_t)offset, whence, nn_strerror(errno)); return -1; } return ftell(file->file); } -bool aki_file_truncate(struct aki_file *file, off_t size) +bool nn_file_truncate(struct nn_file *file, off_t size) { (void)file; (void)size; return true; } -size_t aki_file_get_filesize(struct aki_file *file) +size_t nn_file_get_filesize(struct nn_file *file) { return file->filesize; } -bool aki_file_write(struct aki_file *file, void *buf, size_t size) +bool nn_file_write(struct nn_file *file, void *buf, size_t size) { size_t ret = fwrite(buf, size, 1, file->file); if (ret == 0) { - al_log_error("file_stdio", "fwrite(%zu) failed (%s).", size, aki_strerror(errno)); + al_log_error("file_stdio", "fwrite(%zu) failed (%s).", size, nn_strerror(errno)); return false; } return true; } -bool aki_file_read(struct aki_file *file, void *buf, size_t size) +bool nn_file_read(struct nn_file *file, void *buf, size_t size) { size_t ret = fread(buf, size, 1, file->file); if (ret == 0) { - al_log_error("file_stdio", "fread(%zu) failed (%s).", size, aki_strerror(errno)); + al_log_error("file_stdio", "fread(%zu) failed (%s).", size, nn_strerror(errno)); return false; } return true; } -void *aki_file_mmap(struct aki_file *file) +void *nn_file_mmap(struct nn_file *file) { (void)file; al_assert(false); return NULL; } -void aki_file_close(struct aki_file *file) +void nn_file_close(struct nn_file *file) { fclose(file->file); } -bool aki_dir_open(struct aki_dir *dir, str *path) { (void)dir; (void)path; return false; } -void aki_dir_close(struct aki_dir *dir) { (void)dir; } -bool aki_dir_exists(str *path) { (void)path; return false; } -bool aki_dir_create(str *path) { (void)path; return false; } -bool aki_dir_read(struct aki_dir *dir, struct aki_dir_entry *entry) { (void)dir; (void)entry; return false; } -void aki_dir_entry_free(struct aki_dir_entry *entry) { (void)entry; } +bool nn_dir_open(struct nn_dir *dir, str *path) { (void)dir; (void)path; return false; } +void nn_dir_close(struct nn_dir *dir) { (void)dir; } +bool nn_dir_exists(str *path) { (void)path; return false; } +bool nn_dir_create(str *path) { (void)path; return false; } +bool nn_dir_read(struct nn_dir *dir, struct nn_dir_entry *entry) { (void)dir; (void)entry; return false; } +void nn_dir_entry_free(struct nn_dir_entry *entry) { (void)entry; } diff --git a/src/util/file/util.c b/src/util/file/util.c index fe7785f..50e2c4a 100644 --- a/src/util/file/util.c +++ b/src/util/file/util.c @@ -1,10 +1,10 @@ #include "file.h" -s32 aki_file_read_as_str(struct aki_file *file, str *out) +s32 nn_file_read_as_str(struct nn_file *file, str *out) { size_t size = file->filesize; al_str_sized(out, size); - if (!aki_file_read(file, (void *)out->data, size)) { + if (!nn_file_read(file, (void *)out->data, size)) { al_str_free(out); return -1; } @@ -12,11 +12,11 @@ s32 aki_file_read_as_str(struct aki_file *file, str *out) return size; } -s32 aki_file_read_as_c_str(struct aki_file *file, char **out) +s32 nn_file_read_as_c_str(struct nn_file *file, char **out) { size_t size = file->filesize; *out = al_malloc(size + 1); - if (!aki_file_read(file, (void *)*out, size)) { + if (!nn_file_read(file, (void *)*out, size)) { al_free(*out); return -1; } @@ -24,11 +24,11 @@ s32 aki_file_read_as_c_str(struct aki_file *file, char **out) return size; } -s32 aki_file_replace(struct aki_file *file, str *buf) +s32 nn_file_replace(struct nn_file *file, str *buf) { - if (aki_file_seek(file, 0, SEEK_SET) == (off_t)-1 || - !aki_file_write(file, buf->data, buf->len) || - !aki_file_truncate(file, buf->len)) { + if (nn_file_seek(file, 0, SEEK_SET) == (off_t)-1 || + !nn_file_write(file, buf->data, buf->len) || + !nn_file_truncate(file, buf->len)) { return -1; } return buf->len; diff --git a/src/util/packet.c b/src/util/packet.c index 2658d38..900016c 100644 --- a/src/util/packet.c +++ b/src/util/packet.c @@ -1,44 +1,44 @@ #include "packet.h" -struct aki_packet *aki_packet_create(void) +struct nn_packet *nn_packet_create(void) { - struct aki_packet *packet = (struct aki_packet *)al_malloc(sizeof(struct aki_packet)); - aki_buffer_init(&packet->buffer); - aki_packet_reset(packet); + struct nn_packet *packet = (struct nn_packet *)al_malloc(sizeof(struct nn_packet)); + nn_buffer_init(&packet->buffer); + nn_packet_reset(packet); return packet; } -struct aki_packet *aki_packet_clone(struct aki_packet *packet) +struct nn_packet *nn_packet_clone(struct nn_packet *packet) { - struct aki_packet *c = aki_packet_create(); - aki_buffer_ensure_space(&c->buffer, packet->buffer.size); - al_memcpy(aki_buffer_get_ptr(&c->buffer, 0), aki_buffer_get_ptr(&packet->buffer, 0), packet->buffer.size); + struct nn_packet *c = nn_packet_create(); + nn_buffer_ensure_space(&c->buffer, packet->buffer.size); + 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; return c; } -void aki_packet_reset(struct aki_packet *packet) +void nn_packet_reset(struct nn_packet *packet) { - aki_buffer_ensure_space(&packet->buffer, AKI_PACKET_HEADER_LENGTH); - packet->rindex = AKI_PACKET_HEADER_LENGTH; - packet->windex = AKI_PACKET_HEADER_LENGTH; + nn_buffer_ensure_space(&packet->buffer, NNWT_PACKET_HEADER_LENGTH); + packet->rindex = NNWT_PACKET_HEADER_LENGTH; + packet->windex = NNWT_PACKET_HEADER_LENGTH; } -void aki_packet_write_size(struct aki_packet *packet) +void nn_packet_write_size(struct nn_packet *packet) { - *((u32 *)aki_buffer_get_ptr(&packet->buffer, 0)) = packet->windex; + *((u32 *)nn_buffer_get_ptr(&packet->buffer, 0)) = packet->windex; } -u32 aki_packet_get_size(struct aki_packet *packet) +u32 nn_packet_get_size(struct nn_packet *packet) { - return *((u32 *)aki_buffer_get_ptr(&packet->buffer, 0)); + return *((u32 *)nn_buffer_get_ptr(&packet->buffer, 0)); } #define DEFINE_PACKET_WRITE_FUNC(type) \ - void aki_packet_write_##type(struct aki_packet *packet, type v) \ + void nn_packet_write_##type(struct nn_packet *packet, type v) \ { \ - AKI_PACKET_WRITE_TYPE(packet, type, v); \ + NNWT_PACKET_WRITE_TYPE(packet, type, v); \ } DEFINE_PACKET_WRITE_FUNC(bool) @@ -53,25 +53,25 @@ DEFINE_PACKET_WRITE_FUNC(s64) DEFINE_PACKET_WRITE_FUNC(f32) DEFINE_PACKET_WRITE_FUNC(f64) -void aki_packet_write_str(struct aki_packet *packet, str *s) +void nn_packet_write_str(struct nn_packet *packet, str *s) { - AKI_PACKET_WRITE_TYPE(packet, u32, s->len); - AKI_PACKET_WRITE_DATA(packet, s->data, s->len); + NNWT_PACKET_WRITE_TYPE(packet, u32, s->len); + NNWT_PACKET_WRITE_DATA(packet, s->data, s->len); } -void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf) +void nn_packet_write_buffer(struct nn_packet *packet, struct nn_buffer *buf) { u64 size = (u64)buf->size; - AKI_PACKET_WRITE_TYPE(packet, u64, size); - AKI_PACKET_WRITE_DATA(packet, aki_buffer_get_ptr(buf, 0), buf->size); + NNWT_PACKET_WRITE_TYPE(packet, u64, size); + NNWT_PACKET_WRITE_DATA(packet, nn_buffer_get_ptr(buf, 0), buf->size); } #define DEFINE_PACKET_READ_FUNC(type) \ - type aki_packet_read_##type(struct aki_packet *packet) \ + type nn_packet_read_##type(struct nn_packet *packet) \ { \ - type r; \ - AKI_PACKET_READ_TYPE(packet, type, r); \ - return r; \ + type r; \ + NNWT_PACKET_READ_TYPE(packet, type, r); \ + return r; \ } DEFINE_PACKET_READ_FUNC(bool) @@ -86,38 +86,38 @@ DEFINE_PACKET_READ_FUNC(s64) DEFINE_PACKET_READ_FUNC(f32) DEFINE_PACKET_READ_FUNC(f64) -void aki_packet_read_str(struct aki_packet *packet, str *s) +void nn_packet_read_str(struct nn_packet *packet, str *s) { - AKI_PACKET_READ_TYPE(packet, u32, s->len); - AKI_PACKET_READ_DATA(packet, s->len, s->data); + NNWT_PACKET_READ_TYPE(packet, u32, s->len); + NNWT_PACKET_READ_DATA(packet, s->len, s->data); s->alloc = 0; } -void aki_packet_read_buffer(struct aki_packet *packet, struct aki_buffer *buf) +void nn_packet_read_buffer(struct nn_packet *packet, struct nn_buffer *buf) { u64 size; - AKI_PACKET_READ_TYPE(packet, u64, size); + NNWT_PACKET_READ_TYPE(packet, u64, size); al_assert((sizeof(size_t) > 4 || size <= INT32_MAX)); buf->size = (size_t)size; - AKI_PACKET_READ_DATA(packet, buf->size, buf->data); + NNWT_PACKET_READ_DATA(packet, buf->size, buf->data); buf->alloc = 0; } -void aki_packet_write_wstr(struct aki_packet *packet, wstr *w) +void nn_packet_write_wstr(struct nn_packet *packet, wstr *w) { - AKI_PACKET_WRITE_TYPE(packet, u32, w->len); - AKI_PACKET_WRITE_DATA(packet, w->data, w->len * sizeof(wchar_t)); + NNWT_PACKET_WRITE_TYPE(packet, u32, w->len); + NNWT_PACKET_WRITE_DATA(packet, w->data, w->len * sizeof(wchar_t)); } -void aki_packet_read_wstr(struct aki_packet *packet, wstr *w) +void nn_packet_read_wstr(struct nn_packet *packet, wstr *w) { - AKI_PACKET_READ_TYPE(packet, u32, w->len); - AKI_PACKET_READ_DATA(packet, w->len * sizeof(wchar_t), w->data); + NNWT_PACKET_READ_TYPE(packet, u32, w->len); + NNWT_PACKET_READ_DATA(packet, w->len * sizeof(wchar_t), w->data); w->alloc = 0; } -void aki_packet_free(struct aki_packet *packet) +void nn_packet_free(struct nn_packet *packet) { - aki_buffer_free(&packet->buffer); + nn_buffer_free(&packet->buffer); al_free(packet); } diff --git a/src/util/packet.h b/src/util/packet.h index 72543a6..41f0a77 100644 --- a/src/util/packet.h +++ b/src/util/packet.h @@ -7,69 +7,69 @@ #include "../util/buffer.h" -#define AKI_PACKET_HEADER_LENGTH (u32)(sizeof(u32)) +#define NNWT_PACKET_HEADER_LENGTH (u32)(sizeof(u32)) -struct aki_packet { - struct aki_buffer buffer; +struct nn_packet { + struct nn_buffer buffer; u32 rindex; u32 windex; }; -#define AKI_PACKET_GET_ID(p) \ - *((u32 *)aki_buffer_get_ptr(&p->buffer, AKI_PACKET_HEADER_LENGTH + sizeof(s8))) +#define NNWT_PACKET_GET_ID(p) \ + *((u32 *)nn_buffer_get_ptr(&p->buffer, NNWT_PACKET_HEADER_LENGTH + sizeof(s8))) -#define AKI_PACKET_WRITE_TYPE(p, type, v) \ - aki_buffer_write(&(p)->buffer, &(v), (p)->windex, sizeof(type)); \ +#define NNWT_PACKET_WRITE_TYPE(p, type, v) \ + nn_buffer_write(&(p)->buffer, &(v), (p)->windex, sizeof(type)); \ (p)->windex += (u32)sizeof(type) -#define AKI_PACKET_WRITE_DATA(p, data, len) \ - aki_buffer_write(&(p)->buffer, (data), (p)->windex, (len)); \ +#define NNWT_PACKET_WRITE_DATA(p, data, len) \ + nn_buffer_write(&(p)->buffer, (data), (p)->windex, (len)); \ (p)->windex += (u32)len -#define AKI_PACKET_READ_TYPE(p, type, r) \ - r = *((type *)aki_buffer_get_ptr(&(p)->buffer, (p)->rindex)); \ +#define NNWT_PACKET_READ_TYPE(p, type, r) \ + r = *((type *)nn_buffer_get_ptr(&(p)->buffer, (p)->rindex)); \ (p)->rindex += (u32)sizeof(type); -#define AKI_PACKET_READ_DATA(p, len, r) \ - r = (void *)aki_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; -struct aki_packet *aki_packet_create(void); -struct aki_packet *aki_packet_clone(struct aki_packet *packet); -void aki_packet_reset(struct aki_packet *packet); +struct nn_packet *nn_packet_create(void); +struct nn_packet *nn_packet_clone(struct nn_packet *packet); +void nn_packet_reset(struct nn_packet *packet); -void aki_packet_write_size(struct aki_packet *packet); -u32 aki_packet_get_size(struct aki_packet *packet); +void nn_packet_write_size(struct nn_packet *packet); +u32 nn_packet_get_size(struct nn_packet *packet); -void aki_packet_write_bool(struct aki_packet *packet, bool v); -void aki_packet_write_s8(struct aki_packet *packet, s8 v); -void aki_packet_write_u8(struct aki_packet *packet, u8 v); -void aki_packet_write_s16(struct aki_packet *packet, s16 v); -void aki_packet_write_u16(struct aki_packet *packet, u16 v); -void aki_packet_write_s32(struct aki_packet *packet, s32 v); -void aki_packet_write_u32(struct aki_packet *packet, u32 v); -void aki_packet_write_s64(struct aki_packet *packet, s64 v); -void aki_packet_write_u64(struct aki_packet *packet, u64 v); -void aki_packet_write_f32(struct aki_packet *packet, f32 v); -void aki_packet_write_f64(struct aki_packet *packet, f64 v); -void aki_packet_write_str(struct aki_packet *packet, str *s); -void aki_packet_write_buffer(struct aki_packet *packet, struct aki_buffer *buf); +void nn_packet_write_bool(struct nn_packet *packet, bool v); +void nn_packet_write_s8(struct nn_packet *packet, s8 v); +void nn_packet_write_u8(struct nn_packet *packet, u8 v); +void nn_packet_write_s16(struct nn_packet *packet, s16 v); +void nn_packet_write_u16(struct nn_packet *packet, u16 v); +void nn_packet_write_s32(struct nn_packet *packet, s32 v); +void nn_packet_write_u32(struct nn_packet *packet, u32 v); +void nn_packet_write_s64(struct nn_packet *packet, s64 v); +void nn_packet_write_u64(struct nn_packet *packet, u64 v); +void nn_packet_write_f32(struct nn_packet *packet, f32 v); +void nn_packet_write_f64(struct nn_packet *packet, f64 v); +void nn_packet_write_str(struct nn_packet *packet, str *s); +void nn_packet_write_buffer(struct nn_packet *packet, struct nn_buffer *buf); -bool aki_packet_read_bool(struct aki_packet *packet); -s8 aki_packet_read_s8(struct aki_packet *packet); -u8 aki_packet_read_u8(struct aki_packet *packet); -s16 aki_packet_read_s16(struct aki_packet *packet); -u16 aki_packet_read_u16(struct aki_packet *packet); -s32 aki_packet_read_s32(struct aki_packet *packet); -u32 aki_packet_read_u32(struct aki_packet *packet); -s64 aki_packet_read_s64(struct aki_packet *packet); -u64 aki_packet_read_u64(struct aki_packet *packet); -f32 aki_packet_read_f32(struct aki_packet *packet); -f64 aki_packet_read_f64(struct aki_packet *packet); -void aki_packet_read_str(struct aki_packet *packet, str *s); -void aki_packet_read_buffer(struct aki_packet *packet, struct aki_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); +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); +u32 nn_packet_read_u32(struct nn_packet *packet); +s64 nn_packet_read_s64(struct nn_packet *packet); +u64 nn_packet_read_u64(struct nn_packet *packet); +f32 nn_packet_read_f32(struct nn_packet *packet); +f64 nn_packet_read_f64(struct nn_packet *packet); +void nn_packet_read_str(struct nn_packet *packet, str *s); +void nn_packet_read_buffer(struct nn_packet *packet, struct nn_buffer *buf); -void aki_packet_write_wstr(struct aki_packet *packet, wstr *w); -void aki_packet_read_wstr(struct aki_packet *packet, wstr *w); +void nn_packet_write_wstr(struct nn_packet *packet, wstr *w); +void nn_packet_read_wstr(struct nn_packet *packet, wstr *w); -void aki_packet_free(struct aki_packet *packet); +void nn_packet_free(struct nn_packet *packet); diff --git a/src/util/thread/thread.h b/src/util/thread/thread.h index 5975c84..921de23 100644 --- a/src/util/thread/thread.h +++ b/src/util/thread/thread.h @@ -8,14 +8,14 @@ #include "../../winwrap.h" #endif -#ifdef AKIYO_HAS_THREAD_CANCEL +#ifdef NAUNET_HAS_THREAD_CANCEL enum { - AKI_THREAD_CANCEL_DEFERRED = PTHREAD_CANCEL_DEFERRED, - AKI_THREAD_CANCEL_ASYNCHRONOUS = PTHREAD_CANCEL_ASYNCHRONOUS + NNWT_THREAD_CANCEL_DEFERRED = PTHREAD_CANCEL_DEFERRED, + NNWT_THREAD_CANCEL_ASYNCHRONOUS = PTHREAD_CANCEL_ASYNCHRONOUS }; #endif -struct aki_thread { +struct nn_thread { #ifndef _WIN32 pthread_t thread; #else @@ -23,7 +23,7 @@ struct aki_thread { #endif }; -struct aki_mutex { +struct nn_mutex { #ifndef _WIN32 pthread_mutex_t mutex; #else @@ -31,7 +31,7 @@ struct aki_mutex { #endif }; -struct aki_cond { +struct nn_cond { #ifndef _WIN32 pthread_cond_t cond; #else @@ -40,37 +40,37 @@ struct aki_cond { bool condition; }; -typedef f64 aki_os_tstamp; +typedef f64 nn_os_tstamp; -#define AKI_TS_FROM_USEC(usec) ((usec) * 1e-6) +#define NNWT_TS_FROM_USEC(usec) ((usec) * 1e-6) #ifndef _WIN32 -#define AKI_THREADCALL -typedef void * aki_thread_result; +#define NNWT_THREADCALL +typedef void * nn_thread_result; #else -#define AKI_THREADCALL __stdcall -typedef unsigned long aki_thread_result; +#define NNWT_THREADCALL __stdcall +typedef unsigned long nn_thread_result; #endif -typedef aki_thread_result (AKI_THREADCALL * aki_thread_func)(void *); +typedef nn_thread_result (NNWT_THREADCALL * nn_thread_func)(void *); -void aki_thread_sleep(aki_os_tstamp delay); +void nn_thread_sleep(nn_os_tstamp delay); -void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata); -void aki_thread_join(struct aki_thread *thread); -#ifdef AKIYO_HAS_THREAD_CANCEL -void aki_thread_cancel(struct aki_thread *thread); -void aki_thread_setcanceltype(s32 type); +void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userdata); +void nn_thread_join(struct nn_thread *thread); +#ifdef NAUNET_HAS_THREAD_CANCEL +void nn_thread_cancel(struct nn_thread *thread); +void nn_thread_setcanceltype(s32 type); #endif -//void aki_thread_detach(struct aki_thread *thread); +//void nn_thread_detach(struct nn_thread *thread); -void aki_mutex_init(struct aki_mutex *mutex); -void aki_mutex_lock(struct aki_mutex *mutex); -void aki_mutex_unlock(struct aki_mutex *mutex); -void aki_mutex_destroy(struct aki_mutex *mutex); +void nn_mutex_init(struct nn_mutex *mutex); +void nn_mutex_lock(struct nn_mutex *mutex); +void nn_mutex_unlock(struct nn_mutex *mutex); +void nn_mutex_destroy(struct nn_mutex *mutex); -void aki_cond_init(struct aki_cond *cond); -void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex); -bool aki_cond_is_waiting(struct aki_cond *cond); -void aki_cond_signal(struct aki_cond *cond); -void aki_cond_destroy(struct aki_cond *cond); +void nn_cond_init(struct nn_cond *cond); +void nn_cond_wait(struct nn_cond *cond, struct nn_mutex *mutex); +bool nn_cond_is_waiting(struct nn_cond *cond); +void nn_cond_signal(struct nn_cond *cond); +void nn_cond_destroy(struct nn_cond *cond); diff --git a/src/util/thread/thread_linux.c b/src/util/thread/thread_linux.c index 26c3e18..bb5bcbc 100644 --- a/src/util/thread/thread_linux.c +++ b/src/util/thread/thread_linux.c @@ -4,7 +4,7 @@ #include "thread.h" -void aki_thread_sleep(aki_os_tstamp delay) +void nn_thread_sleep(nn_os_tstamp delay) { struct timespec tv; tv.tv_sec = (s64)delay; @@ -12,81 +12,80 @@ void aki_thread_sleep(aki_os_tstamp delay) while (nanosleep(&tv, &tv) == -1 && errno == EINTR) {} } -void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata) +void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userdata) { pthread_create(&thread->thread, NULL, func, userdata); } -void aki_thread_join(struct aki_thread *thread) +void nn_thread_join(struct nn_thread *thread) { pthread_join(thread->thread, NULL); } -#ifdef AKIYO_HAS_THREAD_CANCEL -void aki_thread_cancel(struct aki_thread *thread) +#ifdef NAUNET_HAS_THREAD_CANCEL +void nn_thread_cancel(struct nn_thread *thread) { pthread_cancel(thread->thread); } -void aki_thread_setcanceltype(s32 type) +void nn_thread_setcanceltype(s32 type) { pthread_setcanceltype(type, NULL); } #endif /* -void aki_thread_detach(struct aki_thread *thread) +void nn_thread_detach(struct nn_thread *thread) { pthread_detach(thread->thread); } */ -void aki_mutex_init(struct aki_mutex *mutex) +void nn_mutex_init(struct nn_mutex *mutex) { pthread_mutex_init(&mutex->mutex, NULL); } -void aki_mutex_lock(struct aki_mutex *mutex) +void nn_mutex_lock(struct nn_mutex *mutex) { pthread_mutex_lock(&mutex->mutex); } -void aki_mutex_unlock(struct aki_mutex *mutex) +void nn_mutex_unlock(struct nn_mutex *mutex) { pthread_mutex_unlock(&mutex->mutex); } -void aki_mutex_destroy(struct aki_mutex *mutex) +void nn_mutex_destroy(struct nn_mutex *mutex) { pthread_mutex_destroy(&mutex->mutex); } -void aki_cond_init(struct aki_cond *cond) +void nn_cond_init(struct nn_cond *cond) { cond->condition = true; pthread_cond_init(&cond->cond, NULL); } -void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex) +void nn_cond_wait(struct nn_cond *cond, struct nn_mutex *mutex) { cond->condition = false; - while (!cond->condition) { + while (!cond->condition) pthread_cond_wait(&cond->cond, &mutex->mutex); - } } -bool aki_cond_is_waiting(struct aki_cond *cond) +bool nn_cond_is_waiting(struct nn_cond *cond) { return !cond->condition; } -void aki_cond_signal(struct aki_cond *cond) +void nn_cond_signal(struct nn_cond *cond) { cond->condition = true; pthread_cond_signal(&cond->cond); } -void aki_cond_destroy(struct aki_cond *cond) +void nn_cond_destroy(struct nn_cond *cond) { pthread_cond_destroy(&cond->cond); } diff --git a/src/util/thread/thread_windows.c b/src/util/thread/thread_windows.c index c7fdd62..ded3b6c 100644 --- a/src/util/thread/thread_windows.c +++ b/src/util/thread/thread_windows.c @@ -2,86 +2,85 @@ NTSTATUS(__stdcall *NtDelayExecution)(BOOL Alertable, PLARGE_INTEGER DelayInterval); -void aki_thread_sleep(aki_os_tstamp delay) +void nn_thread_sleep(nn_os_tstamp delay) { NtDelayExecution(false, &((LARGE_INTEGER){ .QuadPart = -(s64)(delay * 1e7L) })); } -void aki_thread_create(struct aki_thread *thread, aki_thread_func func, void *userdata) +void nn_thread_create(struct nn_thread *thread, nn_thread_func func, void *userdata) { thread->thread = CreateThread(NULL, 0, func, userdata, 0, NULL); } -void aki_thread_join(struct aki_thread *thread) +void nn_thread_join(struct nn_thread *thread) { WaitForSingleObject(thread->thread, INFINITE); CloseHandle(thread->thread); } -#ifdef AKIYO_HAS_THREAD_CANCEL -void aki_thread_cancel(struct aki_thread *thread) +#ifdef NAUNET_HAS_THREAD_CANCEL +void nn_thread_cancel(struct nn_thread *thread) { (void)thread; } -void aki_thread_setcanceltype(s32 type) +void nn_thread_setcanceltype(s32 type) { (void)type; } #endif /* -void aki_thread_detach(struct aki_thread *thread) +void nn_thread_detach(struct nn_thread *thread) { } */ -void aki_mutex_init(struct aki_mutex *mutex) +void nn_mutex_init(struct nn_mutex *mutex) { InitializeCriticalSection(&mutex->mutex); } -void aki_mutex_lock(struct aki_mutex *mutex) +void nn_mutex_lock(struct nn_mutex *mutex) { EnterCriticalSection(&mutex->mutex); } -void aki_mutex_unlock(struct aki_mutex *mutex) +void nn_mutex_unlock(struct nn_mutex *mutex) { LeaveCriticalSection(&mutex->mutex); } -void aki_mutex_destroy(struct aki_mutex *mutex) +void nn_mutex_destroy(struct nn_mutex *mutex) { DeleteCriticalSection(&mutex->mutex); } -void aki_cond_init(struct aki_cond *cond) +void nn_cond_init(struct nn_cond *cond) { cond->condition = true; InitializeConditionVariable(&cond->cond); } -void aki_cond_wait(struct aki_cond *cond, struct aki_mutex *mutex) +void nn_cond_wait(struct nn_cond *cond, struct nn_mutex *mutex) { cond->condition = false; - while (!cond->condition) { + while (!cond->condition) SleepConditionVariableCS(&cond->cond, &mutex->mutex, INFINITE); - } } -bool aki_cond_is_waiting(struct aki_cond *cond) +bool nn_cond_is_waiting(struct nn_cond *cond) { return !cond->condition; } -void aki_cond_signal(struct aki_cond *cond) +void nn_cond_signal(struct nn_cond *cond) { cond->condition = true; WakeConditionVariable(&cond->cond); } -void aki_cond_destroy(struct aki_cond *cond) +void nn_cond_destroy(struct nn_cond *cond) { (void)cond; } diff --git a/src/util/timer/timer.h b/src/util/timer/timer.h index 2bf6b95..cd2daf7 100644 --- a/src/util/timer/timer.h +++ b/src/util/timer/timer.h @@ -2,5 +2,5 @@ #include <al/types.h> -u64 aki_get_timestamp(void); -f64 aki_get_tick(void); +u64 nn_get_timestamp(void); +f64 nn_get_tick(void); diff --git a/src/util/timer/timer_linux.c b/src/util/timer/timer_linux.c index d93bfdc..6585f10 100644 --- a/src/util/timer/timer_linux.c +++ b/src/util/timer/timer_linux.c @@ -2,14 +2,14 @@ #include "timer.h" -u64 aki_get_timestamp(void) +u64 nn_get_timestamp(void) { struct timespec spec; clock_gettime(CLOCK_REALTIME, &spec); return 1e6 * spec.tv_sec + spec.tv_nsec * 1e-3; } -f64 aki_get_tick(void) +f64 nn_get_tick(void) { struct timespec spec; clock_gettime(CLOCK_MONOTONIC, &spec); diff --git a/src/util/timer/timer_windows.c b/src/util/timer/timer_windows.c index edd1297..198cb32 100644 --- a/src/util/timer/timer_windows.c +++ b/src/util/timer/timer_windows.c @@ -2,7 +2,7 @@ #include "timer.h" -f64 aki_get_tick(void) +f64 nn_get_tick(void) { LARGE_INTEGER result; QueryPerformanceCounter(&result); @@ -12,7 +12,7 @@ f64 aki_get_tick(void) // https://stackoverflow.com/a/46024468 static const s64 UNIX_TIME_START = 0x019DB1DED53E8000; // January 1st, 1970 in "ticks" -u64 aki_get_timestamp(void) +u64 nn_get_timestamp(void) { FILETIME ft; GetSystemTimePreciseAsFileTime(&ft); diff --git a/src/winwrap.c b/src/winwrap.c index 601b0f7..c695b83 100644 --- a/src/winwrap.c +++ b/src/winwrap.c @@ -1,6 +1,6 @@ #include "winwrap.h" -void aki_windows_init(void) +void nn_windows_init(void) { // https://learn.microsoft.com/en-us/windows/win32/api/objbase/ne-objbase-coinit CoInitializeEx(NULL, COINIT_MULTITHREADED | COINIT_SPEED_OVER_MEMORY); @@ -26,9 +26,9 @@ _Pragma("GCC diagnostic pop") QueryPerformanceFrequency(&performance_frequency); } -void aki_windows_close(void) +void nn_windows_close(void) { - // Still crashes on windows. + // Still crashes on Windows. //WSACleanup(); CoUninitialize(); } diff --git a/src/winwrap.h b/src/winwrap.h index bd7ad90..e47cd76 100644 --- a/src/winwrap.h +++ b/src/winwrap.h @@ -11,5 +11,5 @@ extern NTSTATUS(__stdcall *NtDelayExecution)(BOOL Alertable, PLARGE_INTEGER DelayInterval); extern LARGE_INTEGER performance_frequency; -void aki_windows_init(void); -void aki_windows_close(void); +void nn_windows_init(void); +void nn_windows_close(void); |