diff --git a/TODO.txt b/TODO.txt index 97aa35f..2c8da46 100644 --- a/TODO.txt +++ b/TODO.txt @@ -1,4 +1,7 @@ [ ] Add log compaction [ ] Add log persistence [ ] Add chunk locations to metadata -[ ] Remove sender_idx fields in messages since it's provided by the message header \ No newline at end of file +[ ] Remove sender_idx fields in messages since it's provided by the message header +[ ] HTTP timeouts +[ ] Clean up the Schannel code +[ ] Expose TLS parameters via CLI \ No newline at end of file diff --git a/lib/message.c b/lib/message.c index 76488f2..95815e4 100644 --- a/lib/message.c +++ b/lib/message.c @@ -13,6 +13,7 @@ int message_system_init(MessageSystem *msys, for (int i = 0; i < num_addrs; i++) msys->addrs[i] = addrs[i]; msys->num_addrs = num_addrs; + // TODO: sort addresses int max_conns = 2 * num_addrs + 1; diff --git a/src/client.c b/src/client.c index 201035d..a85921c 100644 --- a/src/client.c +++ b/src/client.c @@ -96,7 +96,7 @@ static const char *step_name(Step step) static void client_log_impl(ToastyFS *tfs, const char *event, const char *detail) { - Time now = get_current_time(); + Time now = get_current_time(); // TODO: check error printf("[" TIME_FMT "] CLIENT %lu %-12s V%-3lu | %-20s %s\n", TIME_VAL(now), tfs->client_id, @@ -639,12 +639,12 @@ void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, i { message_system_process_events(&tfs->msys, ctxs, pdata, pnum); - void *raw; - while ((raw = get_next_message(&tfs->msys)) != NULL) { - Message *header = (Message *)raw; - ByteView msg_view = { .ptr = raw, .len = header->length }; + void *raw_message; + while ((raw_message = get_next_message(&tfs->msys)) != NULL) { + Message *header = (Message *)raw_message; + ByteView msg_view = { .ptr = raw_message, .len = header->length }; process_message(tfs, header->type, msg_view); - consume_message(&tfs->msys, raw); + consume_message(&tfs->msys, raw_message); } // Check for operation timeout -- retry the current operation if the diff --git a/src/http_proxy.c b/src/http_proxy.c index eb286cb..53b8685 100644 --- a/src/http_proxy.c +++ b/src/http_proxy.c @@ -1,4 +1,14 @@ #include "http_proxy.h" +#include "config.h" +#include +#include +#include +#include +#ifdef _WIN32 +#include +#else +#include +#endif int http_proxy_init(void *state, int argc, char **argv, void **ctxs, struct pollfd *pdata, int pcap, int *pnum, @@ -6,30 +16,251 @@ int http_proxy_init(void *state, int argc, char **argv, { HTTPProxy *proxy = state; - if (http_server_init(&proxy->http_server) < 0) + char *addrs[NODE_LIMIT]; + int num_addrs = 0; + + for (int i = 1; i < argc; i++) { + if (!strcmp(argv[i], "--server")) { + i++; + if (i == argc) { + fprintf(stderr, "Option --server missing value\n"); + return -1; + } + if (num_addrs == NODE_LIMIT) { + fprintf(stderr, "Node limit reached\n"); + return -1; + } + addrs[num_addrs] = argv[i]; + num_addrs++; + } else { + // Ignore unknown options + } + } + + // TODO: Make these configurable + int max_opers = 128; + string http_addr = S("127.0.0.1"); + uint16_t http_port = 3000; + uint64_t client_id = 999; + + proxy->max_opers = max_opers; + proxy->opers = malloc(max_opers * sizeof(ProxyOper)); + if (proxy->opers == NULL) return -1; - if (http_server_listen_tcp(&proxy->http_server, xxx, yyy) < 0) - return -1; + for (int i = 0; i < max_opers; i++) + proxy->opers[i].state = PROXY_OPER_FREE; + if (http_server_init(&proxy->http_server, max_opers) < 0) { + free(proxy->opers); + return -1; + } + + if (http_server_listen_tcp(&proxy->http_server, http_addr, http_port) < 0) { + http_server_free(&proxy->http_server); + free(proxy->opers); + return -1; + } + + proxy->toastyfs = toastyfs_init(client_id, addrs, num_addrs); + if (proxy->toastyfs == NULL) { + http_server_free(&proxy->http_server); + free(proxy->opers); + return -1; + } + + // Register events that need to be monitored + { + int ret; + *pnum = 0; + *timeout = -1; + + ret = toastyfs_register_events(proxy->toastyfs, ctxs, pdata, pcap, timeout); + if (ret < 0) + return -1; + *pnum += ret; + proxy->num_polled_by_toasty = ret; + + ret = http_server_register_events(&proxy->http_server, + ctxs + proxy->num_polled_by_toasty, + pdata + proxy->num_polled_by_toasty, + pcap - proxy->num_polled_by_toasty); + if (ret < 0) + return -1; + *pnum += ret; + } return 0; } int http_proxy_tick(void *state, void **ctxs, struct pollfd *pdata, int pcap, int *pnum, int *timeout) { - http_server_process_events(&proxy->http_server, ctxs, pdata, *pnum); + HTTPProxy *proxy = state; + http_server_process_events(&proxy->http_server, + ctxs + proxy->num_polled_by_toasty, + pdata + proxy->num_polled_by_toasty, + *pnum - proxy->num_polled_by_toasty); + toastyfs_process_events(proxy->toastyfs, + ctxs, pdata, proxy->num_polled_by_toasty); - HTTP_Request *request; - HTTP_ResponseBuilder builder; - while (http_server_next_request(&proxy->http_server, &request, &builder)) { - // TODO - http_response_builder_status(builder, 200); - http_response_builder_submit(builder); + // Process operation resolutions + for (;;) { + ToastyFS_Result result = toastyfs_get_result(proxy->toastyfs); + if (result.type == TOASTYFS_RESULT_VOID) + break; + + // Find the started operation + int i = 0; + while (i < proxy->max_opers && proxy->opers[i].state != PROXY_OPER_STARTED) + i++; + assert(i < proxy->max_opers); // Wasn't expecting this result + + ProxyOper *oper = &proxy->opers[i]; + assert(oper->state == PROXY_OPER_STARTED); + + HTTP_Request *request = oper->request; + HTTP_ResponseBuilder builder = oper->builder; + + switch (result.type) { + case TOASTYFS_RESULT_PUT: + assert(request->method == CHTTP_METHOD_PUT); + if (result.error == TOASTYFS_ERROR_VOID) { + http_response_builder_status(builder, 201); + http_response_builder_submit(builder); + } else { + http_response_builder_status(builder, 500); + http_response_builder_submit(builder); + } + break; + case TOASTYFS_RESULT_GET: + assert(request->method == CHTTP_METHOD_GET); + if (result.error == TOASTYFS_ERROR_VOID) { + http_response_builder_status(builder, 200); + http_response_builder_content(builder, (string) { result.data, result.size }); + http_response_builder_submit(builder); + free(result.data); + } else { + http_response_builder_status(builder, 500); + http_response_builder_submit(builder); + } + break; + case TOASTYFS_RESULT_DELETE: + assert(request->method == CHTTP_METHOD_DELETE); + if (result.error == TOASTYFS_ERROR_VOID) { + http_response_builder_status(builder, 204); + http_response_builder_submit(builder); + } else { + http_response_builder_status(builder, 500); + http_response_builder_submit(builder); + } + break; + default: + UNREACHABLE; + } + + oper->state = PROXY_OPER_FREE; } - *timeout = -1; - *pnum = http_server_register_events(&proxy->http_server, ctxs, pdata, pcap); + // Buffer operation requests + for (;;) { + HTTP_Request *request; + HTTP_ResponseBuilder builder; + if (!http_server_next_request(&proxy->http_server, &request, &builder)) + break; + + // Only allow GET, PUT, DELETE requests + if (request->method != CHTTP_METHOD_GET && + request->method != CHTTP_METHOD_PUT && + request->method != CHTTP_METHOD_DELETE) { + http_response_builder_status(builder, 405); + http_response_builder_submit(builder); + continue; + } + + // Look for a free operation slot + int i = 0; + while (i < proxy->max_opers && proxy->opers[i].state != PROXY_OPER_FREE) + i++; + + if (i == proxy->max_opers) { + // Queue is full + http_response_builder_status(builder, 500); + http_response_builder_submit(builder); + continue; + } + + ProxyOper *oper = &proxy->opers[i]; + assert(oper->state == PROXY_OPER_FREE); + + oper->state = PROXY_OPER_PENDING; + oper->request = request; + oper->builder = builder; + } + + // Start operations + { + // Look for a started operation + bool started = false; + for (int i = 0; i < proxy->max_opers; i++) { + if (proxy->opers[i].state == PROXY_OPER_STARTED) { + started = true; + break; + } + } + + // Start an operation if necessary + if (!started) { + // Look for a pending operation + int i = 0; + while (i < proxy->max_opers && proxy->opers[i].state != PROXY_OPER_PENDING) + i++; + + if (i < proxy->max_opers) { + // Found pending operation + ProxyOper *oper = &proxy->opers[i]; + + // Begin the operation based on the request method + HTTP_Request *request = oper->request; + switch (request->method) { + case CHTTP_METHOD_GET: + toastyfs_async_get(proxy->toastyfs, request->url.path.ptr, request->url.path.len); + break; + case CHTTP_METHOD_PUT: + toastyfs_async_put(proxy->toastyfs, request->url.path.ptr, request->url.path.len, + request->body.ptr, request->body.len); + break; + case CHTTP_METHOD_DELETE: + toastyfs_async_delete(proxy->toastyfs, request->url.path.ptr, request->url.path.len); + break; + default: + UNREACHABLE; + } + + oper->state = PROXY_OPER_STARTED; + } + } + } + + // Register events that need to be monitored + { + int ret; + *pnum = 0; + *timeout = -1; + + ret = toastyfs_register_events(proxy->toastyfs, ctxs, pdata, pcap, timeout); + if (ret < 0) + return -1; + *pnum += ret; + proxy->num_polled_by_toasty = ret; + + ret = http_server_register_events(&proxy->http_server, + ctxs + proxy->num_polled_by_toasty, + pdata + proxy->num_polled_by_toasty, + pcap - proxy->num_polled_by_toasty); + if (ret < 0) + return -1; + *pnum += ret; + } return 0; } @@ -37,5 +268,8 @@ int http_proxy_free(void *state) { HTTPProxy *proxy = state; + toastyfs_free(proxy->toastyfs); http_server_free(&proxy->http_server); + free(proxy->opers); + return 0; } \ No newline at end of file diff --git a/src/http_proxy.h b/src/http_proxy.h index a8c5552..a94f115 100644 --- a/src/http_proxy.h +++ b/src/http_proxy.h @@ -4,9 +4,24 @@ #include #include +typedef enum { + PROXY_OPER_FREE, + PROXY_OPER_PENDING, + PROXY_OPER_STARTED, +} ProxyOperState; + +typedef struct { + ProxyOperState state; + HTTP_Request* request; + HTTP_ResponseBuilder builder; +} ProxyOper; + typedef struct { HTTP_Server http_server; ToastyFS *toastyfs; + ProxyOper *opers; + int max_opers; + int num_polled_by_toasty; } HTTPProxy; struct pollfd; diff --git a/src/main.c b/src/main.c index 537bb00..0afd0a8 100644 --- a/src/main.c +++ b/src/main.c @@ -311,4 +311,68 @@ int main(int argc, char **argv) return 0; } -#endif // MAIN_SERVER \ No newline at end of file +#endif // MAIN_SERVER +//////////////////////////////////////////////////////////////////// +//////////////////////////////////////////////////////////////////// +//////////////////////////////////////////////////////////////////// +#ifdef MAIN_HTTP_PROXY + +#ifdef _WIN32 +#include +#else +#include +#endif + +#include "http_proxy.h" + +#define POLL_CAPACITY 1024 + +int main(int argc, char **argv) +{ + int ret; + + HTTPProxy state; + + void* poll_ctxs[POLL_CAPACITY]; + struct pollfd poll_array[POLL_CAPACITY]; + int poll_count; + int poll_timeout; + + ret = http_proxy_init( + &state, + argc, + argv, + poll_ctxs, + poll_array, + POLL_CAPACITY, + &poll_count, + &poll_timeout + ); + if (ret < 0) + return -1; + + for (;;) { + +#ifdef _WIN32 + WSAPoll(poll_array, poll_count, poll_timeout); +#else + poll(poll_array, poll_count, poll_timeout); +#endif + + ret = http_proxy_tick( + &state, + poll_ctxs, + poll_array, + POLL_CAPACITY, + &poll_count, + &poll_timeout + ); + if (ret < 0) + return -1; + } + + http_proxy_free(&state); + return 0; +} + +#endif // MAIN_HTTP_PROXY \ No newline at end of file