From 174f37c6c00f06f3cf910ddaa217bb52164fadb3 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 29 Oct 2025 21:46:17 +0100 Subject: [PATCH] Continue work on DST framework --- src/basic.c | 14 +- src/byte_queue.c | 24 +- src/chunk_server.c | 62 +-- src/chunk_server.h | 4 +- src/client.c | 47 ++- src/file_system.c | 70 +-- src/file_tree.c | 13 +- src/main.c | 52 --- src/main_server.c | 62 +++ src/main_test.c | 957 ++++++++++++++++++++++++++++++++++++++++++ src/metadata_server.c | 14 +- src/metadata_server.h | 2 +- src/system.c | 218 ++++++++-- src/system.h | 100 +++-- src/tcp.c | 11 +- src/tcp.h | 3 +- src/test.c | 464 -------------------- 17 files changed, 1407 insertions(+), 710 deletions(-) delete mode 100644 src/main.c create mode 100644 src/main_server.c create mode 100644 src/main_test.c delete mode 100644 src/test.c diff --git a/src/basic.c b/src/basic.c index edd41a3..1d4f014 100644 --- a/src/basic.c +++ b/src/basic.c @@ -1,11 +1,5 @@ -#ifdef _WIN32 -#define WIN32_LEAN_AND_MEAN -#include -#else -#include -#endif - #include "basic.h" +#include "system.h" bool streq(string s1, string s2) { @@ -28,10 +22,10 @@ Time get_current_time(void) int64_t freq; int ok; - ok = QueryPerformanceCounter((LARGE_INTEGER*) &count); + ok = sys_QueryPerformanceCounter((LARGE_INTEGER*) &count); if (!ok) return INVALID_TIME; - ok = QueryPerformanceFrequency((LARGE_INTEGER*) &freq); + ok = sys_QueryPerformanceFrequency((LARGE_INTEGER*) &freq); if (!ok) return INVALID_TIME; uint64_t res = 1000 * (double) count / freq; @@ -41,7 +35,7 @@ Time get_current_time(void) { struct timespec time; - if (clock_gettime(CLOCK_REALTIME, &time)) + if (sys_clock_gettime(CLOCK_REALTIME, &time)) return INVALID_TIME; uint64_t res; diff --git a/src/byte_queue.c b/src/byte_queue.c index 9282095..caf2dc6 100644 --- a/src/byte_queue.c +++ b/src/byte_queue.c @@ -2,6 +2,7 @@ #include #include +#include "system.h" #include "byte_queue.h" // This is the implementation of a byte queue useful @@ -13,19 +14,6 @@ // // Only up to 4GB of data can be stored at once. -static void *mymalloc(ByteQueue *queue, uint32_t len) -{ - (void) queue; - return malloc(len); -} - -static void myfree(ByteQueue *queue, void *ptr, uint32_t len) -{ - (void) queue; - (void) len, - free(ptr); -} - // Initialize the queue void byte_queue_init(ByteQueue *queue, uint32_t limit) { @@ -44,12 +32,12 @@ void byte_queue_free(ByteQueue *queue) { if (queue->read_target) { if (queue->read_target != queue->data) - myfree(queue, queue->read_target, queue->read_target_size); + sys_free(queue->read_target); queue->read_target = NULL; queue->read_target_size = 0; } - myfree(queue, queue->data, queue->size); + sys_free(queue->data); queue->data = NULL; } @@ -111,7 +99,7 @@ void byte_queue_read_ack(ByteQueue *queue, uint32_t num) if (queue->read_target) { if (queue->read_target != queue->data) - myfree(queue, queue->read_target, queue->read_target_size); + sys_free(queue->read_target); queue->read_target = NULL; queue->read_target_size = 0; } @@ -236,7 +224,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap) if (size > queue->limit) size = queue->limit; - uint8_t *data = mymalloc(queue, size); + uint8_t *data = sys_malloc(size); if (!data) { queue->flags |= BYTE_QUEUE_ERROR; return 0; @@ -246,7 +234,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap) memcpy(data, queue->data + queue->head, queue->used); if (queue->read_target != queue->data) - myfree(queue, queue->data, queue->size); + sys_free(queue->data); queue->data = data; queue->head = 0; diff --git a/src/chunk_server.c b/src/chunk_server.c index fe845f3..0d68168 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -19,7 +19,7 @@ pending_download_list_init(PendingDownloadList *list) static void pending_download_list_free(PendingDownloadList *list) { - free(list->items); + sys_free(list->items); } static int @@ -36,13 +36,13 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash) if (list->capacity == 0) new_capacity = 8; else new_capacity = 2 * list->capacity; - PendingDownload *new_items = malloc(new_capacity * sizeof(PendingDownload)); + PendingDownload *new_items = sys_malloc(new_capacity * sizeof(PendingDownload)); if (new_items == NULL) return -1; if (list->capacity > 0) { memcpy(new_items, list->items, list->count * sizeof(list->items[0])); - free(list->items); + sys_free(list->items); } list->items = new_items; @@ -135,12 +135,12 @@ static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk, return -1; if (patch_off > SIZE_MAX - patch.len) { - free(data.ptr); + sys_free(data.ptr); return -1; } if (patch_off + (size_t) patch.len > (size_t) data.len) { - free(data.ptr); + sys_free(data.ptr); return -1; } @@ -148,11 +148,11 @@ static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk, ret = store_chunk(store, data, new_hash); if (ret < 0) { - free(data.ptr); + sys_free(data.ptr); return -1; } - free(data.ptr); + sys_free(data.ptr); return 0; } @@ -233,33 +233,33 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView if (!binary_read(&reader, &rem_count, sizeof(rem_count))) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - SHA256 *add_list = malloc(add_count * sizeof(SHA256)); - SHA256 *rem_list = malloc(rem_count * sizeof(SHA256)); + SHA256 *add_list = sys_malloc(add_count * sizeof(SHA256)); + SHA256 *rem_list = sys_malloc(rem_count * sizeof(SHA256)); if (add_list == NULL || rem_list == NULL) { - free(add_list); - free(rem_list); + sys_free(add_list); + sys_free(rem_list); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); } for (uint32_t i = 0; i < add_count; i++) { if (!binary_read(&reader, &add_list[i], sizeof(SHA256))) { - free(add_list); - free(rem_list); + sys_free(add_list); + sys_free(rem_list); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); } } for (uint32_t i = 0; i < rem_count; i++) { if (!binary_read(&reader, &rem_list[i], sizeof(SHA256))) { - free(add_list); - free(rem_list); + sys_free(add_list); + sys_free(rem_list); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); } } if (binary_read(&reader, NULL, 1)) { - free(add_list); - free(rem_list); + sys_free(add_list); + sys_free(rem_list); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); } @@ -302,14 +302,14 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView (string) { main_path, strlen(main_path) }) < 0) { // Failed to move, treat as missing if (missing_chunks == NULL) - missing_chunks = malloc(add_count * sizeof(SHA256)); + missing_chunks = sys_malloc(add_count * sizeof(SHA256)); if (missing_chunks) missing_chunks[missing_count++] = add_list[i]; } } else { // Chunk is missing in both locations if (missing_chunks == NULL) - missing_chunks = malloc(add_count * sizeof(SHA256)); + missing_chunks = sys_malloc(add_count * sizeof(SHA256)); if (missing_chunks) missing_chunks[missing_count++] = add_list[i]; } @@ -338,8 +338,8 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView (string) { orphaned_path, strlen(orphaned_path) }); } - free(add_list); - free(rem_list); + sys_free(add_list); + sys_free(rem_list); // Send response if (missing_count > 0) { @@ -356,7 +356,7 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView for (uint32_t i = 0; i < missing_count; i++) message_write(&writer, &missing_chunks[i], sizeof(SHA256)); - free(missing_chunks); + sys_free(missing_chunks); if (!message_writer_free(&writer)) return -1; @@ -625,7 +625,7 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg) if (binary_read(&reader, NULL, 1)) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Invalid message")); - char *mem = malloc(chunk_size); + char *mem = sys_malloc(chunk_size); if (mem == NULL) return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Out of memory")); @@ -639,7 +639,7 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg) int ret = chunk_store_add(&state->store, (string) { mem, chunk_size }); - free(mem); + sys_free(mem); if (ret < 0) return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("I/O error")); @@ -732,7 +732,7 @@ process_client_download_chunk(ChunkServer *state, int conn_idx, ByteView msg) return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("I/O error")); if (target_off >= (size_t) data.len || target_len > (size_t) data.len - target_off) { - free(data.ptr); + sys_free(data.ptr); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid range")); } string slice = { data.ptr + target_off, target_len }; @@ -746,7 +746,7 @@ process_client_download_chunk(ChunkServer *state, int conn_idx, ByteView msg) message_write(&writer, slice.ptr, slice.len); - free(data.ptr); + sys_free(data.ptr); if (!message_writer_free(&writer)) return -1; @@ -765,7 +765,7 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView return -1; } -int chunk_server_init(ChunkServer *state, int argc, char **argv) +int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled) { (void) argc; (void) argv; @@ -806,7 +806,7 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv) state->metadata_server_disconnect_time = 0; - return 0; + return tcp_register_events(&state->tcp, contexts, polled); } int chunk_server_free(ChunkServer *state) @@ -817,10 +817,10 @@ int chunk_server_free(ChunkServer *state) return 0; } -int chunk_server_step(ChunkServer *state) +int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled, int num_polled) { Event events[MAX_CONNS+1]; - int num_events = tcp_process_events(&state->tcp, events); + int num_events = tcp_translate_events(&state->tcp, contexts, polled, num_polled, events); Time current_time = get_current_time(); if (current_time == INVALID_TIME) @@ -938,5 +938,5 @@ int chunk_server_step(ChunkServer *state) } } - return 0; + return tcp_register_events(&state->tcp, contexts, polled, num_polled); } diff --git a/src/chunk_server.h b/src/chunk_server.h index d017b2e..0125f5e 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -35,8 +35,8 @@ typedef struct { PendingDownloadList pending_download_list; } ChunkServer; -int chunk_server_init(ChunkServer *state, int argc, char **argv); +int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled); int chunk_server_free(ChunkServer *state); -int chunk_server_step(ChunkServer *state); +int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled, int num_polled); #endif // CHUNK_SERVER_INCLUDED diff --git a/src/client.c b/src/client.c index fdfc6be..c0bba09 100644 --- a/src/client.c +++ b/src/client.c @@ -8,6 +8,7 @@ #endif #include "tcp.h" +#include "system.h" #include "config.h" #include "message.h" #include @@ -90,7 +91,7 @@ struct TinyDFS { TinyDFS *tinydfs_init(char *addr, uint16_t port) { - TinyDFS *tdfs = malloc(sizeof(TinyDFS)); + TinyDFS *tdfs = sys_malloc(sizeof(TinyDFS)); if (tdfs == NULL) return NULL; @@ -98,7 +99,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port) addr2.is_ipv4 = true; addr2.port = port; if (inet_pton(AF_INET, addr, &addr2.ipv4) != 1) { - free(tdfs); + sys_free(tdfs); return NULL; } @@ -106,7 +107,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port) if (tcp_connect(&tdfs->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) { tcp_context_free(&tdfs->tcp); - free(tdfs); + sys_free(tdfs); return NULL; } @@ -121,7 +122,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port) void tinydfs_free(TinyDFS *tdfs) { tcp_context_free(&tdfs->tcp); - free(tdfs); + sys_free(tdfs); } static int @@ -404,7 +405,7 @@ int tinydfs_submit_write(TinyDFS *tdfs, char *path, int path_len, int off, void void tinydfs_result_free(TinyDFS_Result *result) { if (result->type == TINYDFS_RESULT_LIST_SUCCESS) - free(result->entities); + sys_free(result->entities); } static void process_event_for_create(TinyDFS *tdfs, @@ -537,7 +538,7 @@ static void process_event_for_list(TinyDFS *tdfs, return; } - TinyDFS_Entity *entities = malloc(item_count * sizeof(TinyDFS_Entity)); + TinyDFS_Entity *entities = sys_malloc(item_count * sizeof(TinyDFS_Entity)); if (entities == NULL) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; return; @@ -548,21 +549,21 @@ static void process_event_for_list(TinyDFS *tdfs, uint8_t is_dir; if (!binary_read(&reader, &is_dir, sizeof(is_dir))) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; - free(entities); + sys_free(entities); return; } uint16_t name_len; if (!binary_read(&reader, &name_len, sizeof(name_len))) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; - free(entities); + sys_free(entities); return; } char *name = reader.src + reader.cur; if (!binary_read(&reader, NULL, name_len)) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; - free(entities); + sys_free(entities); return; } @@ -570,7 +571,7 @@ static void process_event_for_list(TinyDFS *tdfs, if (name_len > sizeof(entities[i].name)-1) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; - free(entities); + sys_free(entities); return; } memcpy(entities[i].name, name, name_len); @@ -580,7 +581,7 @@ static void process_event_for_list(TinyDFS *tdfs, // Check there is nothing else to read if (binary_read(&reader, NULL, 1)) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; - free(entities); + sys_free(entities); return; } @@ -653,7 +654,7 @@ static void process_event_for_read(TinyDFS *tdfs, } // Allocate ranges - Range *ranges = malloc(num_chunks_needed * sizeof(Range)); + Range *ranges = sys_malloc(num_chunks_needed * sizeof(Range)); if (ranges == NULL) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; @@ -667,7 +668,7 @@ static void process_event_for_read(TinyDFS *tdfs, // Read hash SHA256 hash; if (!binary_read(&reader, &hash, sizeof(hash))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -675,7 +676,7 @@ static void process_event_for_read(TinyDFS *tdfs, // Read number of servers uint32_t num_servers; if (!binary_read(&reader, &num_servers, sizeof(num_servers))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -683,7 +684,7 @@ static void process_event_for_read(TinyDFS *tdfs, // Parse IPv4 addresses uint32_t num_ipv4; if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -697,7 +698,7 @@ static void process_event_for_read(TinyDFS *tdfs, uint16_t port; if (!binary_read(&reader, &ipv4, sizeof(ipv4)) || !binary_read(&reader, &port, sizeof(port))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -712,21 +713,21 @@ static void process_event_for_read(TinyDFS *tdfs, // Skip IPv6 addresses uint32_t num_ipv6; if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } for (uint32_t j = 0; j < num_ipv6; j++) { if (!binary_read(&reader, NULL, sizeof(IPv6)) || !binary_read(&reader, NULL, sizeof(uint16_t))) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } } if (!found) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -771,7 +772,7 @@ static void process_event_for_read(TinyDFS *tdfs, Range *r = &ranges[0]; int cs_idx = get_chunk_server_connection(tdfs, r->server_addr); if (cs_idx < 0) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -779,7 +780,7 @@ static void process_event_for_read(TinyDFS *tdfs, if (send_download_chunk(tdfs, cs_idx, r->hash, r->offset_within_chunk, r->length_within_chunk, opidx, 0) < 0) { - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; return; } @@ -788,7 +789,7 @@ static void process_event_for_read(TinyDFS *tdfs, tdfs->operations[opidx].ranges_head = 1; } else { // No chunks to download - free(ranges); + sys_free(ranges); tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS }; } @@ -861,7 +862,7 @@ static void process_event_for_read(TinyDFS *tdfs, // Check if done if (tdfs->operations[opidx].num_pending == 0) { - free(tdfs->operations[opidx].ranges); + sys_free(tdfs->operations[opidx].ranges); tdfs->operations[opidx].ranges = NULL; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS }; } diff --git a/src/file_system.c b/src/file_system.c index fe1ce8d..472e053 100644 --- a/src/file_system.c +++ b/src/file_system.c @@ -17,6 +17,7 @@ #include #endif +#include "system.h" #include "file_system.h" int rename_file_or_dir(string oldpath, string newpath); @@ -30,7 +31,7 @@ int file_open(string path, Handle *fd) memcpy(zt, path.ptr, path.len); zt[path.len] = '\0'; - int ret = open(zt, O_RDWR | O_CREAT | O_APPEND, 0644); + int ret = sys_open(zt, O_RDWR | O_CREAT | O_APPEND, 0644); if (ret < 0) return -1; @@ -43,7 +44,7 @@ int file_open(string path, Handle *fd) MultiByteToWideChar(CP_UTF8, 0, path.ptr, path.len, wpath, MAX_PATH); wpath[path.len] = L'\0'; - HANDLE h = CreateFileW( + HANDLE h = sys_CreateFileW( wpath, GENERIC_WRITE | GENERIC_READ, 0, @@ -63,24 +64,24 @@ int file_open(string path, Handle *fd) void file_close(Handle fd) { #ifdef __linux__ - close((int) fd.data); + sys_close((int) fd.data); #endif #ifdef _WIN32 - CloseHandle((HANDLE) fd.data); + sys_CloseHandle((HANDLE) fd.data); #endif } int file_lock(Handle fd) { #ifdef __linux__ - if (flock((int) fd.data, LOCK_EX) < 0) + if (sys_flock((int) fd.data, LOCK_EX) < 0) return -1; return 0; #endif #ifdef _WIN32 - if (!LockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) + if (!sys_LockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) return -1; return 0; #endif @@ -89,13 +90,13 @@ int file_lock(Handle fd) int file_unlock(Handle fd) { #ifdef __linux__ - if (flock((int) fd.data, LOCK_UN) < 0) + if (sys_flock((int) fd.data, LOCK_UN) < 0) return -1; return 0; #endif #ifdef _WIN32 - if (!UnlockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) + if (!sys_UnlockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) return -1; return 0; #endif @@ -104,13 +105,13 @@ int file_unlock(Handle fd) int file_sync(Handle fd) { #ifdef __linux__ - if (fsync((int) fd.data) < 0) + if (sys_fsync((int) fd.data) < 0) return -1; return 0; #endif #ifdef _WIN32 - if (!FlushFileBuffers((HANDLE) fd.data)) + if (!sys_FlushFileBuffers((HANDLE) fd.data)) return -1; return 0; #endif @@ -119,12 +120,12 @@ int file_sync(Handle fd) int file_read(Handle fd, char *dst, int max) { #ifdef __linux__ - return read((int) fd.data, dst, max); + return sys_read((int) fd.data, dst, max); #endif #ifdef _WIN32 DWORD num; - if (!ReadFile((HANDLE) fd.data, dst, max, &num, NULL)) + if (!sys_ReadFile((HANDLE) fd.data, dst, max, &num, NULL)) return -1; if (num > INT_MAX) return -1; @@ -135,12 +136,12 @@ int file_read(Handle fd, char *dst, int max) int file_write(Handle fd, char *src, int len) { #ifdef __linux__ - return write((int) fd.data, src, len); + return sys_write((int) fd.data, src, len); #endif #ifdef _WIN32 DWORD num; - if (!WriteFile((HANDLE) fd.data, src, len, &num, NULL)) + if (!sys_WriteFile((HANDLE) fd.data, src, len, &num, NULL)) return -1; if (num > INT_MAX) return -1; @@ -152,7 +153,7 @@ int file_size(Handle fd, size_t *len) { #ifdef __linux__ struct stat buf; - if (fstat((int) fd.data, &buf) < 0) + if (sys_fstat((int) fd.data, &buf) < 0) return -1; if (buf.st_size < 0 || (uint64_t) buf.st_size > SIZE_MAX) return -1; @@ -162,7 +163,7 @@ int file_size(Handle fd, size_t *len) #ifdef _WIN32 LARGE_INTEGER buf; - if (!GetFileSizeEx((HANDLE) fd.data, &buf)) + if (!sys_GetFileSizeEx((HANDLE) fd.data, &buf)) return -1; if (buf.QuadPart < 0 || (uint64_t) buf.QuadPart > SIZE_MAX) return -1; @@ -193,7 +194,7 @@ static int write_bytes(int fd, string data) { size_t written = 0; while (written < (size_t) data.len) { - int ret = write(fd, data.ptr + written, data.len - written); + int ret = sys_write(fd, data.ptr + written, data.len - written); if (ret < 0) { if (errno == EINTR) continue; @@ -218,26 +219,26 @@ int file_write_atomic(string path, string content) memcpy(tmp_path + parent.len, pattern, strlen(pattern)); tmp_path[parent.len + strlen(pattern)] = '\0'; - int fd = mkstemp(tmp_path); + int fd = sys_mkstemp(tmp_path); if (fd < 0) return -1; if (write_bytes(fd, content) < 0) { - close(fd); - remove(tmp_path); + sys_close(fd); + sys_remove(tmp_path); return -1; } #ifdef _WIN32 - if (_commit(fd)) { - close(fd); - remove(tmp_path); + if (sys__commit(fd)) { + sys_close(fd); + sys_remove(tmp_path); return -1; } #else - if (fsync(fd)) { - close(fd); - remove(tmp_path); + if (sys_fsync(fd)) { + sys_close(fd); + sys_remove(tmp_path); return -1; } #endif @@ -245,7 +246,7 @@ int file_write_atomic(string path, string content) close(fd); if (rename_file_or_dir((string) { tmp_path, strlen(tmp_path) }, path)) { - remove(tmp_path); + sys_remove(tmp_path); return -1; } return 0; @@ -260,10 +261,10 @@ int create_dir(string path) zt[path.len] = '\0'; #ifdef _WIN32 - if (mkdir(zt) < 0) + if (sys_mkdir(zt) < 0) return -1; #else - if (mkdir(zt, 0766)) + if (sys_mkdir(zt, 0766)) return -1; #endif @@ -284,7 +285,7 @@ int rename_file_or_dir(string oldpath, string newpath) memcpy(newpath_zt, newpath.ptr, newpath.len); newpath_zt[newpath.len] = '\0'; - if (rename(oldpath_zt, newpath_zt)) + if (sys_rename(oldpath_zt, newpath_zt)) return -1; return 0; } @@ -297,7 +298,7 @@ int remove_file_or_dir(string path) memcpy(path_zt, path.ptr, path.len); path_zt[path.len] = '\0'; - if (remove(path_zt)) + if (sys_remove(path_zt)) return -1; return 0; } @@ -311,12 +312,12 @@ int get_full_path(string path, char *dst) path_zt[path.len] = '\0'; #ifdef __linux__ - if (realpath(path_zt, dst) == NULL) + if (sys_realpath(path_zt, dst) == NULL) return -1; #endif #ifdef _WIN32 - if (_fullpath(path_zt, dst, PATH_MAX) == NULL) + if (sys__fullpath(path_zt, dst, PATH_MAX) == NULL) return -1; #endif @@ -341,7 +342,7 @@ int file_read_all(string path, string *data) return -1; } - char *dst = malloc(len); + char *dst = sys_malloc(len); if (dst == NULL) { file_close(fd); return -1; @@ -351,6 +352,7 @@ int file_read_all(string path, string *data) while ((size_t) copied < len) { ret = file_read(fd, dst + copied, len - copied); if (ret < 0) { + sys_free(dst); file_close(fd); return -1; } diff --git a/src/file_tree.c b/src/file_tree.c index 4112a54..5d6e16a 100644 --- a/src/file_tree.c +++ b/src/file_tree.c @@ -2,6 +2,7 @@ #include #include +#include "system.h" #include "file_tree.h" static int parse_path(string path, string *comps, int max) @@ -99,7 +100,7 @@ static void dir_free(Dir *d) { for (uint64_t i = 0; i < d->num_children; i++) entity_free(&d->children[i]); - free(d->children); + sys_free(d->children); } static void dir_remove(Dir *d, int idx) @@ -124,7 +125,7 @@ static void file_init(File *f, uint64_t chunk_size) static void file_free(File *f) { - free(f->chunks); + sys_free(f->chunks); f->chunks = NULL; } @@ -268,14 +269,14 @@ int file_tree_create_entity(FileTree *ft, string path, if (new_max == 0) new_max = 8; - Entity *p = malloc(sizeof(Entity) * new_max); + Entity *p = sys_malloc(sizeof(Entity) * new_max); if (p == NULL) return FILETREE_NOMEM; for (uint64_t i = 0; i < d->num_children; i++) p[i] = d->children[i]; - free(d->children); + sys_free(d->children); d->children = p; d->max_children = new_max; } @@ -340,13 +341,13 @@ int file_tree_write(FileTree *ft, string path, uint64_t last_chunk_index = (off + len - 1) / f->chunk_size; if (last_chunk_index >= f->num_chunks) { - SHA256 *new_chunks = malloc((last_chunk_index+1) * sizeof(SHA256)); + SHA256 *new_chunks = sys_malloc((last_chunk_index+1) * sizeof(SHA256)); if (new_chunks == NULL) return FILETREE_NOMEM; if (f->chunks) { if (f->num_chunks > 0) memcpy(new_chunks, f->chunks, f->num_chunks); - free(f->chunks); + sys_free(f->chunks); } f->chunks = new_chunks; f->num_chunks = last_chunk_index+1; diff --git a/src/main.c b/src/main.c deleted file mode 100644 index 41a380d..0000000 --- a/src/main.c +++ /dev/null @@ -1,52 +0,0 @@ -#ifdef BUILD_SERVER - -#include - -#include "chunk_server.h" -#include "metadata_server.h" - -bool is_leader(int argc, char **argv) -{ - for (int i = 1; i < argc; i++) - if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l")) - return true; - return false; -} - -int main(int argc, char **argv) -{ - int ret; - if (is_leader(argc, argv)) { - - MetadataServer state; - ret = metadata_server_init(&state, argc, argv); - if (ret) - return ret; - - for (;;) { - ret = metadata_server_step(&state); - if (ret) - return ret; - } - - ret = metadata_server_free(&state); - - } else { - - ChunkServer state; - ret = chunk_server_init(&state, argc, argv); - if (ret) - return ret; - - for (;;) { - ret = chunk_server_step(&state); - if (ret) - return ret; - } - - ret = chunk_server_free(&state); - } - return ret; -} - -#endif // BUILD_SERVER diff --git a/src/main_server.c b/src/main_server.c new file mode 100644 index 0000000..c88d6a8 --- /dev/null +++ b/src/main_server.c @@ -0,0 +1,62 @@ +#ifdef BUILD_SERVER + +#include + +#include "core/chunk_server.h" +#include "core/metadata_server.h" + +int metadata_server_main(int argc, char **argv) +{ + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled; + MetadataServer state; + num_polled = metadata_server_init( + &state, argc, argv, contexts, polled); + if (num_polled < 0) return -1; + for (;;) { + POLL(polled, num_polled, -1); + num_polled = metadata_server_step( + &state, contexts, polled, num_polled); + if (num_polled < 0) return -1; + } + metadata_server_free(&state); + return 0; +} + +int chunk_server_main(int argc, char **argv) +{ + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled; + ChunkServer state; + num_polled = chunk_server_init( + &state, argc, argv, contexts, polled); + if (num_polled < 0) return -1; + for (;;) { + POLL(polled, num_polled, -1); + num_polled = chunk_server_step( + &state, contexts, polled, num_polled); + if (num_polled < 0) return -1; + } + chunk_server_free(&state); + return 0; +} + +bool is_leader(int argc, char **argv) +{ + for (int i = 1; i < argc; i++) + if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l")) + return true; + return false; +} + +int main(int argc, char **argv) +{ + if (is_leader(argc, argv)) + return metadata_server_main(argc, argv); + else + return chunk_server_main(argc, argv); +} + +#endif diff --git a/src/main_test.c b/src/main_test.c new file mode 100644 index 0000000..ac26341 --- /dev/null +++ b/src/main_test.c @@ -0,0 +1,957 @@ +#ifdef BUILD_TEST + +#include "core/chunk_server.h" +#include "core/metadata_server.h" + +#define MAX_PROCESSES 128 +#define MAX_DESCRIPTORS 1024 + +typedef enum { + DESCRIPTOR_TYPE_EMPTY, + DESCRIPTOR_TYPE_FILE, + DESCRIPTOR_TYPE_SOCKET, + DESCRIPTOR_TYPE_LISTENER_SOCKET, + DESCRIPTOR_TYPE_CONNECTION_SOCKET, +} DescriptorType; + +typedef struct { + int peer_process; + int peer_fd; +} PendingAccept; + +typedef struct { + + // Common + DescriptorType type; + + // File + int real_fd; + + // Socket + bool bound; + struct sockaddr_in addr; + + // Listener Socket + PendingAccept *accept_queue; + int accept_queue_head; + int accept_queue_size; + int accept_queue_used; + + // Connection Socket + bool pending; + int peer_process; + int peer_fd; + char *input; + int input_used; + int input_size; + char *output; + int output_used; + int output_size; + +} Descriptor; + +typedef struct { + int num_desc; + Descriptor desc[MAX_DESCRIPTORS]; +} DescriptorTable; + +typedef struct { + void *ptr; + size_t len; + char *file; + int line; +} Allocation; + +typedef struct { + size_t mem_usage; + int num_allocs; + Allocation allocs[MAX_ALLOCATIONS]; +} AllocationTable; + +typedef enum { + PROCESS_TYPE_CLIENT, + PROCESS_TYPE_CHUNK_SERVER, + PROCESS_TYPE_METADATA_SERVER, +} ProcessType; + +typedef struct { + ProcessType type; + union { + ChunkServer chunk_server; + MetadataServer metadata_server; + }; + + DescriptorTable *dt; + AllocationTable *at; + +} Process; + +int num_processes = 0; +Process processes[MAX_PROCESSES]; +Process *current_process = NULL; +uint64_t current_time = 0; + +int find_unused_descriptor(Process *process) +{ + if (process->dt->num_desc == MAX_DESCRIPTORS) + return -1; + + int i = 0; + while (process->dt->desc[i].type != DESCRIPTOR_TYPE_EMPTY) + i++; + + return i; +} + +void descriptor_init_as_file(Descriptor *desc, int real_fd) +{ + desc->type = DESCRIPTOR_TYPE_FILE; + desc->real_fd = real_fd; +} + +void descriptor_init_as_socket(Descriptor *desc) +{ + desc->type = DESCRIPTOR_TYPE_SOCKET; + desc->bound = false; +} + +void descriptor_turn_socket_into_listener(Descriptor *desc, int backlog) +{ + desc->type = DESCRIPTOR_TYPE_LISTENER_SOCKET; + desc->accept_queue = malloc(backlog * sizeof(PendingAccept)); + desc->accept_queue_head = 0; + desc->accept_queue_used = 0; + desc->accept_queue_size = backlog; + if (desc->accept_queue == NULL) + __builtin_trap(); +} + +void descriptor_turn_socket_into_connection(Descriptor *desc, int peer_process, int peer_fd) +{ + desc->type = DESCRIPTOR_TYPE_LISTENER_SOCKET; + desc->pending = true; + desc->peer_process = peer_process; + desc->peer_fd = peer_fd; + desc->input = malloc(1<<9); + desc->input_size = 1<<9; + desc->input_used = 0; + desc->output = malloc(1<<9); + desc->output_size = 1<<9; + desc->output_used = 0; + if (desc->input == NULL || desc->output == NULL) + __builtin_trap(); +} + +void descriptor_free(Descriptor *desc) +{ + switch (desc->type) { + + case DESCRIPTOR_TYPE_EMPTY: + // TODO + break; + + case DESCRIPTOR_TYPE_FILE: + close(desc->proxy_fd); + break; + + case DESCRIPTOR_TYPE_SOCKET: + break; + + case DESCRIPTOR_TYPE_LISTENER_SOCKET: + free(desc->accept_queue); + break; + + case DESCRIPTOR_TYPE_CONNECTION_SOCKET: + free(desc->input); + free(desc->output); + break; + } + desc->type = DESCRIPTOR_TYPE_EMPTY; +} + +bool is_leader(int argc, char **argv) +{ + for (int i = 1; i < argc; i++) + if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l")) + return true; + return false; +} + +int split_args(char *args, char **out, int max) +{ + int len = strlen(args); + int cur = 0; + int num = 0; + + for (;;) { + + while (cur < len && (args[cur] == ' ' || args[cur] == '\t')) + cur++; + + int off = cur; + while (cur < len && (args[cur] != ' ' && args[cur] != '\t')) + cur++; + + if (num < max) + out[num++] = args + off; + + args[cur++] = '\0'; + } + + return num; +} + +#define MAX_ARGS 128 + +void process_init(Process *process, char *args) +{ + process->dt = malloc(sizeof(DescriptorTable)); + process->at = malloc(sizeof(AllocationTable)); + if (process->dt == NULL || process->at == NULL) + abort(); + + process->dt->num_desc = 0; + process->at->num_allocs = 0; + process->at->mem_usage = 0; + + char *argv[MAX_ARGS]; + int argc = split_args(args, argv, MAX_ARGS); + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled; + + current_process = process; + if (is_leader(argc, argv)) { + process->type = PROCESS_TYPE_METADATA_SERVER; + num_polled = metadata_server_init(&process->metadata_server, argc, argv, contexts, polled); + } else { + process->type = PROCESS_TYPE_CHUNK_SERVER; + num_polled = chunk_server_init(&process->chunk_server, argc, argv, contexts, polled); + } + current_process = NULL; + + if (num_polled < 0) { + // TODO + } + do_stuff_with_pollfds(contexts, polled, num_polled); +} + +void process_step(Process *process) +{ + if (process->type == PROCESS_TYPE_METADATA_SERVER) + num_polled = metadata_server_step(&process->metadata_server, xxx, yyyy, contexts, polled, num_polled); + else + num_polled = chunk_server_step(&process->chunk_server, xxx, yyyy, contexts, polled, num_polled); + + if (num_polled < 0) { + // TODO + } + do_stuff_with_pollfds(contexts, polled, num_polled); +} + +void process_free(Process *process) +{ + if (process->type == PROCESS_TYPE_METADATA_SERVER) + metadata_server_free(&process->metadata_server); + else + chunk_server_free(&process->chunk_server); + + free(process->at); + free(process->dt); +} + +void spawn_simulated_process(char *args) +{ + if (num_processes == MAX_PROCESSES) + abort(); + + process_init(&processes[num_processes++], args); +} + +void update_simulation(void) +{ + for (int i = 0; i < num_processes; i++) + process_step(&processes[i]); + + // TODO +} + +void cleanup_simulation(void) +{ + for (int i = 0; i < num_processes; i++) + process_free(&processes[i]); +} + +sig_atomic_t simulation_should_stop = false; + +int main(int argc, char **argv) +{ + // TODO: set simulation_should_stop=true on ctrl+C + + spawn_simulated_process("--addr 127.0.0.1 8080 --leader"); + spawn_simulated_process("--addr 127.0.0.1 8081"); + spawn_simulated_process("--addr 127.0.0.1 8082"); + spawn_simulated_process("--addr 127.0.0.1 8083"); + spawn_simulated_process("--addr 127.0.0.1 8084"); + spawn_simulated_process("--addr 127.0.0.1 8085"); + spawn_simulated_process("--addr 127.0.0.1 8086"); + spawn_simulated_process("--addr 127.0.0.1 8087"); + spawn_simulated_process("--addr 127.0.0.1 8088"); + spawn_simulated_process("--addr 127.0.0.1 8089"); + spawn_simulated_process("--addr 127.0.0.1 8090"); + + while (!simulation_should_stop) + update_simulation(); + + cleanup_simulation(); + return 0; +} + +void *sys_malloc_(size_t len, char *file, int line) +{ + if (current_process->at->num_allocs == MAX_ALLOCATIONS) + __builtin_trap(); + + void *ptr = malloc(len); + if (ptr == NULL) + __builtin_trap(); + + current_process->at->allocs[current_process->at->num_allocs++] = (Allocation) { ptr, len, file, line }; + current_process->at->mem_usage += len; + return ptr; +} + +void *sys_realloc_(void *ptr, size_t len, char *file, int line) +{ + int found = -1; + for (int i = 0; i < current_process->at->num_allocs; i++) + if (current_process->at->allocs[i].ptr == ptr) { + found = i; + break; + } + if (found < 0) + __builtin_trap(); + + size_t old_len = current_process->at->allocs[found].len; + void *new_ptr = realloc(ptr, len); + if (new_ptr == NULL) + __builtin_trap(); + + current_process->at->allocs[found].ptr = new_ptr; + current_process->at->allocs[found].len = len; + current_process->at->allocs[found].file = file; + current_process->at->allocs[found].line = line; + + current_process->at->mem_usage -= old_len; + current_process->at->mem_usage += len; + + return new_ptr; +} + +void sys_free_(void *ptr, char *file, int line) +{ + (void) file; + (void) line; + + int found = -1; + for (int i = 0; i < current_process->at->num_allocs; i++) + if (current_process->at->allocs[i].ptr == ptr) { + found = i; + break; + } + if (found < 0) + __builtin_trap(); + + current_process->at->mem_usage -= current_process->at->allocs[found].len; + current_process->at->allocs[found] = current_process->at->allocs[--current_process->num_allocs]; +} + +#ifdef _WIN32 + +SOCKET sys_socket(int domain, int type, int protocol) +{ + if (domain != AF_INET) { + // TODO: errno + return INVALID_SOCKET; + } + + if (type != SOCK_STREAM) { + // TODO: errno + return INVALID_SOCKET; + } + + if (protocol != 0) { + // TODO: errno + return INVALID_SOCKET; + } + + int fd = find_unused_descriptor(current_process); + if (fd < 0) { + // TODO: errno + return INVALID_SOCKET; + } + + // TODO: maybe bind to a random address? + + descriptor_init_as_socket(¤t_process->dt->desc[fd]); + return (SOCKET) fd; +} + +int sys_bind(SOCKET fd, void *addr, size_t addr_len) +{ + Descriptor *desc = ¤t_process->dt->desc[(int) fd]; + if (desc->type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { + // TODO: errno + return -1; + } + + if (addr_len != sizeof(desc->bind)) { + // TODO: errno + return -1; + } + + // TODO: maybe check that no one else is listening + // on this port + + desc->bound = true; + memcpy(&desc->addr, addr, addr_len); + return 0; +} + +int sys_listen(SOCKET fd, int backlog) +{ + Descriptor *desc = ¤t_process->dt->desc[(int) fd]; + if (desc->type != DESCRIPTOR_TYPE_SOCKET) { + // TODO: errno + return -1; + } + descriptor_turn_socket_into_listener(desc, backlog); + return 0; +} + +int sys_closesocket(SOCKET fd) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_SOCKET && + desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET + desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { + __builtin_trap(); + } + descriptor_free(desc); + return 0; +} + +SOCKET sys_accept(SOCKET fd, void *addr, int *addr_len) +{ + Descriptor *desc = ¤t_process->desc[(int) fd]; + if (desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET) { + // TODO: errno + return INVALID_SOCKET; + } + + if (desc->accept_queue_used == 0) { + // TODO: would block + } + PendingAccept *pending_accept = desc->accept_queue[desc->accept_queue_head]; + desc->accept_queue_head = (desc->accept_queue_head + 1) % (desc->accept_queue_size); + desc->accept_queue_used--; + + int new_fd = find_unused_descriptor(current_process); + if (new_fd < 0) { + // TODO + } + descriptor_init_as_socket(¤t_process->desc[new_fd]); + descriptor_turn_socket_into_connection(¤t_process->desc[new_fd], peer_process, peer_fd); + return (SOCKET) new_fd; +} + +int sys_getsockopt(SOCKET fd, int level, int optname, void *optval, int *optlen) +{ + // TODO +} + +int sys_setsockopt(SOCKET fd, int level, int optname, void *optval, int optlen) +{ + // TODO +} + +int sys_recv(SOCKET fd, void *dst, int len, int flags) +{ + // TODO +} + +int sys_send(SOCKET fd, void *src, int len, int flags) +{ + // TODO +} + +int sys_connect(SOCKET fd, void *addr, size_t addr_len) +{ + // TODO +} + +BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount) +{ + // TODO +} + +BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency) +{ + // TODO +} + +HANDLE sys_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess, + DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, + DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, + HANDLE hTemplateFile) +{ + int fd = find_unused_descriptor(current_process); + if (fd < 0) { + // TODO: WSAGetLastError + return INVALID_HANDLE_VALUE; + } + + HANDLE handle = CreateFileW(lpFileName, dwDesiredAccess, + dwShareMode, lpSecurityAttributes, dwCreationDisposition, + dwFlagsAndAttributes, hTemplateFile); + if (handle == INVALID_HANDLE_VALUE) { + // TODO + return INVALID_HANDLE_VALUE; + } + + Descriptor *desc = current_process->dt->desc[fd]; + descriptor_init_as_file(desc, handle); + return (HANDLE) fd; +} + +BOOL sys_CloseHandle(HANDLE handle) +{ + Descriptor *desc = current_process->dt->desc[(int) handle]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO + return false; + } + return CloseHandle(desc->real_fd); +} + +BOOL sys_LockFile(HANDLE handle) +{ + Descriptor *desc = current_process->dt->desc[(int) handle]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO + return false; + } + return LockFile(desc->real_fd); +} + +BOOL sys_UnlockFile(HANDLE handle) +{ + Descriptor *desc = current_process->dt->desc[(int) handle]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO + return false; + } + return UnlockFile(desc->real_fd); +} + +BOOL sys_FlushFileBuffers(HANDLE handle) +{ + Descriptor *desc = current_process->dt->desc[(int) handle]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO + return false; + } + return FlushFileBuffers(desc->real_fd); +} + +BOOL sys_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov) +{ + if (ov) { + // TODO + return false; + } + + Descriptor *desc = ¤t_process->dt->desc[(int) handle]; + switch (desc->type) { + + case DESCRIPTOR_TYPE_FILE: + return ReadFile(desc->real_fd, dst, len, num, ov); + + case DESCRIPTOR_TYPE_CONNECTION_SOCKET: + { + int cpy = len; + if (cpy > desc->input_used) + cpy = desc->input_used; + memcpy(dst, desc->input, cpy); + memmove(desc->input, desc->input + cpy, desc->input_used - cpy); + desc->input_used -= cpy; + *num = cpy; + return true; + } + break; + + default: + // TODO: errno + return false; + } +} + +BOOL sys_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov) +{ + if (ov) { + // TODO + return false; + } + + Descriptor *desc = ¤t_process->dt->desc[(int) handle]; + switch (desc->type) { + + case DESCRIPTOR_TYPE_FILE: + return WriteFile(desc->real_fd, src, len, num, ov); + + case DESCRIPTOR_TYPE_CONNECTION_SOCKET: + { + if (desc->output_size - desc->output_used < len) { + int new_capacity = 2 * desc->output_size; + if (new_capacity - desc->output_used < len) + new_capacity = desc->output_used + len; + desc->output = realloc(desc->output, new_capacity); + if (desc->output == NULL) + __builtin_trap(); + desc->output_size = new_capacity; + } + memcpy(desc->output + desc->output_used, src, len); + desc->output_used += len; + *num = len; + return true; + } + break; + + default: + // TODO: errno + return false; + } +} + +BOOL sys_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf) +{ + Descriptor *desc = current_process->dt->desc[(int) handle]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO + return false; + } + return GetFileSizeEx(desc->real_fd, buf); +} + +char *sys__fullpath(char *path, char *dst, int cap) +{ + return _fullpath(path, dst, cap); +} + +#else + +int sys_socket(int domain, int type, int protocol) +{ + if (domain != AF_INET) { + // TODO: errno + return -1; + } + + if (type != SOCK_STREAM) { + // TODO: errno + return -1; + } + + if (protocol != 0) { + // TODO: errno + return -1; + } + + int fd = find_unused_descriptor(current_process); + if (fd < 0) { + // TODO: errno + return -1; + } + + // TODO: maybe bind to a random address? + + descriptor_init_as_socket(¤t_process->dt->desc[fd]); + return fd; +} + +int sys_bind(int fd, void *addr, size_t addr_len) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { + // TODO: errno + return -1; + } + + if (addr_len != sizeof(desc->bind)) { + // TODO: errno + return -1; + } + + // TODO: maybe check that no one else is listening + // on this port + + desc->bound = true; + memcpy(&desc->addr, addr, addr_len); + return 0; +} + +int sys_listen(int fd, int backlog) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_SOCKET) { + // TODO: errno + return -1; + } + descriptor_turn_socket_into_listener(desc, backlog); + return 0; +} + +int sys_accept(int fd, void *addr, int *addr_len) +{ + Descriptor *desc = ¤t_process->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET) { + // TODO: errno + return -1; + } + + if (desc->accept_queue_used == 0) { + // TODO: would block + } + PendingAccept *pending_accept = desc->accept_queue[desc->accept_queue_head]; + desc->accept_queue_head = (desc->accept_queue_head + 1) % (desc->accept_queue_size); + desc->accept_queue_used--; + + int new_fd = find_unused_descriptor(current_process); + if (new_fd < 0) { + // TODO + } + descriptor_init_as_socket(¤t_process->desc[new_fd]); + descriptor_turn_socket_into_connection(¤t_process->desc[new_fd], peer_process, peer_fd); + return new_fd; +} + +int sys_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen) +{ + // TODO +} + +int sys_setsockopt(int fd, int level, int optname, void *optval, socklen_t optlen) +{ + // TODO +} + +int sys_recv(int fd, void *dst, int len, int flags) +{ + if (flags) + __builtin_trap(); + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { + // TODO: errno + return -1; + } + return sys_read(fd, dst, len); +} + +int sys_send(int fd, void *src, int len, int flags) +{ + if (flags) + __builtin_trap(); + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { + // TODO: errno + return -1; + } + return sys_write(fd, src, len); +} + +bool sockaddr_eql(struct sockaddr_in a, struct sockaddr_in b) +{ + return a.sin_family == b.sin_family + && a.sin_port == b.sin_port + && a.sin_addr == b.sin_addr; +} + +int sys_connect(int fd, void *addr, size_t addr_len) +{ + if (addr_len != sizeof(struct sockaddr_in)) { + // TODO + } + struct sockaddr_in tmp; + memcpy(&tmp, addr, sizeof(tmp)); + + int peer_process = -1; + int peer_fd = -1; + for (int i = 0; i < num_processes; i++) { + for (int j = 0; j < processes[i].dt->num_desc; j++) { + Descriptor *desc = &processes[i].dt->desc[j]; + if (desc->type == DESCRIPTOR_TYPE_LISTENER_SOCKET + && sockaddr_eql(desc->addr, tmp)) { + peer_process = i; + peer_fd = j; + goto found; + } + } + } +found: + if (peer_process < 0) { + // TODO + } + + int fd = find_unused_descriptor(current_process); + if (fd < 0) { + // TODO + } + + Descriptor *desc = ¤t_process->desc[fd]; + descriptor_init_as_socket(desc); + descriptor_turn_socket_into_connection(desc, peer_process, peer_fd); + return fd; +} + +int sys_clock_gettime(clockid_t clockid, struct timespec *tp) +{ + if (clockid != CLOCK_REALTIME) { + // TODO + } + + if (tp == NULL) { + // TODO + } + + tp->tv_sec = current_time / 1000000000; + tp->tv_nsec = current_time % 1000000000; + return 0; +} + +int sys_open(char *path, int flags, int mode) +{ + int fd = find_unused_descriptor(current_process); + if (fd < 0) { + // TODO: errno + return -1; + } + + int real_fd = open(path, flags, mode); + if (real_fd < 0) + return real_fd; + + descriptor_init_as_file(¤t_process->dt->desc[fd], real_fd); + return fd; +} + +int sys_close(int fd) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + descriptor_free(desc); + return 0; +} + +int sys_flock(int fd, int op) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO: errno + return -1; + } + + return flock(desc->real_fd, op); +} + +int sys_fsync(int fd) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO: errno + return -1; + } + + return fsync(desc->real_fd); +} + +int sys_read(int fd, char *dst, int len) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + switch (desc->type) { + + case DESCRIPTOR_TYPE_FILE: + return read(desc->real_fd, dst, len); + + case DESCRIPTOR_TYPE_CONNECTION_SOCKET: + { + int cpy = len; + if (cpy > desc->input_used) + cpy = desc->input_used; + memcpy(dst, desc->input, cpy); + memmove(desc->input, desc->input + cpy, desc->input_used - cpy); + desc->input_used -= cpy; + return cpy; + } + break; + + default: + // TODO: errno + return -1; + } +} + +int sys_write(int fd, char *src, int len) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + switch (desc->type) { + + case DESCRIPTOR_TYPE_FILE: + return write(desc->real_fd, dst, len); + + case DESCRIPTOR_TYPE_CONNECTION_SOCKET: + { + if (desc->output_size - desc->output_used < len) { + int new_capacity = 2 * desc->output_size; + if (new_capacity - desc->output_used < len) + new_capacity = desc->output_used + len; + desc->output = realloc(desc->output, new_capacity); + if (desc->output == NULL) + __builtin_trap(); + desc->output_size = new_capacity; + } + memcpy(desc->output + desc->output_used, src, len); + desc->output_used += len; + return len; + } + break; + + default: + // TODO: errno + return -1; + } +} + +int sys_stat(int fd, struct stat *buf) +{ + Descriptor *desc = ¤t_process->dt->desc[fd]; + if (desc->type != DESCRIPTOR_TYPE_FILE) { + // TODO: errno + return -1; + } + + return stat(desc->real_fd, buf); +} + +int sys_mkstemp(char *path) +{ + return mkstemp(path); +} + +char* sys_realpath(char *path, char *dst) +{ + return realpath(path, dst); +} + +#endif + +#endif // BUILD_TEST diff --git a/src/metadata_server.c b/src/metadata_server.c index 82981a3..902641b 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -16,7 +16,7 @@ static void hash_list_init(HashList *hash_list) static void hash_list_free(HashList *hash_list) { - free(hash_list->items); + sys_free(hash_list->items); } static int hash_list_insert(HashList *hash_list, SHA256 hash) @@ -30,7 +30,7 @@ static int hash_list_insert(HashList *hash_list, SHA256 hash) int new_capacity = hash_list->capacity ? hash_list->capacity * 2 : 16; - SHA256 *new_items = realloc(hash_list->items, new_capacity * sizeof(SHA256)); + SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256)); if (new_items == NULL) return -1; @@ -750,7 +750,7 @@ static bool is_chunk_server_message_type(uint16_t type) return false; } -int metadata_server_init(MetadataServer *state, int argc, char **argv) +int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled) { (void) argc; (void) argv; @@ -778,7 +778,7 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv) return -1; } - return 0; + return tcp_register_events(&state->tcp, contexts, polled); } int metadata_server_free(MetadataServer *state) @@ -788,10 +788,10 @@ int metadata_server_free(MetadataServer *state) return 0; } -int metadata_server_step(MetadataServer *state) +int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *polled, int num_polled) { Event events[MAX_CONNS+1]; - int num_events = tcp_process_events(&state->tcp, events); + int num_events = tcp_translate_events(&state->tcp, events, contexts, polled, num_polled); for (int i = 0; i < num_events; i++) { int conn_idx = events[i].conn_idx; @@ -852,5 +852,5 @@ int metadata_server_step(MetadataServer *state) } } - return 0; + return tcp_register_events(&state->tcp, contexts, polled); } diff --git a/src/metadata_server.h b/src/metadata_server.h index 120c224..ec381d5 100644 --- a/src/metadata_server.h +++ b/src/metadata_server.h @@ -50,6 +50,6 @@ typedef struct { int metadata_server_init(MetadataServer *state, int argc, char **argv); int metadata_server_free(MetadataServer *state); -int metadata_server_step(MetadataServer *state); +int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *polled, int num_polled); #endif // METADATA_SERVER_INCLUDED diff --git a/src/system.c b/src/system.c index 26de8cf..02a0274 100644 --- a/src/system.c +++ b/src/system.c @@ -1,68 +1,238 @@ -#ifndef BUILD_TEST - #include "system.h" -Socket sys_socket(int domain, int type, int protocol) +void *sys_malloc_(size_t len, char *file, int line) +{ + (void) file; + (void) line; + return malloc(len); +} + +void *sys_realloc_(void *ptr, size_t len, char *file, int line) +{ + (void) file; + (void) line; + return realloc(ptr, len); +} + +void sys_free_(void *ptr, char *file, int line) +{ + (void) file; + (void) line; + free(ptr); +} + +#ifdef _WIN32 + +SOCKET sys_socket(int domain, int type, int protocol) { return socket(domain, type, protocol); } -int sys_bind(Socket fd, void *addr, size_t addr_len) +int sys_bind(SOCKET fd, void *addr, size_t addr_len) { return bind(fd, addr, addr_len); } -int sys_listen(Socket fd, int backlog) +int sys_listen(SOCKET fd, int backlog) { return listen(fd, backlog); } -int sys_closesocket(Socket fd) +int sys_closesocket(SOCKET fd) { -#ifdef _WIN32 - closesocket(fd); -#else - close(fd); -#endif + return closesocket(fd); } -int sys_poll(struct pollfd *polled, int num_polled, int timeout) +int sys_WSAPoll(struct pollfd *polled, int num_polled, int timeout) { -#ifdef _WIN32 return WSAPoll(polled, num_polled, timeout); -#else - return poll(polled, num_polled, timeout); -#endif } -Socket sys_accept(Socket fd, void *addr, int *addr_len) +SOCKET sys_accept(SOCKET fd, void *addr, int *addr_len) { return accept(fd, addr, addr_len); } -int sys_getsockopt(Socket fd, int level, int optname, void *optval, socklen_t *optlen) +int sys_getsockopt(SOCKET fd, int level, int optname, void *optval, int *optlen) { - return getsockopt(fd, level, optname, optval, optlen); + return getsockopt(fd, level, optname, optval); } -int sys_setsockopt(Socket fd, int level, int optname, void *optval, socklen_t optlen) +int sys_setsockopt(SOCKET fd, int level, int optname, void *optval, int optlen) { return setsockopt(fd, level, optname, optval, optlen); } -int sys_recv(Socket fd, void *dst, int len, int flags) +int sys_recv(SOCKET fd, void *dst, int len, int flags) { return recv(fd, dst, len, flags); } -int sys_send(Socket fd, void *src, int len, int flags) +int sys_send(SOCKET fd, void *src, int len, int flags) { return send(fd, src, len, flags); } -int sys_connect(Socket fd, void *addr, size_t addr_len) +int sys_connect(SOCKET fd, void *addr, size_t addr_len) { return connect(fd, addr, addr_len); } -#endif // BUILD_TEST +BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount) +{ + return QueryPerformanceCounter(lpPerformanceCount); +} + +BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency) +{ + return QueryPerformanceFrequency(lpFrequency); +} + +HANDLE sys_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, HANDLE hTemplateFile) +{ + return CreateFileW(lpFileName, dwDesiredAccess, dwShareMode, lpSecurityAttributes, dwCreationDisposition, dwFlagsAndAttributes, hTemplateFile); +} + +BOOL sys_CloseHandle(HANDLE handle) +{ + return CloseHandle(handle); +} + +BOOL sys_LockFile(HANDLE handle) +{ + return LockFile(handle); +} + +BOOL sys_UnlockFile(HANDLE handle) +{ + return UnlockFile(handle); +} + +BOOL sys_FlushFileBuffers(HANDLE handle) +{ + return FlushFileBuffers(handle); +} + +BOOL sys_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov) +{ + return ReadFile(handle, dst, len, num, ov); +} + +BOOL sys_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov) +{ + return WriteFile(handle, src, len, num, ov); +} + +BOOL sys_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf) +{ + // TODO +} + +char *sys__fullpath(char *path, char *dst, int cap) +{ + return _fullpath(path, dst, cap); +} + +#else + +int sys_socket(int domain, int type, int protocol) +{ + return socket(domain, type, protocol); +} + +int sys_bind(int fd, void *addr, size_t addr_len) +{ + return bind(fd, addr, addr_len); +} + +int sys_listen(int fd, int backlog) +{ + return listen(fd, backlog); +} + +int sys_close(int fd) +{ + return close(fd); +} + +int sys_poll(struct pollfd *polled, int num_polled, int timeout) +{ + return poll(polled, num_polled, timeout); +} + +int sys_accept(int fd, void *addr, int *addr_len) +{ + return accept(fd, addr, addr_len); +} + +int sys_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen) +{ + return getsockopt(fd, level, optname, optval, optlen); +} + +int sys_setsockopt(int fd, int level, int optname, void *optval, socklen_t optlen) +{ + return setsockopt(fd, level, optname, optval, optlen); +} + +int sys_recv(int fd, void *dst, int len, int flags) +{ + return recv(fd, dst, len, flags); +} + +int sys_send(int fd, void *src, int len, int flags) +{ + return send(fd, src, len, flags); +} + +int sys_connect(int fd, void *addr, size_t addr_len) +{ + return connect(fd, addr, addr_len); +} + +int sys_clock_gettime(clockid_t clockid, struct timespec *tp) +{ + return clock_gettime(clockid, tp); +} + +int sys_open(char *path, int flags, int mode) +{ + return open(path, flags, mode); +} + +int sys_close(int fd) +{ + return close(fd); +} + +int sys_flock(int fd, int op) +{ + return flock(fd, op); +} + +int sys_fsync(int fd) +{ + return fsync(fd); +} + +int sys_read(int fd, char *dst, int len) +{ + return read(fd, dst, len); +} + +int sys_stat(int fd, struct stat *buf) +{ + return stat(fd, buf); +} + +int sys_mkstemp(char *path) +{ + return mkstemp(path); +} + +char* sys_realpath(char *path, char *dst) +{ + return realpath(path, dst); +} + +#endif diff --git a/src/system.h b/src/system.h index 370ea4e..953f298 100644 --- a/src/system.h +++ b/src/system.h @@ -1,36 +1,72 @@ -#ifndef SYSTEM_INCLUDED -#define SYSTEM_INCLUDED + +void *sys_malloc_ (size_t len, char *file, int line); +void *sys_realloc_(void *ptr, size_t len, char *file, int line); +void sys_free_ (void *ptr, char *file, int line); + +#define sys_malloc(len) sys_malloc_ ((len), __FILE__, __LINE__) +#define sys_realloc(ptr, len) sys_realloc_((ptr), (len), __FILE__, __LINE__) +#define sys_free(ptr) sys_free_ ((ptr), __FILE__, __LINE__) #ifdef _WIN32 -#include -#include + +typedef union _LARGE_INTEGER { + struct { + DWORD LowPart; + LONG HighPart; + } DUMMYSTRUCTNAME; + struct { + DWORD LowPart; + LONG HighPart; + } u; + LONGLONG QuadPart; +} LARGE_INTEGER; + +#define BOOL int +#define WCHAR wchar_t +#define SOCKET void* +#define HANDLE void* + +SOCKET sys_socket (int domain, int type, int protocol); +int sys_bind (SOCKET fd, void *addr, size_t addr_len); +int sys_listen (SOCKET fd, int backlog); +int sys_closesocket (SOCKET fd); +SOCKET sys_accept (SOCKET fd, void *addr, int *addr_len); +int sys_getsockopt (SOCKET fd, int level, int optname, void *optval, int *optlen); +int sys_setsockopt (SOCKET fd, int level, int optname, void *optval, int optlen); +int sys_recv (SOCKET fd, void *dst, int len, int flags); +int sys_send (SOCKET fd, void *src, int len, int flags); +int sys_connect (SOCKET fd, void *addr, size_t addr_len); +BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount); +BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency); +HANDLE sys_CreateFileW (WCHAR *lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, HANDLE hTemplateFile); +BOOL sys_CloseHandle (HANDLE handle); +BOOL sys_LockFile (HANDLE handle); +BOOL sys_UnlockFile (HANDLE handle); +BOOL sys_FlushFileBuffers (HANDLE handle); +BOOL sys_ReadFile (HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov); +BOOL sys_WriteFile (HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov); +BOOL sys_GetFileSizeEx (HANDLE handle, LARGE_INTEGER *buf); +char* sys__fullpath (char *path, char *dst, int cap); + #else -#include -#include -#include -#include -#define SOCKET int -#define INVALID_SOCKET -1 + +int sys_socket (int domain, int type, int protocol); +int sys_bind (int fd, void *addr, size_t addr_len); +int sys_listen (int fd, int backlog); +int sys_accept (int fd, void *addr, int *addr_len); +int sys_getsockopt (int fd, int level, int optname, void *optval, socklen_t *optlen); +int sys_setsockopt (int fd, int level, int optname, void *optval, socklen_t optlen); +int sys_recv (int fd, void *dst, int len, int flags); +int sys_send (int fd, void *src, int len, int flags); +int sys_connect (int fd, void *addr, size_t addr_len); +int sys_clock_gettime (clockid_t clockid, struct timespec *tp); +int sys_open (char *path, int flags, int mode); +int sys_close (int fd); +int sys_flock (int fd, int op); +int sys_fsync (int fd); +int sys_read (int fd, char *dst, int len); +int sys_stat (int fd, struct stat *buf); +int sys_mkstemp (char *path); +char* sys_realpath (char *path, char *dst); + #endif - -#if !defined(_WIN32) || defined(BUILD_TEST) -typedef int Socket; -#define BAD_SOCKET ((Socket) -1) -#else -typedef SOCKET Socket; -#define BAD_SOCKET INVALID_SOCKET -#endif - -Socket sys_socket (int domain, int type, int protocol); -int sys_bind (Socket fd, void *addr, size_t addr_len); -int sys_listen (Socket fd, int backlog); -int sys_closesocket (Socket fd); -int sys_poll (struct pollfd *polled, int num_polled, int timeout); -Socket sys_accept (Socket fd, void *addr, int *addr_len); -int sys_getsockopt (Socket fd, int level, int optname, void *optval, socklen_t *optlen); -int sys_setsockopt (Socket fd, int level, int optname, void *optval, socklen_t optlen); -int sys_recv (Socket fd, void *dst, int len, int flags); -int sys_send (Socket fd, void *src, int len, int flags); -int sys_connect (Socket fd, void *addr, size_t addr_len); - -#endif // SYSTEM_INCLUDED diff --git a/src/tcp.c b/src/tcp.c index cd46e38..0522381 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -147,11 +147,8 @@ void tcp_consume_message(TCP *tcp, int conn_idx) tcp->conns[conn_idx].msglen = 0; } -// The "events" array must be an array of capacity MAX_CONNS+1 -int tcp_process_events(TCP *tcp, Event *events) +int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled) { - struct pollfd polled[MAX_CONNS + 1]; - void *contexts[MAX_CONNS + 1]; int num_polled = 0; if (tcp->listen_fd != INVALID_SOCKET && tcp->num_conns < MAX_CONNS) { @@ -173,8 +170,12 @@ int tcp_process_events(TCP *tcp, Event *events) } } - sys_poll(polled, num_polled, -1); + return 0; +} +// The "events" array must be an array of capacity MAX_CONNS+1 +int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled) +{ bool removed[MAX_CONNS+1]; int num_events = 0; diff --git a/src/tcp.h b/src/tcp.h index d726120..8b47caf 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -59,7 +59,8 @@ int tcp_index_from_tag(TCP *tcp, int tag); int tcp_listen(TCP *tcp, char *addr, uint16_t port); int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type); void tcp_consume_message(TCP *tcp, int conn_idx); -int tcp_process_events(TCP *tcp, Event *events); +int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled); +int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled); ByteQueue *tcp_output_buffer(TCP *tcp, int conn_idx); int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output); void tcp_close(TCP *tcp, int conn_idx); diff --git a/src/test.c b/src/test.c deleted file mode 100644 index 7d8157e..0000000 --- a/src/test.c +++ /dev/null @@ -1,464 +0,0 @@ -#ifdef BUILD_TEST - -#include -#include -#include "system.h" -#include "chunk_server.h" -#include "metadata_server.h" -#include "byte_queue.h" - -#define MAX_DESCRIPTORS 1024 -#define MAX_PENDING_CONNECTIONS 128 - -typedef enum { - DESCRIPTOR_TYPE_EMPTY, - DESCRIPTOR_TYPE_FILE, - DESCRIPTOR_TYPE_PRECONF_SOCKET, - DESCRIPTOR_TYPE_LISTEN_SOCKET, - DESCRIPTOR_TYPE_CONNECTION_SOCKET, -} DescriptorType; - -typedef struct { - - DescriptorType type; - - // Socket fields - struct sockaddr_in bind; - bool no_bind; - - // Listen socket fields - int backlog; - int pending_connections[MAX_PENDING_CONNECTIONS]; - int num_pending; - - // Data socket fields - int peer_process; // Index of the peer process (-1 if not connected) - int peer_fd; // Descriptor in peer process - ByteQueue input; - ByteQueue output; - -} Descriptor; - -typedef struct { - int num_desc; - Descriptor desc[MAX_DESCRIPTORS]; -} Process; - -#define MAX_PROCESSES (MAX_CHUNK_SERVERS + 1) - -MetadataServer metadata_server; - -int num_chunk_servers = 0; -ChunkServer chunk_servers[MAX_CHUNK_SERVERS]; - -int num_processes = 0; -Process processes[MAX_PROCESSES]; -Process *current_process; - -Socket sys_socket(int domain, int type, int protocol) -{ - if (domain != AF_INET) { - // TODO: errno - return BAD_SOCKET; - } - - if (type != SOCK_STREAM) { - // TODO: errno - return BAD_SOCKET; - } - - if (protocol != 0) { - // TODO: errno - return BAD_SOCKET; - } - - if (num_processes == MAX_PROCESSES) { - // TODO: errno - return BAD_SOCKET; - } - - Socket fd = 0; - while (fd < MAX_DESCRIPTORS && current_process->desc[fd].type != DESCRIPTOR_TYPE_EMPTY) - fd++; - - if (fd >= MAX_DESCRIPTORS) { - // TODO: errno - return BAD_SOCKET; - } - - current_process->desc[fd].type = DESCRIPTOR_TYPE_PRECONF_SOCKET; - current_process->desc[fd].no_bind = true; - current_process->desc[fd].num_pending = 0; - current_process->desc[fd].peer_process = -1; - current_process->desc[fd].peer_fd = -1; - return fd; -} - -int sys_bind(Socket fd, void *addr, size_t addr_len) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { - // TODO: errno - return -1; - } - - if (addr_len != sizeof(current_process->desc[fd].bind)) { - // TODO: errno - return -1; - } - - // TODO: maybe check that no one else is listening - // on this port - - current_process->desc[fd].no_bind = false; - memcpy(¤t_process->desc[fd].bind, addr, addr_len); - return 0; -} - -int sys_listen(Socket fd, int backlog) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { - // TODO: errno - return -1; - } - - current_process->desc[fd].type = DESCRIPTOR_TYPE_LISTEN_SOCKET; - current_process->desc[fd].backlog = backlog; - return 0; -} - -int sys_closesocket(Socket fd) -{ - current_process->desc[fd].type = DESCRIPTOR_TYPE_EMPTY; - return 0; -} - -int sys_poll(struct pollfd *polled, int num_polled, int timeout) -{ - int num = 0; - for (int i = 0; i < num_polled; i++) { - - polled[i].revents = 0; - - Socket fd = polled[i].fd; - if (fd < 0) - continue; - - if (polled[i].events & POLLNVAL) { - // TODO - } - - if (polled[i].events & POLLERR) { - // TODO - } - - if (polled[i].events & POLLHUP) { - // TODO - } - - if (polled[i].events & POLLIN) { - if (!byte_queue_empty(¤t_process->desc[fd].input)) - polled[i].revents |= POLLIN; - } - - if (polled[i].events & POLLOUT) { - if (!byte_queue_full(¤t_process->desc[fd].output)) - polled[i].revents |= POLLOUT; - } - - if (polled[i].revents) - num++; - } - - return num; -} - -Socket sys_accept(Socket fd, void *addr, int *addr_len) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_LISTEN_SOCKET) { - // TODO: errno - return BAD_SOCKET; - } - - if (current_process->desc[fd].num_pending == 0) { - // No pending connections - would block in real system - // TODO: errno (EAGAIN/EWOULDBLOCK) - return BAD_SOCKET; - } - - // Get the first pending connection descriptor - Socket new_fd = current_process->desc[fd].pending_connections[0]; - - // Remove from pending queue - current_process->desc[fd].num_pending--; - for (int i = 0; i < current_process->desc[fd].num_pending; i++) { - current_process->desc[fd].pending_connections[i] = - current_process->desc[fd].pending_connections[i + 1]; - } - - // Fill in peer address if requested - if (addr != NULL && addr_len != NULL) { - int peer_process = current_process->desc[new_fd].peer_process; - int peer_fd = current_process->desc[new_fd].peer_fd; - if (peer_process > -1 && peer_fd > -1) { - struct sockaddr_in *peer_addr = (struct sockaddr_in *)addr; - *peer_addr = processes[peer_process].desc[peer_fd].bind; - *addr_len = sizeof(struct sockaddr_in); - } - } - - return new_fd; -} - -int sys_getsockopt(Socket fd, int level, int optname, void *optval, socklen_t *optlen) -{ - if (level != SOL_SOCKET) { - // TODO: errno - return -1; - } - - switch (optname) { - - case SO_ERROR: - { - if (optlen == NULL || *optlen < sizeof(int)) { - // TODO: errno - return -1; - } - - if (optval == NULL) { - // TODO: errno - return -1; - } - - // In our simulation, all connections succeed immediately - // and we don't track socket errors, so always return 0 - *(int *)optval = 0; - *optlen = sizeof(int); - return 0; - } - - default: - // Unsupported socket option - // TODO: errno (ENOPROTOOPT) - return -1; - } -} - -int sys_setsockopt(Socket fd, int level, int optname, void *optval, socklen_t optlen) -{ - // TODO: errno - return -1; -} - -int sys_recv(Socket fd, void *dst, int len, int flags) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { - // TODO: errno - return -1; - } - - ByteQueue *input = ¤t_process->desc[fd].input; - ByteView buf = byte_queue_read_buf(input); - if (buf.len > len) - buf.len = len; - memcpy(dst, buf.ptr, buf.len); - byte_queue_read_ack(input, buf.len); - - return buf.len; -} - -int sys_send(Socket fd, void *src, int len, int flags) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { - // TODO: errno - return -1; - } - - ByteQueue *output = ¤t_process->desc[fd].output; - ByteView buf = byte_queue_write_buf(output); - if (buf.len > len) - buf.len = len; - memcpy(buf.ptr, src, buf.len); - byte_queue_write_ack(output, buf.len); - - return buf.len; -} - -int sys_connect(Socket fd, void *addr, size_t addr_len) -{ - if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { - // TODO: errno - return -1; - } - - if (addr_len != sizeof(struct sockaddr_in)) { - // TODO: errno - return -1; - } - - struct sockaddr_in *target_addr = (struct sockaddr_in *)addr; - - // Find the process with a listen socket on this address - int target_process_idx = -1; - Socket target_listen_fd = BAD_SOCKET; - - for (int i = 0; i < num_processes; i++) { - for (int j = 0; j < MAX_DESCRIPTORS; j++) { - Descriptor *desc = &processes[i].desc[j]; - if (desc->type == DESCRIPTOR_TYPE_LISTEN_SOCKET && !desc->no_bind) { - if (desc->bind.sin_port == target_addr->sin_port && - (desc->bind.sin_addr.s_addr == target_addr->sin_addr.s_addr || - desc->bind.sin_addr.s_addr == INADDR_ANY)) { - target_process_idx = i; - target_listen_fd = j; - goto found; - } - } - } - } - -found: - if (target_process_idx < 0) { - // No listener found - connection refused - // TODO: errno (ECONNREFUSED) - return -1; - } - - // Create a new socket in the target process for the accepted connection - Socket accept_fd = 0; - while (accept_fd < MAX_DESCRIPTORS && - processes[target_process_idx].desc[accept_fd].type != DESCRIPTOR_TYPE_EMPTY) - accept_fd++; - - if (accept_fd >= MAX_DESCRIPTORS) { - // TODO: errno - return -1; - } - - // Check if pending queue is full - if (processes[target_process_idx].desc[target_listen_fd].num_pending >= MAX_PENDING_CONNECTIONS) { - // TODO: errno (ECONNREFUSED or EAGAIN) - return -1; - } - - // Initialize byte queues for both ends of the connection - byte_queue_init(¤t_process->desc[fd].input, 1<<16); - byte_queue_init(¤t_process->desc[fd].output, 1<<16); - byte_queue_init(&processes[target_process_idx].desc[accept_fd].input, 1<<16); - byte_queue_init(&processes[target_process_idx].desc[accept_fd].output, 1<<16); - - // Set up the client socket - current_process->desc[fd].type = DESCRIPTOR_TYPE_CONNECTION_SOCKET; - int current_process_idx = current_process - processes; - current_process->desc[fd].peer_process = target_process_idx; - current_process->desc[fd].peer_fd = accept_fd; - - // Set up the accepted socket - processes[target_process_idx].desc[accept_fd].type = DESCRIPTOR_TYPE_CONNECTION_SOCKET; - processes[target_process_idx].desc[accept_fd].no_bind = false; - processes[target_process_idx].desc[accept_fd].bind = *target_addr; - processes[target_process_idx].desc[accept_fd].peer_process = current_process_idx; - processes[target_process_idx].desc[accept_fd].peer_fd = fd; - processes[target_process_idx].desc[accept_fd].num_pending = 0; - - // Add to pending connections queue - processes[target_process_idx].desc[target_listen_fd].pending_connections[ - processes[target_process_idx].desc[target_listen_fd].num_pending++] = accept_fd; - - // Connection succeeds immediately in simulation (no EINPROGRESS needed) - return 0; -} - -int main(void) -{ - num_processes = 8; - - if (num_processes == 0) { - // TODO - return -1; - } - - current_process = &processes[0]; - int ret = metadata_server_init(&metadata_server, 0, NULL); - if (ret) { - // TODO - return -1; - } - - for (int i = 0; i < num_processes-1; i++) { - current_process = &processes[i+1]; - ret = chunk_server_init(&chunk_servers[i], 0, NULL); - if (ret) { - // TODO - return -1; - } - } - - for (;;) { - // Step each process - for (int i = 0; i < num_processes; i++) { - - current_process = &processes[i]; - - int ret; - if (i == 0) ret = metadata_server_step(&metadata_server); - else ret = chunk_server_step(&chunk_servers[i-1]); - - if (ret) { - // TODO: handle error - } - } - - // Process network transit: move data from output queues to peer input queues - for (int i = 0; i < num_processes; i++) { - for (int fd = 0; fd < MAX_DESCRIPTORS; fd++) { - Descriptor *desc = &processes[i].desc[fd]; - - if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) - continue; - - if (desc->peer_process < 0 || desc->peer_fd < 0) - continue; - - // Get peer descriptor - Descriptor *peer = &processes[desc->peer_process].desc[desc->peer_fd]; - - // Transfer data from this socket's output to peer's input - ByteView output_data = byte_queue_read_buf(&desc->output); - if (output_data.len > 0) { - // Get available space in peer's input queue - ByteView peer_input_space = byte_queue_write_buf(&peer->input); - - // Transfer as much as possible - size_t transfer_size = output_data.len; - if (transfer_size > peer_input_space.len) - transfer_size = peer_input_space.len; - - if (transfer_size > 0) { - memcpy(peer_input_space.ptr, output_data.ptr, transfer_size); - byte_queue_write_ack(&peer->input, transfer_size); - byte_queue_read_ack(&desc->output, transfer_size); - } - } - } - } - } - - for (int i = 0; i < num_processes-1; i++) { - ret = chunk_server_free(&chunk_servers[i]); - if (ret) { - // TODO - return -1; - } - } - - ret = metadata_server_free(&metadata_server); - if (ret) { - // TODO - return -1; - } - - return 0; -} - -#endif // BUILD_TEST