From 7012f582d26702bde9138583d7fc162b5ca2e04a Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 12 Nov 2025 12:13:52 +0100 Subject: [PATCH 01/21] Add chunk management notes to DESIGN.txt --- DESIGN.txt | 53 ++++++++++++++++------ src/message.h | 1 + src/metadata_server.c | 100 +++++++++++++++++++++++++----------------- 3 files changed, 100 insertions(+), 54 deletions(-) diff --git a/DESIGN.txt b/DESIGN.txt index 7f4447d..accc156 100644 --- a/DESIGN.txt +++ b/DESIGN.txt @@ -137,26 +137,51 @@ Chunk Management: servers need to forget some copies. If they are under-replicated, some chunk servers need to copy chunks from elsewhere. - Event 1: A chunk server connects - If the chunk server is holding some chunks, for each chunk - one of the following must be true: - - The chunk is not used by the file system - => Mark the chunk for removal - - The chunk is used and, with this copy, perfectly replicated - => Do nothing - - The chunk is used and under-replicated - => ??? - - The chunk is used and over-replicated - => Mark the chunk for removal + The metadata server holds an authoritative list of hashes that each + chunk server needs to hold. These lists are enforced periodically and + automatically during the state updates, therefore the metadata server + only needs to reorganize the chunks in its in-memory lists. The changes + will propagate over the chunk servers in time. The metadata server's + in-memory representation is the ideal state the system is converging + to, which means every chunk if always replicated correctly. Given this + setup, a chunk can only be under-replicated when there are not enough + chunk servers. - Event 2: A write operation on metadata (adding chunks) + Case 1: The first chunk server connects with no chunks + CS connects + MS accepts + CS sends auth message + MS validates auth + MS requests list of chunks + CS sends the chunk list + MS notices it's empty + + Case 2: The first chunk server connects with some chunks + CS connects + MS accepts + CS sends auth message + MS valudates auth + MS requests list of chunks + CS sends the chunk list + MS iterates over each chunk. For each chunk it determines + whether it's useless, under-replicated, over-replicated, + or properly replicated. + MS adds all useless or over-replicated chunks to the rem_list + of the newly connected chunk server. Note that it's only + possible for a chunk to be over-replicated by 1, so removing + it from the new chunk server will fix the issue. + MS If a chunk is under-replicated, it means that there are not + enough chunk servers to hold all the replicas, so there is + nothing to be done. + + Case 3: Chunk server with some chunks disconnects TODO Event 3: A chunk server disconnects A number of chunks are lost, but there is a chance the server will come back in a short while (10s or so). - - TODO + => Wait for about 10s. If the chunk server doesn't come back, distribute + its hashes to other chunk servers Event 4: A write operation on metadata occurs (overwriting old chunks) Assuming the write overwrites some chunks: diff --git a/src/message.h b/src/message.h index 5e943d3..ddaf786 100644 --- a/src/message.h +++ b/src/message.h @@ -38,6 +38,7 @@ enum { // Chunk server -> Metadata server MESSAGE_TYPE_AUTH, + MESSAGE_TYPE_CHUNK_LIST, MESSAGE_TYPE_STATE_UPDATE_ERROR, MESSAGE_TYPE_STATE_UPDATE_SUCCESS, diff --git a/src/metadata_server.c b/src/metadata_server.c index f230cf1..879a111 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -768,55 +768,49 @@ static int process_chunk_server_auth(MetadataServer *state, if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - // Read IPv4s - { - uint32_t num_ipv4; - if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) + uint32_t num_ipv4; + if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) + return -1; + + for (uint32_t i = 0; i < num_ipv4; i++) { + + IPv4 ipv4; + if (!binary_read(&reader, &ipv4, sizeof(ipv4))) return -1; - for (uint32_t i = 0; i < num_ipv4; i++) { + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) + return -1; - IPv4 ipv4; - if (!binary_read(&reader, &ipv4, sizeof(ipv4))) - return -1; - - uint16_t port; - if (!binary_read(&reader, &port, sizeof(port))) - return -1; - - if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { - Address addr = {0}; - addr.ipv4 = ipv4; - addr.is_ipv4 = true; - addr.port = port; - chunk_server->addrs[chunk_server->num_addrs++] = addr; - } + if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { + Address addr = {0}; + addr.ipv4 = ipv4; + addr.is_ipv4 = true; + addr.port = port; + chunk_server->addrs[chunk_server->num_addrs++] = addr; } } - // Read IPv6s - { - uint32_t num_ipv6; - if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) + uint32_t num_ipv6; + if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) + return -1; + + for (uint32_t i = 0; i < num_ipv6; i++) { + + IPv6 ipv6; + if (!binary_read(&reader, &ipv6, sizeof(ipv6))) return -1; - for (uint32_t i = 0; i < num_ipv6; i++) { + uint16_t port; + if (!binary_read(&reader, &port, sizeof(port))) + return -1; - IPv6 ipv6; - if (!binary_read(&reader, &ipv6, sizeof(ipv6))) - return -1; - - uint16_t port; - if (!binary_read(&reader, &port, sizeof(port))) - return -1; - - if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { - Address addr = {0}; - addr.ipv6 = ipv6; - addr.is_ipv4 = false; - addr.port = port; - chunk_server->addrs[chunk_server->num_addrs++] = addr; - } + if (chunk_server->num_addrs < MAX_SERVER_ADDRS) { + Address addr = {0}; + addr.ipv6 = ipv6; + addr.is_ipv4 = false; + addr.port = port; + chunk_server->addrs[chunk_server->num_addrs++] = addr; } } @@ -833,9 +827,32 @@ static int process_chunk_server_auth(MetadataServer *state, // we accept all connections that provide valid address information. chunk_server->auth = true; + // TODO: Request the chunk list held by this chunk server + return 0; } +static int process_chunk_server_chunk_list(MetadataServer *state, + int conn_idx, ByteView msg) +{ + // Iterate over each chunk held by this chunk + // server and determine whether it's useless, + // under-replicated, over-replicated, or + // properly replicated. + // Chunks that are useless or over-replicated + // are added to the remove list of this chunk + // server (note that any given chunk can only + // be over-replicated by 1, so removing it from + // this chunk server will fix the issue). If + // the chunk is under-replicated or properly + // replicated, do nothing. Note that it's only + // possible for a chunk to be under-replicated + // when the number of chunk servers is lower + // than the replication factor. + + // TODO +} + static int process_chunk_server_state_update_success(MetadataServer *state, int conn_idx, ByteView msg) { @@ -948,6 +965,9 @@ process_chunk_server_message(MetadataServer *state, case MESSAGE_TYPE_AUTH: return process_chunk_server_auth(state, conn_idx, msg); + case MESSAGE_TYPE_CHUNK_LIST: + return process_chunk_server_chunk_list(state, conn_idx, msg); + case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: return process_chunk_server_state_update_success(state, conn_idx, msg); From e4c5390ca717ae2a93c802528f76731a9dda37d0 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 00:43:18 +0100 Subject: [PATCH 02/21] Implement CHUNK_LIST endpoints on the metadata server --- src/message.h | 1 + src/metadata_server.c | 76 +++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 74 insertions(+), 3 deletions(-) diff --git a/src/message.h b/src/message.h index ddaf786..30fc0be 100644 --- a/src/message.h +++ b/src/message.h @@ -35,6 +35,7 @@ enum { // Metadata server -> Chunk server MESSAGE_TYPE_STATE_UPDATE, MESSAGE_TYPE_DOWNLOAD_LOCATIONS, + MESSAGE_TYPE_CHUNK_LIST_REQUEST, // Chunk server -> Metadata server MESSAGE_TYPE_AUTH, diff --git a/src/metadata_server.c b/src/metadata_server.c index 879a111..f1db7d2 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -827,7 +827,13 @@ static int process_chunk_server_auth(MetadataServer *state, // we accept all connections that provide valid address information. chunk_server->auth = true; - // TODO: Request the chunk list held by this chunk server + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_CHUNK_LIST_REQUEST); + if (!message_writer_free(&writer)) + return -1; return 0; } @@ -835,6 +841,12 @@ static int process_chunk_server_auth(MetadataServer *state, static int process_chunk_server_chunk_list(MetadataServer *state, int conn_idx, ByteView msg) { + int chunk_server_idx = tcp_get_tag(&state->tcp, conn_idx); + assert(chunk_server_idx > -1); + assert(chunk_server_idx < MAX_CHUNK_SERVERS); + assert(state->chunk_servers[chunk_server_idx].used); + ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + // Iterate over each chunk held by this chunk // server and determine whether it's useless, // under-replicated, over-replicated, or @@ -850,7 +862,65 @@ static int process_chunk_server_chunk_list(MetadataServer *state, // when the number of chunk servers is lower // than the replication factor. - // TODO + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + // version + if (!binary_read(&reader, NULL, sizeof(uint16_t))) + return -1; + + uint16_t type; + if (!binary_read(&reader, &type, sizeof(type))) + return -1; + + if (type != MESSAGE_TYPE_CHUNK_LIST) + return -1; + + // length + if (!binary_read(&reader, NULL, sizeof(uint32_t))) + return -1; + + uint32_t num_hashes; + if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) + return -1; + + for (uint32_t i = 0; i < num_hashes; i++) { + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) + return -1; + + bool drop = false; + if (!file_tree_uses_hash(&state->file_tree, hash)) + drop = true; + else { + + int holders[MAX_CHUNK_SERVERS]; + int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); + assert(num_holders > -1); + assert(num_holders <= MAX_CHUNK_SERVERS); + + assert(num_holders <= state->replication_factor); + + // Adding this chunk server as a holder will cause + // the chunk to be over-replicated + if (num_holders == state->replication_factor) + drop = true; + } + + HashList *list; + if (drop) list = &chunk_server->rem_list; + else list = &chunk_server->add_list; + + if (hash_list_insert(list, hash) < 0) { + assert(0); // TODO + } + } + + if (binary_read(&reader, NULL, 1)) + return -1; + + // No need to reply here + return 0; } static int process_chunk_server_state_update_success(MetadataServer *state, @@ -862,7 +932,7 @@ static int process_chunk_server_state_update_success(MetadataServer *state, // Merge add_list into old_list for (int i = 0; i < chunk_server->add_list.count; i++) { if (!hash_list_contains(&chunk_server->old_list, chunk_server->add_list.items[i])) - hash_list_insert(&chunk_server->old_list, chunk_server->add_list.items[i]); + hash_list_insert(&chunk_server->old_list, chunk_server->add_list.items[i]); // TODO: what if this fails? } // Clear add_list and rem_list From 88189c669348e258c9f16874ad982f46d009ea6d Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 01:09:01 +0100 Subject: [PATCH 03/21] Draft of the chunk server endpoint that returns the list of all chunks --- src/chunk_server.c | 94 ++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/src/chunk_server.c b/src/chunk_server.c index 6b48271..b443eeb 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -4,10 +4,12 @@ #include #include "basic.h" +#include "byte_queue.h" #include "config.h" #include "sha256.h" #include "message.h" #include "file_system.h" +#include "tcp.h" #include "chunk_server.h" static void @@ -562,6 +564,95 @@ process_metadata_server_download_locations(ChunkServer *state, int conn_idx, Byt return 0; } +static int +process_metadata_server_chunk_list_request(ChunkServer *state, int conn_idx, ByteView msg) +{ + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + // version + if (!binary_read(&reader, NULL, sizeof(uint16_t))) + return -1; + + // type + if (!binary_read(&reader, NULL, sizeof(uint16_t))) + return -1; + + // length + if (!binary_read(&reader, NULL, sizeof(uint32_t))) + return -1; + + if (binary_read(&reader, NULL, 1)) + return 1; + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_CHUNK_LIST); + + // Open the folder of chunks and write all hashes + // to the metadata server. First, write the number + // of hashes as a u32 integer, then that number + // of hashes. + // If the number is not known ahead of time, write + // a dummy value and then patch it later. + + ByteQueueOffset offset = byte_queue_offset(writer.output); + + uint32_t num_hashes = 0; // Dummy value + message_write(&writer, &num_hashes, sizeof(num_hashes)); + +#ifdef _WIN32 + WIN32_FIND_DATA find_data; + HANDLE handle = sys_FindFirstFileA(path, &find_data); + if (handle == INVALID_HANDLE_VALUE) { + if (sys_GetLastError() == ERROR_FILE_NOT_FOUND) { + // TODO + } + return -1; + } + + do { + + SHA256 hash; + + // TODO + + message_write(&writer, &hash, sizeof(hash)); + num_hashes++; + + } while (sys_FindNextFileA(handle, &find_data)); + + if (sys_GetLastError() != ERROR_NO_MORE_FILES) + return -1; + + sys_FindClose(handle); +#else + DIR *d = sys_opendir(path); + if (d == NULL) + return -1; + + struct dirent *e; + while ((e = sys_readdir(d))) { + + SHA256 hash; + + // TODO + + message_write(&writer, &hash, sizeof(hash)); + num_hashes++; + } + + sys_closedir(d); +#endif + + byte_queue_patch(writer.output, offset, &num_hashes, sizeof(num_hashes)); + + if (!message_writer_free(&writer)) + return -1; + return 0; +} + static int process_metadata_server_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) { @@ -572,6 +663,9 @@ process_metadata_server_message(ChunkServer *state, int conn_idx, uint16_t type, case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: return process_metadata_server_download_locations(state, conn_idx, msg); + + case MESSAGE_TYPE_CHUNK_LIST_REQUEST: + return process_metadata_server_chunk_list_request(state, conn_idx, msg); } return -1; From 83c2ccb2cabc5e2f85bdac3c4deb310dacad765a Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 01:35:15 +0100 Subject: [PATCH 04/21] Implement DirectoryScanner object --- src/chunk_server.c | 54 ++++++++++------------------ src/file_system.c | 89 ++++++++++++++++++++++++++++++++++++++++++++++ src/file_system.h | 18 ++++++++++ 3 files changed, 125 insertions(+), 36 deletions(-) diff --git a/src/chunk_server.c b/src/chunk_server.c index b443eeb..8d020a7 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -223,6 +223,7 @@ static bool chunk_store_exists(ChunkStore *store, SHA256 hash) string path = hash2path(store, hash, buf); // Try to open the file to check if it exists + // TODO: this isn't right Handle fd; if (file_open(path, &fd) == 0) { file_close(fd); @@ -602,49 +603,30 @@ process_metadata_server_chunk_list_request(ChunkServer *state, int conn_idx, Byt uint32_t num_hashes = 0; // Dummy value message_write(&writer, &num_hashes, sizeof(num_hashes)); -#ifdef _WIN32 - WIN32_FIND_DATA find_data; - HANDLE handle = sys_FindFirstFileA(path, &find_data); - if (handle == INVALID_HANDLE_VALUE) { - if (sys_GetLastError() == ERROR_FILE_NOT_FOUND) { - // TODO + DirectoryScanner scanner; + if (directory_scanner_init(&scanner, xxx) < 0) + return -1; + + for (;;) { + + string name; + int ret = directory_scanner_next(&scanner, &name); + + if (ret == 1) + break; + + if (ret < 0) { + directory_scanner_free(&scanner); + return -1; } - return -1; - } - - do { SHA256 hash; - - // TODO - - message_write(&writer, &hash, sizeof(hash)); - num_hashes++; - - } while (sys_FindNextFileA(handle, &find_data)); - - if (sys_GetLastError() != ERROR_NO_MORE_FILES) - return -1; - - sys_FindClose(handle); -#else - DIR *d = sys_opendir(path); - if (d == NULL) - return -1; - - struct dirent *e; - while ((e = sys_readdir(d))) { - - SHA256 hash; - - // TODO + // TODO: file name to hash message_write(&writer, &hash, sizeof(hash)); num_hashes++; } - - sys_closedir(d); -#endif + directory_scanner_free(&scanner); byte_queue_patch(writer.output, offset, &num_hashes, sizeof(num_hashes)); diff --git a/src/file_system.c b/src/file_system.c index c64f630..91cef4a 100644 --- a/src/file_system.c +++ b/src/file_system.c @@ -268,3 +268,92 @@ int file_read_all(string path, string *data) file_close(fd); return 0; } + +#ifdef _WIN32 + +int directory_scanner_init(DirectoryScanner *scanner, string path) +{ + char pattern[PATH_MAX]; + int ret = snprintf(pattern, "%.*s\\*", path.len, path.ptr); + if (ret < 0 || ret >= sizeof(pattern)) + return -1; + + scanner->handle = sys_FindFirstFileA(path, &scanner->find_data); + if (scanner->handle == INVALID_HANDLE_VALUE) { + if (sys_GetLastError() == ERROR_FILE_NOT_FOUND) { + scanner->done = true; + return 0; + } + return -1; + } + + scanner->done = false; + scanner->first = true; + return 0; +} + +int directory_scanner_next(DirectoryScanner *scanner, string *name) +{ + if (scanner->done) + return 1; + + if (!scanner->first) { + BOOL ok = sys_FindNextFileA(scanner->handle, &scanner->find_data); + if (!ok) { + scanner->done = true; + if (sys_GetLastError() == ERROR_NO_MORE_FILES) + return 1; + return -1; + } + } + + char *p = scanner->find_data.cFileName;// TODO: is this the right field? + *name = (string) { p, strlen(p) }; + return 0; +} + +void directory_scanner_free(DirectoryScanner *scanner) +{ + sys_FindClose(scanner->handle); +} + +#else + +int directory_scanner_init(DirectoryScanner *scanner, string path) +{ + char path_copy[PATH_MAX]; + if (path.len >= PATH_MAX) + return -1; + memcpy(path_copy, path.ptr, path.len); + path_copy[path.len] = '\0'; + + scanner->d = sys_opendir(path_copy); + if (scanner->d == NULL) { + scanner->done = true; + return -1; + } + + return 0; +} + +int directory_scanner_next(DirectoryScanner *scanner, string *name) +{ + if (scanner->done) + return -1; + + scanner->e = sys_reddir(scanner->d); + if (scanner->e == NULL) { + scanner->done = true; + return 1; + } + + *name = (string) { scanner->e->d_name, strlen(scanner->e->d_name) }; + return 0; +} + +void directory_scanner_free(DirectoryScanner *scanner) +{ + sys_closedir(scanner->d); +} + +#endif diff --git a/src/file_system.h b/src/file_system.h index 125388d..f86ec2f 100644 --- a/src/file_system.h +++ b/src/file_system.h @@ -7,6 +7,20 @@ typedef struct { uint64_t data; } Handle; +#ifdef _WIN32 +typedef struct { + HANDLE handle; + WIN32_FIND_DATA find_data; + bool first; + bool done; +} DirectoryScanner; +#else +typedef struct { + DIR *d; + struct dirent *e; +} DirectoryScanner; +#endif + int file_open(string path, Handle *fd); void file_close(Handle fd); int file_lock(Handle fd); @@ -22,4 +36,8 @@ int remove_file_or_dir(string path); int get_full_path(string path, char *dst); int file_read_all(string path, string *data); +int directory_scanner_init(DirectoryScanner *scanner, string path); +int directory_scanner_next(DirectoryScanner *scanner, string *name); +void directory_scanner_free(DirectoryScanner *scanner); + #endif // FILE_SYSTEM_INCLUDED From c90356730980d5e273e1b1785cb22583baf1a9fc Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 00:44:31 +0000 Subject: [PATCH 05/21] Add mocks for file search system calls (Windows and Linux) Added mock implementations for file/directory search operations to support both Windows and Linux in the simulation environment. Windows mocks: - FindFirstFileA: Wraps real search handle in descriptor - FindNextFileA: Forwards to real API - FindClose: Properly cleans up search handles Linux mocks: - opendir: Wraps real DIR* handle in descriptor - readdir: Forwards to real API - closedir: Properly cleans up directory handles Common changes: - Added DESC_DIRECTORY descriptor type to track directory handles - Added real_d field to Descriptor for directory handles (HANDLE on Windows, DIR* on Linux) - Updated close_desc to handle directory cleanup on both platforms - Added sys_* macros for both BUILD_TEST and non-BUILD_TEST modes - Added include for Linux directory operations --- src/system.c | 157 +++++++++++++++++++++++++++++++++++++++++++++++++++ src/system.h | 19 +++++++ 2 files changed, 176 insertions(+) diff --git a/src/system.c b/src/system.c index a98a999..be02667 100644 --- a/src/system.c +++ b/src/system.c @@ -31,6 +31,7 @@ typedef enum { DESC_SOCKET, DESC_LISTEN_SOCKET, DESC_CONNECTION_SOCKET, + DESC_DIRECTORY, } DescriptorType; typedef enum { @@ -84,6 +85,14 @@ typedef struct { NATIVE_HANDLE real_fd; + // ------ Directory ------------- + +#ifdef _WIN32 + HANDLE real_d; +#else + DIR *real_d; +#endif + // ------ Socket ---------------- // Events reported by the last "poll" call @@ -1236,6 +1245,14 @@ static void close_desc(Descriptor *desc) } // TODO break; + + case DESC_DIRECTORY: +#ifdef _WIN32 + FindClose(desc->real_d); +#else + closedir(desc->real_d); +#endif + break; } desc->type = DESC_EMPTY; desc->generation++; @@ -1503,6 +1520,71 @@ int mock_ioctlsocket(SOCKET fd, long cmd, u_long *argp) return -1; } +HANDLE mock_FindFirstFileA(char *lpFileName, WIN32_FIND_DATAA *lpFindFileData) +{ + HANDLE handle = FindFirstFileA(lpFileName, lpFindFileData); + if (handle == INVALID_HANDLE_VALUE) + return INVALID_HANDLE_VALUE; + + if (current_process->num_desc == MAX_DESCRIPTORS) { + FindClose(handle); + SetLastError(ERROR_TOO_MANY_OPEN_FILES); + return INVALID_HANDLE_VALUE; + } + + int idx = 0; + while (current_process->desc[idx].type != DESC_EMPTY) { + idx++; + assert(idx < MAX_DESCRIPTORS); + } + + Descriptor *desc = ¤t_process->desc[idx]; + desc->type = DESC_DIRECTORY; + desc->real_d = handle; + + current_process->num_desc++; + CHECK_NON_EMPTY_DESC_INVARIANT; + return (HANDLE) idx; +} + +BOOL mock_FindNextFileA(HANDLE hFindFile, WIN32_FIND_DATAA *lpFindFileData) +{ + if (hFindFile == INVALID_HANDLE_VALUE || (int)hFindFile < 0 || (int)hFindFile >= MAX_DESCRIPTORS) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + int idx = (int) hFindFile; + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + + // Forward to real FindNextFileA, last error is set by the real call + return FindNextFileA(desc->real_d, lpFindFileData); +} + +BOOL mock_FindClose(HANDLE hFindFile) +{ + if (hFindFile == INVALID_HANDLE_VALUE || (int)hFindFile < 0 || (int)hFindFile >= MAX_DESCRIPTORS) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + int idx = (int) hFindFile; + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + SetLastError(ERROR_INVALID_HANDLE); + return FALSE; + } + + close_desc(desc); + current_process->num_desc--; + CHECK_NON_EMPTY_DESC_INVARIANT; + return TRUE; +} + #else int mock_clock_gettime(clockid_t clockid, struct timespec *tp) @@ -1736,6 +1818,81 @@ int mock_fcntl(int fd, int cmd, ...) return -1; } +DIR *mock_opendir(char *name) +{ + DIR *dir = opendir(name); + if (dir == NULL) + return NULL; + + if (current_process->num_desc == MAX_DESCRIPTORS) { + closedir(dir); + errno = EMFILE; // Too many open files + return NULL; + } + + int idx = 0; + while (current_process->desc[idx].type != DESC_EMPTY) { + idx++; + assert(idx < MAX_DESCRIPTORS); + } + + Descriptor *desc = ¤t_process->desc[idx]; + desc->type = DESC_DIRECTORY; + desc->real_d = dir; + + current_process->num_desc++; + CHECK_NON_EMPTY_DESC_INVARIANT; + return (DIR *)(intptr_t) idx; +} + +struct dirent *mock_readdir(DIR *dirp) +{ + if (dirp == NULL) { + errno = EBADF; // Bad file descriptor + return NULL; + } + int idx = (int)(intptr_t) dirp; + + if (idx < 0 || idx >= MAX_DESCRIPTORS) { + errno = EBADF; + return NULL; + } + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + errno = EBADF; + return NULL; + } + + // Forward to real readdir, errno is set by the real call + return readdir(desc->real_d); +} + +int mock_closedir(DIR *dirp) +{ + if (dirp == NULL) { + errno = EBADF; // Bad file descriptor + return -1; + } + int idx = (int)(intptr_t) dirp; + + if (idx < 0 || idx >= MAX_DESCRIPTORS) { + errno = EBADF; + return -1; + } + + Descriptor *desc = ¤t_process->desc[idx]; + if (desc->type != DESC_DIRECTORY) { + errno = EBADF; + return -1; + } + + close_desc(desc); + current_process->num_desc--; + CHECK_NON_EMPTY_DESC_INVARIANT; + return 0; +} + #endif // !_WIN32 #endif // BUILD_TEST diff --git a/src/system.h b/src/system.h index aeea168..31ba995 100644 --- a/src/system.h +++ b/src/system.h @@ -21,6 +21,7 @@ #include #include #include +#include #include #include #include @@ -69,6 +70,9 @@ BOOL mock_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount); BOOL mock_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency); char* mock__fullpath(char *path, char *dst, int cap); int mock__mkdir(char *path); +HANDLE mock_FindFirstFileA(char *lpFileName, WIN32_FIND_DATAA *lpFindFileData); +BOOL mock_FindNextFileA(HANDLE hFindFile, WIN32_FIND_DATAA *lpFindFileData); +BOOL mock_FindClose(HANDLE hFindFile); #else int mock_clock_gettime(clockid_t clockid, struct timespec *tp); int mock_open(char *path, int flags, int mode); @@ -82,6 +86,9 @@ int mock_mkstemp(char *path); char* mock_realpath(char *path, char *dst); int mock_mkdir(char *path, mode_t mode); int mock_fcntl(int fd, int cmd, ...); +DIR* mock_opendir(char *name); +struct dirent* mock_readdir(DIR *dirp); +int mock_closedir(DIR *dirp); #endif // Common @@ -114,6 +121,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys__fullpath mock__fullpath #define sys_QueryPerformanceCounter mock_QueryPerformanceCounter #define sys_QueryPerformanceFrequency mock_QueryPerformanceFrequency +#define sys_FindFirstFileA mock_FindFirstFileA +#define sys_FindNextFileA mock_FindNextFileA +#define sys_FindClose mock_FindClose // Linux #define sys_mkdir mock_mkdir @@ -129,6 +139,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys_clock_gettime mock_clock_gettime #define sys_fcntl mock_fcntl #define sys_getrandom mock_getrandom +#define sys_opendir mock_opendir +#define sys_readdir mock_readdir +#define sys_closedir mock_closedir #else @@ -162,6 +175,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys__fullpath _fullpath #define sys_QueryPerformanceCounter QueryPerformanceCounter #define sys_QueryPerformanceFrequency QueryPerformanceFrequency +#define sys_FindFirstFileA FindFirstFileA +#define sys_FindNextFileA FindNextFileA +#define sys_FindClose FindClose // Linux #define sys_mkdir mkdir @@ -176,6 +192,9 @@ int mock_fcntl(int fd, int cmd, ...); #define sys_realpath realpath #define sys_clock_gettime clock_gettime #define sys_fcntl fcntl +#define sys_opendir opendir +#define sys_readdir readdir +#define sys_closedir closedir #endif #endif // SYSTEM_INCLUDED From 6c2ea4bce90b9673325e11249b558fdc8c31f26f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 00:45:51 +0000 Subject: [PATCH 06/21] Fix DirectoryScanner implementation bugs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixed several bugs in the DirectoryScanner implementation: Windows fixes: - Fixed snprintf call missing buffer size parameter - Fixed FindFirstFileA to use pattern (with wildcard) instead of path - Changed sys_GetLastError() to GetLastError() - Added scanner->first = false after using first result - Removed TODO comment from cFileName usage (it's the correct field) Linux fixes: - Fixed typo: sys_reddir → sys_readdir - Added missing 'done' field to DirectoryScanner struct - Initialize scanner->done = false on successful opendir - Fixed return value when scanner->done (should return 1, not -1) --- src/file_system.c | 17 ++++++++++------- src/file_system.h | 1 + 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/src/file_system.c b/src/file_system.c index 91cef4a..8f23a6c 100644 --- a/src/file_system.c +++ b/src/file_system.c @@ -274,13 +274,13 @@ int file_read_all(string path, string *data) int directory_scanner_init(DirectoryScanner *scanner, string path) { char pattern[PATH_MAX]; - int ret = snprintf(pattern, "%.*s\\*", path.len, path.ptr); + int ret = snprintf(pattern, sizeof(pattern), "%.*s\\*", path.len, path.ptr); if (ret < 0 || ret >= sizeof(pattern)) return -1; - scanner->handle = sys_FindFirstFileA(path, &scanner->find_data); + scanner->handle = sys_FindFirstFileA(pattern, &scanner->find_data); if (scanner->handle == INVALID_HANDLE_VALUE) { - if (sys_GetLastError() == ERROR_FILE_NOT_FOUND) { + if (GetLastError() == ERROR_FILE_NOT_FOUND) { scanner->done = true; return 0; } @@ -301,13 +301,15 @@ int directory_scanner_next(DirectoryScanner *scanner, string *name) BOOL ok = sys_FindNextFileA(scanner->handle, &scanner->find_data); if (!ok) { scanner->done = true; - if (sys_GetLastError() == ERROR_NO_MORE_FILES) + if (GetLastError() == ERROR_NO_MORE_FILES) return 1; return -1; } + } else { + scanner->first = false; } - char *p = scanner->find_data.cFileName;// TODO: is this the right field? + char *p = scanner->find_data.cFileName; *name = (string) { p, strlen(p) }; return 0; } @@ -333,15 +335,16 @@ int directory_scanner_init(DirectoryScanner *scanner, string path) return -1; } + scanner->done = false; return 0; } int directory_scanner_next(DirectoryScanner *scanner, string *name) { if (scanner->done) - return -1; + return 1; - scanner->e = sys_reddir(scanner->d); + scanner->e = sys_readdir(scanner->d); if (scanner->e == NULL) { scanner->done = true; return 1; diff --git a/src/file_system.h b/src/file_system.h index f86ec2f..19b9c67 100644 --- a/src/file_system.h +++ b/src/file_system.h @@ -18,6 +18,7 @@ typedef struct { typedef struct { DIR *d; struct dirent *e; + bool done; } DirectoryScanner; #endif From 14cd2f1ac644ccc206eb8c70cd7f0783593a4791 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 00:47:56 +0000 Subject: [PATCH 07/21] Fix compilation issues Fixed compilation errors: - Added include to file_system.h for DIR type on Linux - Replaced xxx placeholder in chunk_server.c with actual chunks_dir path using state->store.path --- src/chunk_server.c | 3 ++- src/file_system.h | 4 ++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/src/chunk_server.c b/src/chunk_server.c index 8d020a7..288599e 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -604,7 +604,8 @@ process_metadata_server_chunk_list_request(ChunkServer *state, int conn_idx, Byt message_write(&writer, &num_hashes, sizeof(num_hashes)); DirectoryScanner scanner; - if (directory_scanner_init(&scanner, xxx) < 0) + string chunks_dir = (string) { state->store.path, strlen(state->store.path) }; + if (directory_scanner_init(&scanner, chunks_dir) < 0) return -1; for (;;) { diff --git a/src/file_system.h b/src/file_system.h index 19b9c67..a6ad0de 100644 --- a/src/file_system.h +++ b/src/file_system.h @@ -3,6 +3,10 @@ #include "basic.h" +#ifndef _WIN32 +#include +#endif + typedef struct { uint64_t data; } Handle; From 4ec47e66b42da546bf942d6d6229f3d21e835c42 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 12:09:29 +0100 Subject: [PATCH 08/21] Clean up chunk management section of DESIGN.txt --- DESIGN.txt | 98 +++++++++++++++++++++++++----------------------------- TODO.txt | 6 ++++ 2 files changed, 52 insertions(+), 52 deletions(-) diff --git a/DESIGN.txt b/DESIGN.txt index accc156..f45a8dd 100644 --- a/DESIGN.txt +++ b/DESIGN.txt @@ -137,62 +137,56 @@ Chunk Management: servers need to forget some copies. If they are under-replicated, some chunk servers need to copy chunks from elsewhere. - The metadata server holds an authoritative list of hashes that each - chunk server needs to hold. These lists are enforced periodically and - automatically during the state updates, therefore the metadata server - only needs to reorganize the chunks in its in-memory lists. The changes - will propagate over the chunk servers in time. The metadata server's - in-memory representation is the ideal state the system is converging - to, which means every chunk if always replicated correctly. Given this - setup, a chunk can only be under-replicated when there are not enough - chunk servers. + Metadata server variables for a chunk server: + ms_old_list: List of chunks that are known to be held by CS + ms_add_list: List of chunks that should be held by CS + ms_rem_list: List of chunks that may be held by CS but should removed from it - Case 1: The first chunk server connects with no chunks - CS connects - MS accepts - CS sends auth message - MS validates auth - MS requests list of chunks - CS sends the chunk list - MS notices it's empty + Chunk server variables: + cs_add_list: List of chunks added since the last update + cs_rem_list: List of chunks marked for removal after a timeout - Case 2: The first chunk server connects with some chunks - CS connects - MS accepts - CS sends auth message - MS valudates auth - MS requests list of chunks - CS sends the chunk list - MS iterates over each chunk. For each chunk it determines - whether it's useless, under-replicated, over-replicated, - or properly replicated. - MS adds all useless or over-replicated chunks to the rem_list - of the newly connected chunk server. Note that it's only - possible for a chunk to be over-replicated by 1, so removing - it from the new chunk server will fix the issue. - MS If a chunk is under-replicated, it means that there are not - enough chunk servers to hold all the replicas, so there is - nothing to be done. + Metadata change for write: + When clients commit a write by adding new hashes to the metadata, + MS adds those hashes to the ms_add_list for the CS where the client + uploaded the chunks. If a hash was overwritten and became useless, + that hash is added to the ms_rem_list for all CS holding it. - Case 3: Chunk server with some chunks disconnects - TODO + Metadata change for delete: + All hashes that are no longer reachable by the file tree are added + to the ms_rem_list of their holders - Event 3: A chunk server disconnects - A number of chunks are lost, but there is a chance the server will come back - in a short while (10s or so). - => Wait for about 10s. If the chunk server doesn't come back, distribute - its hashes to other chunk servers + Chunk upload: + When a chunk is uploaded to a chunk server, its hash is added to + the cs_add_list. - Event 4: A write operation on metadata occurs (overwriting old chunks) - Assuming the write overwrites some chunks: - - The chunks may still be referenced by some other file or section of the same file - => Do nothing - - The chunks are not used anymore - => Find all chunk servers holding the chunk and mark them for removal + Periodically on the chunk server: + Elements in the cs_rem_list have a 30 minute timeout after which they are deleted + permanently. - Event 5: A delete operation on metadata occurs - Do the same as event 4 + Periodically: + CS sends cs_add_list to MS + MS may add a subset of cs_add_list to ms_add_list based on the chunk replication and distribution policy + MS sends ms_add_list and ms_rem_list to CS + CS (1) Adds all elements from ms_rem_list to cs_rem_list + (2) Adds elements in cs_add_list that are not in ms_add_list to cs_rem_list + (3) Removes elements that are in ms_add_list and in cs_rem_list from cs_rem_list + (4) Elements in ms_add_list that are not held by the chunk server are added to + a temporary list tmp_list are sent to the metadata server + MS (1) Receives tmp_list and sends download locations to CS for those chunks + (2) Moves elements from ms_add_list that are not in tmp_list into ms_old_list + (3) Sets ms_add_list equal to tmp_list - Event 6: A chunk is corrupted or removed forcefully from the chunk server - The chunk server adds the chunk name to a list of lost chunks and - sends them to the metadata server at the next periodic update. + Chunk replication and distribution policy: + During an update, when CS reports a new chunk to MS, MS has to decide whether + to allow the CS to keep it or not. + + There are 4 cases: + - The chunk is useless (not referenced by any file) + - The chunk is under-replicated even counting the new copy + - The chunk is properly replicated with the new copy + - The chunk is over-replicated with the new copy + + If the chunk is useless, it is added to the cs_rem_list. If the chunk is + over-replicated, it is added to the cs_rem_list of this chunk or any other + holder of that chunk. Otherwise, add the chunk to the cs_add_list. diff --git a/TODO.txt b/TODO.txt index 8061155..23e1421 100644 --- a/TODO.txt +++ b/TODO.txt @@ -8,3 +8,9 @@ - find a way to remove over-replicated chunks and formalize the chunk management policy - add logging of: - when chunk servers connect and disconnect to the metadata server +- Should list scenarios that need testing, like those where chunks would be dropped + +Roadmap: + [ ] Complete all endpoints + [ ] Add fault injections to the simulation + [ ] Implement WAL From facd220c59521b51e2f6ead7ad8dd205fc1b31e0 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 12:18:34 +0100 Subject: [PATCH 09/21] Fix corner case in DESIGN.txt --- DESIGN.txt | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/DESIGN.txt b/DESIGN.txt index f45a8dd..ef16de9 100644 --- a/DESIGN.txt +++ b/DESIGN.txt @@ -145,6 +145,7 @@ Chunk Management: Chunk server variables: cs_add_list: List of chunks added since the last update cs_rem_list: List of chunks marked for removal after a timeout + cs_lst_list: List of chunks that were lost due to errors or forceful removals of chunk files Metadata change for write: When clients commit a write by adding new hashes to the metadata, @@ -172,7 +173,9 @@ Chunk Management: (2) Adds elements in cs_add_list that are not in ms_add_list to cs_rem_list (3) Removes elements that are in ms_add_list and in cs_rem_list from cs_rem_list (4) Elements in ms_add_list that are not held by the chunk server are added to - a temporary list tmp_list are sent to the metadata server + a temporary list tmp_list + (5) Elements in cs_lst_list are added to tmp_list + (6) tmp_list is sent to MS MS (1) Receives tmp_list and sends download locations to CS for those chunks (2) Moves elements from ms_add_list that are not in tmp_list into ms_old_list (3) Sets ms_add_list equal to tmp_list From cdce267768b5760bca69b05163c4b05a3fc4ab13 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 17:27:00 +0100 Subject: [PATCH 10/21] Update DESIGN.txt and code accordingly --- DESIGN.txt | 32 +++-- TODO.txt | 2 + src/chunk_server.c | 288 +++++++++++++------------------------- src/chunk_server.h | 38 +++-- src/message.h | 8 +- src/metadata_server.c | 315 ++++++++++++++++-------------------------- src/metadata_server.h | 15 +- 7 files changed, 278 insertions(+), 420 deletions(-) diff --git a/DESIGN.txt b/DESIGN.txt index ef16de9..aefba33 100644 --- a/DESIGN.txt +++ b/DESIGN.txt @@ -170,14 +170,14 @@ Chunk Management: MS may add a subset of cs_add_list to ms_add_list based on the chunk replication and distribution policy MS sends ms_add_list and ms_rem_list to CS CS (1) Adds all elements from ms_rem_list to cs_rem_list - (2) Adds elements in cs_add_list that are not in ms_add_list to cs_rem_list - (3) Removes elements that are in ms_add_list and in cs_rem_list from cs_rem_list - (4) Elements in ms_add_list that are not held by the chunk server are added to + (2) Elements in ms_add_list that are not held by the chunk server are added to a temporary list tmp_list - (5) Elements in cs_lst_list are added to tmp_list + (3) Removes elements in ms_add_list from cs_add_list and cs_rem_list, then merges cs_add_list into cs_rem_list and clears cs_add_list. + (4) Elements in cs_lst_list are added to tmp_list, then cs_lst_list is cleared. (6) tmp_list is sent to MS + (7) cs_add_list is cleared MS (1) Receives tmp_list and sends download locations to CS for those chunks - (2) Moves elements from ms_add_list that are not in tmp_list into ms_old_list + (2) Merges ms_add_list into ms_old_list, then removes all items in tmp_list from ms_old_list (3) Sets ms_add_list equal to tmp_list Chunk replication and distribution policy: @@ -190,6 +190,22 @@ Chunk Management: - The chunk is properly replicated with the new copy - The chunk is over-replicated with the new copy - If the chunk is useless, it is added to the cs_rem_list. If the chunk is - over-replicated, it is added to the cs_rem_list of this chunk or any other - holder of that chunk. Otherwise, add the chunk to the cs_add_list. + If the chunk is not referenced by the file tree, do nothing. + + If the chunk is properly replicated or under-replicated, add it to the ms_add_list. + + If the chunk is over-replicated, either don't add it to the ms_add_list or add it to the ms_rem_list of some other holder. + + TODO: The way chunk servers tell about the chunks they are holding + recently changed. Now instead of sending a full chunk list when + connecting, they send batches of chunks to the metadata server + during state updates. It also changed that now chunk servers + initiate updates. + + When a chunk server connects (and authenticates) + it sends alongside the auth message the hash of + the hash list of all chunks. The metadata server + then replies with a message saying whether it + already cached that chunk list or not. If it didn't, the chunk server sends the entire list + in chunks during state updates, with an increased + update frequency. diff --git a/TODO.txt b/TODO.txt index 23e1421..e495686 100644 --- a/TODO.txt +++ b/TODO.txt @@ -9,6 +9,8 @@ - add logging of: - when chunk servers connect and disconnect to the metadata server - Should list scenarios that need testing, like those where chunks would be dropped +- Update DESIGN.txt and the code to remove the chunk list message. The information of chunks + held by chunk servers is now transmitted to the metadata server during state updates Roadmap: [ ] Complete all endpoints diff --git a/src/chunk_server.c b/src/chunk_server.c index 288599e..b484766 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -321,139 +321,92 @@ static void start_download_if_necessary(ChunkServer *state) } } -static int -process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView msg) +static int process_metadata_server_sync_2(ChunkServer *state, + int conn_idx, ByteView msg) { BinaryReader reader = { msg.ptr, msg.len, 0 }; - // Read header if (!binary_read(&reader, NULL, sizeof(MessageHeader))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); + return -1; uint32_t add_count; if (!binary_read(&reader, &add_count, sizeof(add_count))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); + return -1; + + HashList tmp_list; + hash_list_init(&tmp_list); + + for (uint32_t i = 0; i < add_count; i++) { + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) + return -1; + + // Elements in ms_add_list that are not held by the + // chunk server are added to a temporary list tmp_list + + if (chunk_store_exists(&state->store, hash) < 0) { // TODO: does it return a boolean? + if (hash_list_insert(&tmp_list, hash) < 0) { + assert(0); // TODO + } + continue; + } + + hash_list_remove(&state->cs_rem_list, hash); + hash_list_remove(&state->cs_add_list, hash); + } + + if (hash_list_merge(&state->cs_rem_list, state->cs_add_list) < 0) { + assert(0); // TODO + } + + hash_list_clear(&state->cs_add_list); uint32_t rem_count; 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 = sys_malloc(add_count * sizeof(SHA256)); - if (add_list == NULL) - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); - - SHA256 *rem_list = sys_malloc(rem_count * sizeof(SHA256)); - if (rem_list == NULL) { - sys_free(add_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))) { - sys_free(add_list); - sys_free(rem_list); - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - } - } + return -1; for (uint32_t i = 0; i < rem_count; i++) { - if (!binary_read(&reader, &rem_list[i], sizeof(SHA256))) { - 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)) { - sys_free(add_list); - sys_free(rem_list); - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); - } - - // Check that all items in the add_list are in the chunk directory - // Any hashes that are missing are added to a missing list - SHA256 *missing = NULL; - uint32_t num_missing = 0; - - Time current_time = get_current_time(); - - for (uint32_t i = 0; i < add_count; i++) { - // If chunk is in removal list, unmark it (remove from removal list) - removal_list_remove(&state->removal_list, add_list[i]); - - // Check if chunk exists in the chunk store - if (!chunk_store_exists(&state->store, add_list[i])) { - - // Chunk is missing, add to missing list - - if (missing == NULL) { - missing = sys_malloc(add_count * sizeof(SHA256)); - if (missing == NULL) { - assert(0); // TODO - } - } - missing[num_missing++] = add_list[i]; - } - } - - // Append items from the rem_list to the removal list with timestamps - for (uint32_t i = 0; i < rem_count; i++) { - if (removal_list_add(&state->removal_list, rem_list[i], current_time) < 0) { - sys_free(add_list); - sys_free(rem_list); - sys_free(missing); - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); - } - } - - sys_free(add_list); - sys_free(rem_list); - - // Respond to the metadata server - if (num_missing == 0) { - - // No missing chunks, send success - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_SUCCESS); - if (!message_writer_free(&writer)) + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - } else { - - // Some chunks are missing, send error with missing list - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_ERROR); - - // Write error message - string error_msg = S("Missing chunks"); - uint16_t error_len = (uint16_t)error_msg.len; - message_write(&writer, &error_len, sizeof(error_len)); - message_write(&writer, error_msg.ptr, error_msg.len); - - // Write missing count and missing hashes - uint32_t tmp = num_missing; - message_write(&writer, &tmp, sizeof(tmp)); - message_write(&writer, missing, num_missing * sizeof(SHA256)); - - sys_free(missing); - - if (!message_writer_free(&writer)) - return -1; + if (timed_hash_list_insert(&state->cs_rem_list, hash, current_time) < 0) { + assert(0); // TODO + } } + if (binary_read(&reader, NULL, 1)) + return -1; + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_3); + + uint32_t count = tmp_list.count + state->cs_lst_list.count; // TODO: overflow + message_write(&writer, &count, sizeof(count)); + + for (uint32_t i = 0; i < tmp_list.count; i++) { + SHA256 hash = tmp_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + for (uint32_t i = 0; i < state->cs_lst_list.count; i++) { + SHA256 hash = state->cs_lst_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; } static int -process_metadata_server_download_locations(ChunkServer *state, int conn_idx, ByteView msg) +process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) { (void) conn_idx; @@ -565,92 +518,13 @@ process_metadata_server_download_locations(ChunkServer *state, int conn_idx, Byt return 0; } -static int -process_metadata_server_chunk_list_request(ChunkServer *state, int conn_idx, ByteView msg) -{ - BinaryReader reader = { msg.ptr, msg.len, 0 }; - - // version - if (!binary_read(&reader, NULL, sizeof(uint16_t))) - return -1; - - // type - if (!binary_read(&reader, NULL, sizeof(uint16_t))) - return -1; - - // length - if (!binary_read(&reader, NULL, sizeof(uint32_t))) - return -1; - - if (binary_read(&reader, NULL, 1)) - return 1; - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_CHUNK_LIST); - - // Open the folder of chunks and write all hashes - // to the metadata server. First, write the number - // of hashes as a u32 integer, then that number - // of hashes. - // If the number is not known ahead of time, write - // a dummy value and then patch it later. - - ByteQueueOffset offset = byte_queue_offset(writer.output); - - uint32_t num_hashes = 0; // Dummy value - message_write(&writer, &num_hashes, sizeof(num_hashes)); - - DirectoryScanner scanner; - string chunks_dir = (string) { state->store.path, strlen(state->store.path) }; - if (directory_scanner_init(&scanner, chunks_dir) < 0) - return -1; - - for (;;) { - - string name; - int ret = directory_scanner_next(&scanner, &name); - - if (ret == 1) - break; - - if (ret < 0) { - directory_scanner_free(&scanner); - return -1; - } - - SHA256 hash; - // TODO: file name to hash - - message_write(&writer, &hash, sizeof(hash)); - num_hashes++; - } - directory_scanner_free(&scanner); - - byte_queue_patch(writer.output, offset, &num_hashes, sizeof(num_hashes)); - - if (!message_writer_free(&writer)) - return -1; - return 0; -} - static int process_metadata_server_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) { switch (type) { - - case MESSAGE_TYPE_STATE_UPDATE: - return process_metadata_server_state_update(state, conn_idx, msg); - - case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: - return process_metadata_server_download_locations(state, conn_idx, msg); - - case MESSAGE_TYPE_CHUNK_LIST_REQUEST: - return process_metadata_server_chunk_list_request(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_2: return process_metadata_server_sync_2(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_4: return process_metadata_server_sync_4(state, conn_idx, msg); } - return -1; } @@ -961,6 +835,26 @@ start_connecting_to_metadata_server(ChunkServer *state) return 0; } +static int send_sync_message(ChunkServer *state) +{ + assert(state->disconnect_time == INVALID_TIME); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC); + + uint32_t count = state->cs_add_list.count; // TODO: check implicit conversions + message_write(&writer, &count, sizeof(count)); + + for (uint32_t i = 0; i < count; i++) { + SHA256 hash = state->cs_add_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; +} + int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout) { string addr = getargs(argc, argv, "--addr", "127.0.0.1"); @@ -1148,6 +1042,14 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } + // TODO: periodically send a SYNC message + bool should_sync = false; + if (should_sync) { + if (send_sync_message(state) < 0) { + assert(0); // TODO + } + } + // TODO: periodically look for chunks that have their hashes messed up and delete them // Periodically retry pending downloads diff --git a/src/chunk_server.h b/src/chunk_server.h index 9e13331..b80d495 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -3,6 +3,7 @@ #include +#include "metadata_server.h" #include "tcp.h" #define TAG_METADATA_SERVER 1 @@ -27,25 +28,42 @@ typedef struct { typedef struct { SHA256 hash; - Time marked_time; -} PendingRemoval; + Time time; +} TimedHash; typedef struct { int count; int capacity; - PendingRemoval *items; -} RemovalList; + TimedHash *items; +} TimedHashList; typedef struct { - bool trace; - Address local_addr; - Address remote_addr; - Time disconnect_time; - TCP tcp; + + bool trace; + + Address local_addr; + + Address remote_addr; + + Time disconnect_time; + + TCP tcp; + ChunkStore store; + bool downloading; + PendingDownloadList pending_download_list; - RemovalList removal_list; + + // List of chunks added since the last update + HashList cs_add_list; + + // List of chunks marked for removal after a timeout + TimedHashList cs_rem_list; + + // List of chunks that were lost due to errors or forceful removals of chunk files + HashList cs_lst_list; + } ChunkServer; int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout); diff --git a/src/message.h b/src/message.h index 30fc0be..9d8ba83 100644 --- a/src/message.h +++ b/src/message.h @@ -33,15 +33,13 @@ enum { MESSAGE_TYPE_WRITE_SUCCESS, // Metadata server -> Chunk server - MESSAGE_TYPE_STATE_UPDATE, + MESSAGE_TYPE_SYNC_2, + MESSAGE_TYPE_SYNC_4, MESSAGE_TYPE_DOWNLOAD_LOCATIONS, - MESSAGE_TYPE_CHUNK_LIST_REQUEST, // Chunk server -> Metadata server MESSAGE_TYPE_AUTH, - MESSAGE_TYPE_CHUNK_LIST, - MESSAGE_TYPE_STATE_UPDATE_ERROR, - MESSAGE_TYPE_STATE_UPDATE_SUCCESS, + MESSAGE_TYPE_SYNC, // Chunk server -> Client MESSAGE_TYPE_CREATE_CHUNK_ERROR, diff --git a/src/metadata_server.c b/src/metadata_server.c index f1db7d2..c1f49e6 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -1,3 +1,8 @@ +#include "basic.h" +#include "byte_queue.h" +#include "config.h" +#include "file_tree.h" +#include "tcp.h" #define _GNU_SOURCE #include @@ -59,18 +64,18 @@ static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_t chunk_server->used = true; chunk_server->auth = false; chunk_server->num_addrs = 0; - hash_list_init(&chunk_server->old_list); - hash_list_init(&chunk_server->add_list); - hash_list_init(&chunk_server->rem_list); + hash_list_init(&chunk_server->ms_old_list); + hash_list_init(&chunk_server->ms_add_list); + hash_list_init(&chunk_server->ms_rem_list); chunk_server->last_sync_time = current_time; chunk_server->last_response_time = current_time; } static void chunk_server_peer_free(ChunkServerPeer *chunk_server) { - hash_list_free(&chunk_server->rem_list); - hash_list_free(&chunk_server->add_list); - hash_list_free(&chunk_server->old_list); + hash_list_free(&chunk_server->ms_rem_list); + hash_list_free(&chunk_server->ms_add_list); + hash_list_free(&chunk_server->ms_old_list); chunk_server->used = false; } @@ -284,6 +289,7 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg) if (binary_read(&reader, NULL, 1)) return -1; + // TODO: return unused hashes and add them to the ms_rem_list of holder chunk servers int ret = file_tree_delete_entity(&state->file_tree, path); if (ret < 0) { @@ -671,7 +677,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) int k = find_chunk_server_by_addr(state, results[i].addrs[j]); if (k == -1) return -1; - if (hash_list_insert(&state->chunk_servers[k].add_list, new_hashes[i]) < 0) + if (hash_list_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0) return -1; } } @@ -684,7 +690,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) // Add to rem_list for all chunk servers that have this chunk for (int j = 0; j < state->num_chunk_servers; j++) { if (chunk_server_peer_contains(&state->chunk_servers[j], removed_hash)) { - if (!hash_list_insert(&state->chunk_servers[j].rem_list, removed_hash)) + if (!hash_list_insert(&state->chunk_servers[j].ms_rem_list, removed_hash)) return -1; } } @@ -726,36 +732,6 @@ chunk_server_from_conn(MetadataServer *state, int conn_idx) return &state->chunk_servers[tag]; } -static int send_state_update(MetadataServer *state, int chunk_server_idx) -{ - ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; - - int conn_idx = tcp_index_from_tag(&state->tcp, chunk_server_idx); - assert(conn_idx > -1); - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE); - - uint32_t add_count = chunk_server->add_list.count; - uint32_t rem_count = chunk_server->rem_list.count; - message_write(&writer, &add_count, sizeof(add_count)); - message_write(&writer, &rem_count, sizeof(rem_count)); - - for (uint32_t i = 0; i < add_count; i++) - message_write(&writer, &chunk_server->add_list.items[i], sizeof(SHA256)); - - for (uint32_t i = 0; i < rem_count; i++) - message_write(&writer, &chunk_server->rem_list.items[i], sizeof(SHA256)); - - if (!message_writer_free(&writer)) - return -1; - - return 0; -} - static int process_chunk_server_auth(MetadataServer *state, int conn_idx, ByteView msg) { @@ -838,172 +814,130 @@ static int process_chunk_server_auth(MetadataServer *state, return 0; } -static int process_chunk_server_chunk_list(MetadataServer *state, +static int process_chunk_server_sync(MetadataServer *state, int conn_idx, ByteView msg) { int chunk_server_idx = tcp_get_tag(&state->tcp, conn_idx); assert(chunk_server_idx > -1); - assert(chunk_server_idx < MAX_CHUNK_SERVERS); - assert(state->chunk_servers[chunk_server_idx].used); - ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + assert(chunk_server_idx <= MAX_CHUNK_SERVERS); - // Iterate over each chunk held by this chunk - // server and determine whether it's useless, - // under-replicated, over-replicated, or - // properly replicated. - // Chunks that are useless or over-replicated - // are added to the remove list of this chunk - // server (note that any given chunk can only - // be over-replicated by 1, so removing it from - // this chunk server will fix the issue). If - // the chunk is under-replicated or properly - // replicated, do nothing. Note that it's only - // possible for a chunk to be under-replicated - // when the number of chunk servers is lower - // than the replication factor. + ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + assert(chunk_server->used); BinaryReader reader = { msg.ptr, msg.len, 0 }; - // version - if (!binary_read(&reader, NULL, sizeof(uint16_t))) + if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - uint16_t type; - if (!binary_read(&reader, &type, sizeof(type))) + uint32_t count; + if (!binary_read(&reader, &count, sizeof(count))) return -1; - if (type != MESSAGE_TYPE_CHUNK_LIST) - return -1; - - // length - if (!binary_read(&reader, NULL, sizeof(uint32_t))) - return -1; - - uint32_t num_hashes; - if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) - return -1; - - for (uint32_t i = 0; i < num_hashes; i++) { + for (uint32_t i = 0; i < count; i++) { SHA256 hash; if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - bool drop = false; + // If the chunk is not referenced by the file tree, do + // nothing. + if (!file_tree_uses_hash(&state->file_tree, hash)) - drop = true; - else { + continue; - int holders[MAX_CHUNK_SERVERS]; - int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); - assert(num_holders > -1); - assert(num_holders <= MAX_CHUNK_SERVERS); + // If the chunk is properly replicated or under-replicated, + // add it to the ms_add_list. - assert(num_holders <= state->replication_factor); + int holders[MAX_CHUNK_SERVERS]; + int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); + assert(num_holders > -1); + assert(num_holders <= MAX_CHUNK_SERVERS); - // Adding this chunk server as a holder will cause - // the chunk to be over-replicated - if (num_holders == state->replication_factor) - drop = true; + if (num_holders <= state->replication_factor) { + if (hash_list_insert(&chunk_server->ms_add_list, hash) < 0) { + assert(0); // TODO + } + continue; } - HashList *list; - if (drop) list = &chunk_server->rem_list; - else list = &chunk_server->add_list; + // If the chunk is over-replicated, either don't add + // it to the ms_add_list or add it to the ms_rem_list + // of some other holder. - if (hash_list_insert(list, hash) < 0) { + if (hash_list_insert(xxx, hash) < 0) { assert(0); // TODO } } - if (binary_read(&reader, NULL, 1)) + if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion return -1; - // No need to reply here - return 0; -} - -static int process_chunk_server_state_update_success(MetadataServer *state, - int conn_idx, ByteView msg) -{ - (void) msg; // Success message has no body - ChunkServerPeer *chunk_server = chunk_server_from_conn(state, conn_idx); - - // Merge add_list into old_list - for (int i = 0; i < chunk_server->add_list.count; i++) { - if (!hash_list_contains(&chunk_server->old_list, chunk_server->add_list.items[i])) - hash_list_insert(&chunk_server->old_list, chunk_server->add_list.items[i]); // TODO: what if this fails? - } - - // Clear add_list and rem_list - chunk_server->add_list.count = 0; - chunk_server->rem_list.count = 0; - - if (state->trace) { - int tag = tcp_get_tag(&state->tcp, conn_idx); - printf("Received STATE_UPDATE_SUCCESS from chunk server %d\n", tag); - } - - return 0; -} - -static int process_chunk_server_state_update_error(MetadataServer *state, - int conn_idx, ByteView msg) -{ - BinaryReader reader = { msg.ptr, msg.len, 0 }; - - // Read header - if (!binary_read(&reader, NULL, sizeof(MessageHeader))) - return -1; - - // Read error message - uint16_t error_len; - if (!binary_read(&reader, &error_len, sizeof(error_len))) - return -1; - - char error_msg[256]; - int read_len = error_len < sizeof(error_msg) - 1 ? error_len : sizeof(error_msg) - 1; - if (!binary_read(&reader, error_msg, read_len)) - return -1; - error_msg[read_len] = '\0'; - - // Skip remaining error message if it was too long - if (error_len > read_len) - binary_read(&reader, NULL, error_len - read_len); - - // Read missing chunks - uint32_t num_missing; - if (!binary_read(&reader, &num_missing, sizeof(num_missing))) - return -1; - - SHA256 *missing_chunks = sys_malloc(num_missing * sizeof(SHA256)); - if (missing_chunks == NULL) - return -1; - - for (uint32_t i = 0; i < num_missing; i++) { - if (!binary_read(&reader, &missing_chunks[i], sizeof(SHA256))) { - sys_free(missing_chunks); - return -1; - } - } - - if (state->trace) { - int tag = tcp_get_tag(&state->tcp, conn_idx); - printf("Received STATE_UPDATE_ERROR from chunk server %d: %s (missing %u chunks)\n", - tag, error_msg, num_missing); - } + // Respond with ms_add_list and ms_rem_list ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); assert(output); MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_LOCATIONS); + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_2); - message_write(&writer, &num_missing, sizeof(num_missing)); + uint32_t add_count = chunk_server->ms_add_list.count; // TODO: check implicit casts + message_write(&writer, &add_count, sizeof(add_count)); - for (uint32_t i = 0; i < num_missing; i++) { + for (uint32_t i = 0; i < add_count; i++) { + SHA256 hash = chunk_server->ms_add_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } - SHA256 hash = missing_chunks[i]; + uint32_t rem_count = chunk_server->ms_rem_list.count; // TODO: check implicit casts + message_write(&writer, &rem_count, sizeof(rem_count)); + + for (uint32_t i = 0; i < rem_count; i++) { + SHA256 hash = chunk_server->ms_rem_list.items[i]; + message_write(&writer, &hash, sizeof(hash)); + } + + if (!message_writer_free(&writer)) + return -1; + return 0; +} + +static int process_chunk_server_sync_3(MetadataServer *state, + int conn_idx, ByteView msg) +{ + int chunk_server_idx = tcp_get_tag(&state->tcp, conn_idx); + assert(chunk_server_idx > -1); + assert(chunk_server_idx <= MAX_CHUNK_SERVERS); + + ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; + assert(chunk_server->used); + + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + if (!binary_read(&reader, NULL, sizeof(MessageHeader))) + return -1; + + uint32_t count; + if (!binary_read(&reader, &count, sizeof(count))) + return -1; + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_4); + + HashList tmp_list; + hash_list_init(&tmp_list); + + for (uint32_t i = 0; i < count; i++) { + + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) + return -1; + + if (hash_list_insert(&tmp_list, hash) < 0) { + assert(0); // TODO + } int holders[MAX_CHUNK_SERVERS]; int num_holders = all_chunk_servers_holding_chunk(state, hash, holders, MAX_CHUNK_SERVERS); @@ -1013,17 +947,26 @@ static int process_chunk_server_state_update_error(MetadataServer *state, uint32_t tmp = num_holders; message_write(&writer, &tmp, sizeof(tmp)); - for (int j = 0; j < num_holders; j++) - message_write_server_addr(&writer, &state->chunk_servers[holders[j]]); - - message_write(&writer, &hash, sizeof(hash)); + for (int i = 0; i < num_holders; i++) + message_write_server_addr(&writer, xxx); } - sys_free(missing_chunks); + if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion + return -1; + + if (hash_list_merge(&state->ms_old_list, state->ms_add_list) < 0) { + assert(0); // TODO + } + + if (hash_list_remove_set(&state->ms_old_list, tmp_list) < 0) { + assert(0); // TODO + } + + hash_list_free(&state->ms_add_list); + state->ms_add_list = tmp_list; if (!message_writer_free(&writer)) return -1; - return 0; } @@ -1035,14 +978,11 @@ process_chunk_server_message(MetadataServer *state, case MESSAGE_TYPE_AUTH: return process_chunk_server_auth(state, conn_idx, msg); - case MESSAGE_TYPE_CHUNK_LIST: - return process_chunk_server_chunk_list(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC: + return process_chunk_server_sync(state, conn_idx, msg); - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: - return process_chunk_server_state_update_success(state, conn_idx, msg); - - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - return process_chunk_server_state_update_error(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_3: + return process_chunk_server_sync_3(state, conn_idx, msg); } return -1; } @@ -1051,8 +991,7 @@ static bool is_chunk_server_message_type(uint16_t type) { switch (type) { case MESSAGE_TYPE_AUTH: - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: + case MESSAGE_TYPE_SYNC: return true; default: @@ -1115,7 +1054,6 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * Event events[MAX_CONNS+1]; int num_events = tcp_translate_events(&state->tcp, events, contexts, polled, num_polled); - // Implement periodic health checks: send STATE_UPDATE to chunk servers Time current_time = get_current_time(); if (current_time == INVALID_TIME) { assert(0); // TODO @@ -1213,19 +1151,6 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * continue; } nearest_deadline(&next_wakeup, response_timeout); - - if (chunk_server->auth && chunk_server->last_sync_done) { - Time sync_timeout = chunk_server->last_sync_time + (Time) SYNC_INTERVAL * 1000000000; - if (current_time > sync_timeout) { - if (send_state_update(state, i) < 0) { - assert(0); // TODO - } - chunk_server->last_sync_time = current_time; - chunk_server->last_sync_done = false; - continue; - } - nearest_deadline(&next_wakeup, sync_timeout); - } } *timeout = deadline_to_timeout(next_wakeup, current_time); diff --git a/src/metadata_server.h b/src/metadata_server.h index 7edc499..88c5eb8 100644 --- a/src/metadata_server.h +++ b/src/metadata_server.h @@ -23,17 +23,14 @@ typedef struct { int num_addrs; Address addrs[MAX_SERVER_ADDRS]; - // Chunks held by the chunk server during - // the last update - HashList old_list; + // List of chunks that are known to be held by CS + HashList ms_old_list; - // Chunks added to the chunk server since - // the last update - HashList add_list; + // List of chunks that should be held by CS + HashList ms_add_list; - // Chunks removed from the chunk server - // since the last update - HashList rem_list; + // List of chunks that may be held by CS but should removed from it + HashList ms_rem_list; // Time when last STATE_UPDATE was sent Time last_sync_time; From 7b01ac63842d3150d84d078a08e19f9aa795cfc8 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 20:21:26 +0100 Subject: [PATCH 11/21] Fix compilation errors --- src/chunk_server.c | 124 ++++++++++------------------------- src/chunk_server.h | 17 +---- src/hash_set.c | 148 ++++++++++++++++++++++++++++++++++++++++++ src/hash_set.h | 38 +++++++++++ src/message.c | 101 ++-------------------------- src/message.h | 1 + src/metadata_server.c | 115 ++++++++------------------------ src/metadata_server.h | 13 ++-- 8 files changed, 262 insertions(+), 295 deletions(-) create mode 100644 src/hash_set.c create mode 100644 src/hash_set.h diff --git a/src/chunk_server.c b/src/chunk_server.c index b484766..24131cd 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -59,77 +59,6 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash) return 0; } -static void -removal_list_init(RemovalList *list) -{ - list->count = 0; - list->capacity = 0; - list->items = NULL; -} - -static void -removal_list_free(RemovalList *list) -{ - sys_free(list->items); -} - -static int -removal_list_find(RemovalList *list, SHA256 hash) -{ - for (int i = 0; i < list->count; i++) - if (!memcmp(&list->items[i].hash, &hash, sizeof(SHA256))) - return i; - return -1; -} - -static int -removal_list_add(RemovalList *list, SHA256 hash, Time marked_time) -{ - // Check if already in list - int idx = removal_list_find(list, hash); - if (idx >= 0) { - // Already marked, keep the original time - return 0; - } - - if (list->count == list->capacity) { - int new_capacity; - if (list->capacity == 0) - new_capacity = 8; - else - new_capacity = 2 * list->capacity; - - PendingRemoval *new_items = sys_malloc(new_capacity * sizeof(PendingRemoval)); - if (new_items == NULL) - return -1; - - if (list->capacity > 0) { - memcpy(new_items, list->items, list->count * sizeof(list->items[0])); - sys_free(list->items); - } - - list->items = new_items; - list->capacity = new_capacity; - } - - list->items[list->count++] = (PendingRemoval) { hash, marked_time }; - return 0; -} - -static void -removal_list_remove(RemovalList *list, SHA256 hash) -{ - int idx = removal_list_find(list, hash); - if (idx >= 0) { - // Remove by shifting remaining items - if (idx < list->count - 1) { - memmove(&list->items[idx], &list->items[idx + 1], - (list->count - idx - 1) * sizeof(list->items[0])); - } - list->count--; - } -} - static int chunk_store_init(ChunkStore *store, string path) { if (create_dir(path) && errno != EEXIST) @@ -333,8 +262,13 @@ static int process_metadata_server_sync_2(ChunkServer *state, if (!binary_read(&reader, &add_count, sizeof(add_count))) return -1; - HashList tmp_list; - hash_list_init(&tmp_list); + Time current_time = get_current_time(); + if (current_time == INVALID_TIME) { + assert(0); // TODO + } + + HashSet tmp_list; + hash_set_init(&tmp_list); for (uint32_t i = 0; i < add_count; i++) { @@ -345,22 +279,24 @@ static int process_metadata_server_sync_2(ChunkServer *state, // Elements in ms_add_list that are not held by the // chunk server are added to a temporary list tmp_list - if (chunk_store_exists(&state->store, hash) < 0) { // TODO: does it return a boolean? - if (hash_list_insert(&tmp_list, hash) < 0) { + if (!chunk_store_exists(&state->store, hash)) { + if (hash_set_insert(&tmp_list, hash) < 0) { assert(0); // TODO } continue; } - hash_list_remove(&state->cs_rem_list, hash); - hash_list_remove(&state->cs_add_list, hash); + timed_hash_set_remove(&state->cs_rem_list, hash); + hash_set_remove(&state->cs_add_list, hash); } - if (hash_list_merge(&state->cs_rem_list, state->cs_add_list) < 0) { - assert(0); // TODO + for (int i = 0; i < state->cs_add_list.count; i++) { + if (timed_hash_set_insert(&state->cs_rem_list, state->cs_add_list.items[i], current_time) < 0) { + assert(0); // TODO + } } - hash_list_clear(&state->cs_add_list); + hash_set_clear(&state->cs_add_list); uint32_t rem_count; if (!binary_read(&reader, &rem_count, sizeof(rem_count))) @@ -372,7 +308,7 @@ static int process_metadata_server_sync_2(ChunkServer *state, if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - if (timed_hash_list_insert(&state->cs_rem_list, hash, current_time) < 0) { + if (timed_hash_set_insert(&state->cs_rem_list, hash, current_time) < 0) { assert(0); // TODO } } @@ -389,12 +325,12 @@ static int process_metadata_server_sync_2(ChunkServer *state, uint32_t count = tmp_list.count + state->cs_lst_list.count; // TODO: overflow message_write(&writer, &count, sizeof(count)); - for (uint32_t i = 0; i < tmp_list.count; i++) { + for (int i = 0; i < tmp_list.count; i++) { SHA256 hash = tmp_list.items[i]; message_write(&writer, &hash, sizeof(hash)); } - for (uint32_t i = 0; i < state->cs_lst_list.count; i++) { + for (int i = 0; i < state->cs_lst_list.count; i++) { SHA256 hash = state->cs_lst_list.items[i]; message_write(&writer, &hash, sizeof(hash)); } @@ -839,6 +775,12 @@ static int send_sync_message(ChunkServer *state) { assert(state->disconnect_time == INVALID_TIME); + int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); + assert(conn_idx > -1); + + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_SYNC); @@ -888,7 +830,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts state->downloading = false; pending_download_list_init(&state->pending_download_list); - removal_list_init(&state->removal_list); + hash_set_init(&state->cs_add_list); + hash_set_init(&state->cs_lst_list); + timed_hash_set_init(&state->cs_rem_list); char tmp[1<<10]; if (addr.len >= (int) sizeof(tmp)) { @@ -942,7 +886,9 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts int chunk_server_free(ChunkServer *state) { pending_download_list_free(&state->pending_download_list); - removal_list_free(&state->removal_list); + timed_hash_set_free(&state->cs_rem_list); + hash_set_free(&state->cs_lst_list); + hash_set_free(&state->cs_add_list); chunk_store_free(&state->store); tcp_context_free(&state->tcp); return 0; @@ -1031,12 +977,12 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled Time deadline = INVALID_TIME; // Remove items from the remove list that got too old - for (int i = 0; i < state->removal_list.count; i++) { - PendingRemoval *removal = &state->removal_list.items[i]; - Time removal_time = removal->marked_time + (Time) DELETION_TIMEOUT * 1000000000; + for (int i = 0; i < state->cs_rem_list.count; i++) { + TimedHash *removal = &state->cs_rem_list.items[i]; + Time removal_time = removal->time + (Time) DELETION_TIMEOUT * 1000000000; if (removal_time < current_time) { if (chunk_store_remove(&state->store, removal->hash) == 0) - *removal = state->removal_list.items[--state->removal_list.count]; + *removal = state->cs_rem_list.items[--state->cs_rem_list.count]; } else { nearest_deadline(&deadline, removal_time); } diff --git a/src/chunk_server.h b/src/chunk_server.h index b80d495..e3c7854 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -26,17 +26,6 @@ typedef struct { PendingDownload *items; } PendingDownloadList; -typedef struct { - SHA256 hash; - Time time; -} TimedHash; - -typedef struct { - int count; - int capacity; - TimedHash *items; -} TimedHashList; - typedef struct { bool trace; @@ -56,13 +45,13 @@ typedef struct { PendingDownloadList pending_download_list; // List of chunks added since the last update - HashList cs_add_list; + HashSet cs_add_list; // List of chunks marked for removal after a timeout - TimedHashList cs_rem_list; + TimedHashSet cs_rem_list; // List of chunks that were lost due to errors or forceful removals of chunk files - HashList cs_lst_list; + HashSet cs_lst_list; } ChunkServer; diff --git a/src/hash_set.c b/src/hash_set.c new file mode 100644 index 0000000..d4075d8 --- /dev/null +++ b/src/hash_set.c @@ -0,0 +1,148 @@ +#include +#include + +#include "system.h" +#include "hash_set.h" + +void hash_set_init(HashSet *set) +{ + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +void hash_set_free(HashSet *set) +{ + sys_free(set->items); + set->items = NULL; +} + +void hash_set_clear(HashSet *set) +{ + free(set->items); + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +int hash_set_insert(HashSet *set, SHA256 hash) +{ + // Avoid duplicates + for (int i = 0; i < set->count; i++) + if (!memcmp(&set->items[i], &hash, sizeof(SHA256))) + return 0; // Already present + + if (set->count == set->capacity) { + + int new_capacity; + if (set->items == NULL) + new_capacity = 16; + else + new_capacity = 2 * set->capacity; + + SHA256 *new_items = sys_realloc(set->items, new_capacity * sizeof(SHA256)); + if (new_items == NULL) + return -1; + + set->items = new_items; + set->capacity = new_capacity; + } + + set->items[set->count++] = hash; + return 0; +} + +bool hash_set_remove(HashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&hash, &set->items[i], sizeof(SHA256))) { + set->items[i] = set->items[--set->count]; + return true; + } + return false; +} + +bool hash_set_contains(HashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&hash, &set->items[i], sizeof(SHA256))) + return true; + return false; +} + +int hash_set_merge(HashSet *dst, HashSet src) +{ + assert(0); // TODO +} + +void hash_set_remove_set(HashSet *dst, HashSet src) +{ + assert(0); // TODO +} + +void timed_hash_set_init(TimedHashSet *set) +{ + set->items = NULL; + set->count = 0; + set->capacity = 0; +} + +void timed_hash_set_free(TimedHashSet *set) +{ + sys_free(set->items); + set->items = NULL; +} + +int timed_hash_set_find(TimedHashSet *set, SHA256 hash) +{ + for (int i = 0; i < set->count; i++) + if (!memcmp(&set->items[i].hash, &hash, sizeof(SHA256))) + return i; + return -1; +} + +int timed_hash_set_insert(TimedHashSet *set, SHA256 hash, Time time) +{ + // Check if already in set + int idx = timed_hash_set_find(set, hash); + if (idx >= 0) { + // Already marked, keep the original time + return 0; + } + + if (set->count == set->capacity) { + int new_capacity; + if (set->capacity == 0) + new_capacity = 8; + else + new_capacity = 2 * set->capacity; + + TimedHash *new_items = sys_malloc(new_capacity * sizeof(TimedHash)); + if (new_items == NULL) + return -1; + + if (set->capacity > 0) { + memcpy(new_items, set->items, set->count * sizeof(set->items[0])); + sys_free(set->items); + } + + set->items = new_items; + set->capacity = new_capacity; + } + + set->items[set->count++] = (TimedHash) { hash, time }; + return 0; +} + +void timed_hash_set_remove(TimedHashSet *set, SHA256 hash) +{ + int idx = timed_hash_set_find(set, hash); + if (idx >= 0) { + // Remove by shifting remaining items + if (idx < set->count - 1) { + memmove(&set->items[idx], &set->items[idx + 1], + (set->count - idx - 1) * sizeof(set->items[0])); + } + set->count--; + } +} diff --git a/src/hash_set.h b/src/hash_set.h new file mode 100644 index 0000000..31834d5 --- /dev/null +++ b/src/hash_set.h @@ -0,0 +1,38 @@ +#ifndef HASH_SET_INCLUDED +#define HASH_SET_INCLUDED + +#include "basic.h" + +typedef struct { + SHA256 *items; + int count; + int capacity; +} HashSet; + +typedef struct { + SHA256 hash; + Time time; +} TimedHash; + +typedef struct { + TimedHash *items; + int count; + int capacity; +} TimedHashSet; + +void hash_set_init (HashSet *set); +void hash_set_free (HashSet *set); +void hash_set_clear (HashSet *set); +int hash_set_insert (HashSet *set, SHA256 hash); +bool hash_set_remove (HashSet *set, SHA256 hash); +int hash_set_merge (HashSet *dst, HashSet src); +void hash_set_remove_set(HashSet *dst, HashSet src); +bool hash_set_contains (HashSet *set, SHA256 hash); + +void timed_hash_set_init (TimedHashSet *set); +void timed_hash_set_free (TimedHashSet *set); +int timed_hash_set_find (TimedHashSet *set, SHA256 hash); +int timed_hash_set_insert (TimedHashSet *set, SHA256 hash, Time time); +void timed_hash_set_remove (TimedHashSet *set, SHA256 hash); + +#endif // HASH_SET_INCLUDED diff --git a/src/message.c b/src/message.c index bbe6738..cfdcb49 100644 --- a/src/message.c +++ b/src/message.c @@ -94,13 +94,14 @@ static char *message_type_to_str(uint16_t type) case MESSAGE_TYPE_WRITE_SUCCESS: return "WRITE_SUCCESS"; // Metadata server -> Chunk server - case MESSAGE_TYPE_STATE_UPDATE: return "STATE_UPDATE"; + case MESSAGE_TYPE_SYNC_2: return "SYNC_2"; + case MESSAGE_TYPE_SYNC_4: return "SYNC_4"; case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: return "DOWNLOAD_LOCATIONS"; // Chunk server -> Metadata server case MESSAGE_TYPE_AUTH: return "AUTH"; - case MESSAGE_TYPE_STATE_UPDATE_ERROR: return "UPDATE_ERROR"; - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: return "UPDATE_SUCCESS"; + case MESSAGE_TYPE_SYNC: return "SYNC"; + case MESSAGE_TYPE_SYNC_3: return "SYNC_3"; // Chunk server -> Client case MESSAGE_TYPE_CREATE_CHUNK_ERROR: return "CREATE_CHUNK_ERROR"; @@ -418,99 +419,7 @@ void message_dump(FILE *stream, ByteView msg) } break; - case MESSAGE_TYPE_DOWNLOAD_CHUNK: - printf(" (TODO)\n"); - break; - - // Metadata server -> Client - - case MESSAGE_TYPE_CREATE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_CREATE_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DELETE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DELETE_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_LIST_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_LIST_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_READ_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_READ_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_WRITE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_WRITE_SUCCESS: - printf(" (TODO)\n"); - break; - - // Metadata server -> Chunk server - - case MESSAGE_TYPE_STATE_UPDATE: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: - printf(" (TODO)\n"); - break; - - // Chunk server -> Metadata server - - case MESSAGE_TYPE_AUTH: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_STATE_UPDATE_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: - printf(" (TODO)\n"); - break; - - // Chunk server -> Client - - case MESSAGE_TYPE_CREATE_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_CREATE_CHUNK_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_UPLOAD_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR: - printf(" (TODO)\n"); - break; - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS: + default: printf(" (TODO)\n"); break; } diff --git a/src/message.h b/src/message.h index 9d8ba83..641babb 100644 --- a/src/message.h +++ b/src/message.h @@ -40,6 +40,7 @@ enum { // Chunk server -> Metadata server MESSAGE_TYPE_AUTH, MESSAGE_TYPE_SYNC, + MESSAGE_TYPE_SYNC_3, // Chunk server -> Client MESSAGE_TYPE_CREATE_CHUNK_ERROR, diff --git a/src/metadata_server.c b/src/metadata_server.c index c1f49e6..5533cbe 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -1,8 +1,3 @@ -#include "basic.h" -#include "byte_queue.h" -#include "config.h" -#include "file_tree.h" -#include "tcp.h" #define _GNU_SOURCE #include @@ -12,82 +7,36 @@ #include "message.h" #include "metadata_server.h" -static void hash_list_init(HashList *hash_list) -{ - hash_list->count = 0; - hash_list->capacity = 0; - hash_list->items = NULL; -} - -static void hash_list_free(HashList *hash_list) -{ - sys_free(hash_list->items); -} - -static int hash_list_insert(HashList *hash_list, SHA256 hash) -{ - // Avoid duplicates - for (int i = 0; i < hash_list->count; i++) - if (!memcmp(&hash_list->items[i], &hash, sizeof(SHA256))) - return 0; // Already present - - if (hash_list->count == hash_list->capacity) { - - int new_capacity; - if (hash_list->items == NULL) - new_capacity = 16; - else - new_capacity = 2 * hash_list->capacity; - - SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256)); - if (new_items == NULL) - return -1; - - hash_list->items = new_items; - hash_list->capacity = new_capacity; - } - - hash_list->items[hash_list->count++] = hash; - return 0; -} - -static bool hash_list_contains(HashList *hash_list, SHA256 hash) -{ - for (int j = 0; j < hash_list->count; j++) - if (!memcmp(&hash, &hash_list->items[j], sizeof(SHA256))) - return true; - return false; -} - static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time) { chunk_server->used = true; chunk_server->auth = false; chunk_server->num_addrs = 0; - hash_list_init(&chunk_server->ms_old_list); - hash_list_init(&chunk_server->ms_add_list); - hash_list_init(&chunk_server->ms_rem_list); + hash_set_init(&chunk_server->ms_old_list); + hash_set_init(&chunk_server->ms_add_list); + hash_set_init(&chunk_server->ms_rem_list); chunk_server->last_sync_time = current_time; chunk_server->last_response_time = current_time; } static void chunk_server_peer_free(ChunkServerPeer *chunk_server) { - hash_list_free(&chunk_server->ms_rem_list); - hash_list_free(&chunk_server->ms_add_list); - hash_list_free(&chunk_server->ms_old_list); + hash_set_free(&chunk_server->ms_rem_list); + hash_set_free(&chunk_server->ms_add_list); + hash_set_free(&chunk_server->ms_old_list); chunk_server->used = false; } static bool chunk_server_peer_contains(ChunkServerPeer *chunk_server, SHA256 hash) { - return hash_list_contains(&chunk_server->old_list, hash) - || hash_list_contains(&chunk_server->add_list, hash); + return hash_set_contains(&chunk_server->ms_old_list, hash) + || hash_set_contains(&chunk_server->ms_add_list, hash); } static bool chunk_server_peer_load(ChunkServerPeer *chunk_server) { - return chunk_server->old_list.count + chunk_server->add_list.count; + return chunk_server->ms_old_list.count + + chunk_server->ms_add_list.count; } // Returns all chunk servers holding the given chunk @@ -677,7 +626,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) int k = find_chunk_server_by_addr(state, results[i].addrs[j]); if (k == -1) return -1; - if (hash_list_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0) + if (hash_set_insert(&state->chunk_servers[k].ms_add_list, new_hashes[i]) < 0) return -1; } } @@ -690,7 +639,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) // Add to rem_list for all chunk servers that have this chunk for (int j = 0; j < state->num_chunk_servers; j++) { if (chunk_server_peer_contains(&state->chunk_servers[j], removed_hash)) { - if (!hash_list_insert(&state->chunk_servers[j].ms_rem_list, removed_hash)) + if (!hash_set_insert(&state->chunk_servers[j].ms_rem_list, removed_hash)) return -1; } } @@ -803,14 +752,7 @@ static int process_chunk_server_auth(MetadataServer *state, // we accept all connections that provide valid address information. chunk_server->auth = true; - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - MessageWriter writer; - message_writer_init(&writer, output, MESSAGE_TYPE_CHUNK_LIST_REQUEST); - if (!message_writer_free(&writer)) - return -1; - + // No need to respond return 0; } @@ -854,7 +796,7 @@ static int process_chunk_server_sync(MetadataServer *state, assert(num_holders <= MAX_CHUNK_SERVERS); if (num_holders <= state->replication_factor) { - if (hash_list_insert(&chunk_server->ms_add_list, hash) < 0) { + if (hash_set_insert(&chunk_server->ms_add_list, hash) < 0) { assert(0); // TODO } continue; @@ -863,10 +805,9 @@ static int process_chunk_server_sync(MetadataServer *state, // If the chunk is over-replicated, either don't add // it to the ms_add_list or add it to the ms_rem_list // of some other holder. - - if (hash_list_insert(xxx, hash) < 0) { - assert(0); // TODO - } + // + // TODO: For now we don't add it to the ms_add_list, + // but there may be a better solution. } if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion @@ -926,8 +867,8 @@ static int process_chunk_server_sync_3(MetadataServer *state, MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_SYNC_4); - HashList tmp_list; - hash_list_init(&tmp_list); + HashSet tmp_list; + hash_set_init(&tmp_list); for (uint32_t i = 0; i < count; i++) { @@ -935,7 +876,7 @@ static int process_chunk_server_sync_3(MetadataServer *state, if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - if (hash_list_insert(&tmp_list, hash) < 0) { + if (hash_set_insert(&tmp_list, hash) < 0) { assert(0); // TODO } @@ -947,23 +888,23 @@ static int process_chunk_server_sync_3(MetadataServer *state, uint32_t tmp = num_holders; message_write(&writer, &tmp, sizeof(tmp)); - for (int i = 0; i < num_holders; i++) - message_write_server_addr(&writer, xxx); + for (int j = 0; j < num_holders; j++) { + int k = holders[j]; + message_write_server_addr(&writer, &state->chunk_servers[k]); + } } if (binary_read(&reader, NULL, 1)) // TODO: this should probably be an assertion return -1; - if (hash_list_merge(&state->ms_old_list, state->ms_add_list) < 0) { + if (hash_set_merge(&chunk_server->ms_old_list, chunk_server->ms_add_list) < 0) { assert(0); // TODO } - if (hash_list_remove_set(&state->ms_old_list, tmp_list) < 0) { - assert(0); // TODO - } + hash_set_remove_set(&chunk_server->ms_old_list, tmp_list); - hash_list_free(&state->ms_add_list); - state->ms_add_list = tmp_list; + hash_set_free(&chunk_server->ms_add_list); + chunk_server->ms_add_list = tmp_list; if (!message_writer_free(&writer)) return -1; diff --git a/src/metadata_server.h b/src/metadata_server.h index 88c5eb8..d8b7ed8 100644 --- a/src/metadata_server.h +++ b/src/metadata_server.h @@ -5,16 +5,11 @@ #include "file_tree.h" #include "config.h" #include "basic.h" +#include "hash_set.h" #define CONNECTION_TAG_CLIENT -2 #define CONNECTION_TAG_UNKNOWN -3 -typedef struct { - int count; - int capacity; - SHA256 *items; -} HashList; - typedef struct { bool used; @@ -24,13 +19,13 @@ typedef struct { Address addrs[MAX_SERVER_ADDRS]; // List of chunks that are known to be held by CS - HashList ms_old_list; + HashSet ms_old_list; // TODO: rename all *_list symbols to *_set // List of chunks that should be held by CS - HashList ms_add_list; + HashSet ms_add_list; // List of chunks that may be held by CS but should removed from it - HashList ms_rem_list; + HashSet ms_rem_list; // Time when last STATE_UPDATE was sent Time last_sync_time; From 4b97b176bd61235fd5afd89a9b17aee0b2f431fb Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 19:35:06 +0000 Subject: [PATCH 12/21] Add branch coverage measurement for random simulation This commit adds tooling to measure how many branches the random simulation reaches during execution, which helps understand code coverage and identify untested code paths. Changes: - Add coverage build target to Makefile with --coverage flags - Create measure_coverage.sh script to build, run, and report branch coverage - Add signal handlers (SIGINT/SIGTERM) to main_test.c for clean shutdown - Update clean target to remove coverage files (*.gcda, *.gcno) The script runs the simulation for a specified duration (default 5 seconds), then uses gcov to analyze which branches were executed. In a 3-second run, the simulation reaches approximately 32% of all branches (781/2431). Usage: ./measure_coverage.sh [duration_in_seconds] Example output shows per-file and total branch coverage statistics. --- Makefile | 33 +++++++++----- measure_coverage.sh | 106 ++++++++++++++++++++++++++++++++++++++++++++ src/main_test.c | 10 ++++- 3 files changed, 138 insertions(+), 11 deletions(-) create mode 100755 measure_coverage.sh diff --git a/Makefile b/Makefile index d8c62fc..74fdbff 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,7 @@ CFLAGS = -Wall -Wextra -ggdb +COVERAGE_CFLAGS = $(CFLAGS) --coverage +COVERAGE_LFLAGS = --coverage ifeq ($(OS),Windows_NT) LFLAGS = -lws2_32 @@ -13,16 +15,21 @@ CFILES = $(shell find src -name '*.c') HFILES = $(shell find src -name '*.h') OFILES = $(CFILES:.c=.o) -.PHONY: all clean +.PHONY: all clean coverage all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a +coverage: mousefs_random_test_coverage$(EXT) + mousefs$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER mousefs_random_test$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_TEST +mousefs_random_test_coverage$(EXT): $(CFILES) $(HFILES) + gcc -o $@ $(CFILES) $(COVERAGE_CFLAGS) $(LFLAGS) $(COVERAGE_LFLAGS) -Iinc -DBUILD_TEST + example_client$(EXT): libmousefs.a gcc -o $@ examples/main.c $(CFLAGS) -lmousefs $(LFLAGS) -Iinc -L. @@ -33,12 +40,18 @@ libmousefs.a: $(OFILES) ar rcs $@ $^ clean: - rm -f \ - mousefs.exe \ - mousefs.out \ - mousefs_random_test.exe \ - mousefs_random_test.out \ - example_client.exe \ - example_client.out \ - libmousefs.a \ - src/*.o + rm -f \ + mousefs.exe \ + mousefs.out \ + mousefs_random_test.exe \ + mousefs_random_test.out \ + mousefs_random_test_coverage.exe \ + mousefs_random_test_coverage.out \ + example_client.exe \ + example_client.out \ + libmousefs.a \ + src/*.o \ + src/*.gcda \ + src/*.gcno \ + *.gcda \ + *.gcno diff --git a/measure_coverage.sh b/measure_coverage.sh new file mode 100755 index 0000000..d513ec8 --- /dev/null +++ b/measure_coverage.sh @@ -0,0 +1,106 @@ +#!/bin/bash +# Script to measure branch coverage of the random simulation + +set -e + +# Colors for output +GREEN='\033[0;32m' +BLUE='\033[0;34m' +YELLOW='\033[1;33m' +NC='\033[0m' # No Color + +echo -e "${BLUE}========================================${NC}" +echo -e "${BLUE}Branch Coverage Measurement Tool${NC}" +echo -e "${BLUE}========================================${NC}" +echo + +# Clean previous coverage data +echo -e "${YELLOW}Cleaning previous coverage data...${NC}" +rm -f src/*.gcda src/*.gcno *.gcda *.gcno +make clean > /dev/null 2>&1 + +# Build with coverage +echo -e "${YELLOW}Building with coverage instrumentation...${NC}" +make coverage + +# Run the simulation for a limited time +echo -e "${YELLOW}Running simulation...${NC}" +SIMULATION_TIME=${1:-5} # Default to 5 seconds if not specified +echo "Running for ${SIMULATION_TIME} seconds..." + +# Run simulation in background and kill it after specified time +timeout ${SIMULATION_TIME}s ./mousefs_random_test_coverage.out || true + +# Generate coverage reports +echo +echo -e "${YELLOW}Generating coverage reports...${NC}" + +# Find all .gcda files and generate coverage reports +GCDA_FILES=$(find . -name "*.gcda") + +TOTAL_BRANCHES=0 +TAKEN_BRANCHES=0 + +# Generate gcov reports from the coverage data files +for gcda_file in $GCDA_FILES; do + # Extract the source file name from the gcda filename + # Files are named like: mousefs_random_test_coverage.out-basic.gcda + basename=$(basename "$gcda_file" .gcda) + source_name=${basename#*-} # Remove prefix up to and including '-' + + # Run gcov to generate the .gcov file + gcov -b "$gcda_file" > /dev/null 2>&1 || true +done + +# Parse gcov output to count branches +echo +echo -e "${BLUE}========================================${NC}" +echo -e "${BLUE}Branch Coverage Summary${NC}" +echo -e "${BLUE}========================================${NC}" +echo + +# Process each .gcov file +for gcov_file in *.c.gcov; do + if [ -f "$gcov_file" ]; then + # Extract branch statistics from the gcov file + branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null) + if [ -z "$branches" ]; then branches=0; fi + + if [ "$branches" -gt 0 ]; then + taken_count=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") + # Clean up the value - remove any whitespace/newlines + taken_count=$(echo "$taken_count" | tr -d '[:space:]') + if [ -z "$taken_count" ] || [ "$taken_count" = "" ]; then taken_count=0; fi + + TOTAL_BRANCHES=$((TOTAL_BRANCHES + branches)) + TAKEN_BRANCHES=$((TAKEN_BRANCHES + taken_count)) + percentage=$((taken_count * 100 / branches)) + filename=$(echo "$gcov_file" | sed 's/.gcov$//') + printf "%-40s %5d / %5d branches (%3d%%)\n" "$filename" "$taken_count" "$branches" "$percentage" + fi + fi +done + +echo +echo -e "${BLUE}========================================${NC}" +if [ "$TOTAL_BRANCHES" -gt 0 ]; then + COVERAGE_PERCENT=$((TAKEN_BRANCHES * 100 / TOTAL_BRANCHES)) + echo -e "${GREEN}Total: $TAKEN_BRANCHES / $TOTAL_BRANCHES branches reached (${COVERAGE_PERCENT}%)${NC}" +else + echo -e "${YELLOW}No branch coverage data found${NC}" +fi +echo -e "${BLUE}========================================${NC}" +echo + +# Optionally show detailed coverage for specific files +if [ "$2" == "--detailed" ]; then + echo -e "${YELLOW}Detailed coverage reports available in *.gcov files${NC}" + echo "Use 'less .gcov' to view detailed line and branch coverage" +fi + +# Clean up gcov files +echo -e "${YELLOW}Cleaning up coverage files...${NC}" +rm -f *.gcov + +echo +echo -e "${GREEN}Coverage measurement complete!${NC}" diff --git a/src/main_test.c b/src/main_test.c index 294a0e7..17d0cfb 100644 --- a/src/main_test.c +++ b/src/main_test.c @@ -8,12 +8,20 @@ static sig_atomic_t simulation_should_stop = false; +static void signal_handler(int signum) +{ + (void)signum; + simulation_should_stop = true; +} + int main(int argc, char **argv) { (void)argc; (void)argv; - // TODO: set simulation_should_stop=true on ctrl+C + // Set up signal handlers for clean shutdown + signal(SIGINT, signal_handler); + signal(SIGTERM, signal_handler); startup_simulation(2); From b465d5282e5b5010fdb2089198e435e65fe2c0c5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 19:55:08 +0000 Subject: [PATCH 13/21] Add HTML coverage reports with branch-level detail Enhance the branch coverage measurement tool with HTML report generation that shows exactly which branches were taken during simulation execution. Changes: - Add generate_coverage_html.sh script to create interactive HTML reports - Update measure_coverage.sh with --html flag to generate reports - Add *.gcov, *.gcda, *.gcno to .gitignore - HTML reports show: - Overall coverage summary with visual progress bars - Per-file coverage breakdown with clickable links - Source code view with color-coded branch coverage - Green highlight: branches taken - Red highlight: branches not taken - Branch execution counts and percentages Usage: ./measure_coverage.sh [duration] --html The HTML report is generated in coverage_report/index.html and can be viewed in any web browser. Each source file links to a detailed view showing which specific branches were executed. --- .gitignore | 5 + generate_coverage_html.sh | 353 ++++++++++++++++++++++++++++++++++++++ measure_coverage.sh | 19 +- 3 files changed, 370 insertions(+), 7 deletions(-) create mode 100755 generate_coverage_html.sh diff --git a/.gitignore b/.gitignore index 2b915ce..7eb8ffc 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,8 @@ *.o *.a chunk_server_data_* +# Coverage reports +coverage_report/ +*.gcov +*.gcda +*.gcno diff --git a/generate_coverage_html.sh b/generate_coverage_html.sh new file mode 100755 index 0000000..1eec3ae --- /dev/null +++ b/generate_coverage_html.sh @@ -0,0 +1,353 @@ +#!/bin/bash +# Generate HTML coverage report from gcov files + +set -e + +OUTPUT_DIR="coverage_report" +mkdir -p "$OUTPUT_DIR" + +# Create main index.html +cat > "$OUTPUT_DIR/index.html" <<'HTMLHEADER' + + + + + Branch Coverage Report + + + +
+

🎯 Branch Coverage Report

+

Generated from random simulation coverage analysis

+ +
+

Overall Coverage Summary

+HTMLHEADER + +# Calculate totals +TOTAL_BRANCHES=0 +TAKEN_BRANCHES=0 + +for gcov_file in *.c.gcov; do + if [ -f "$gcov_file" ]; then + branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null || echo "0") + if [ "$branches" -gt 0 ]; then + taken=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") + taken=$(echo "$taken" | tr -d '[:space:]') + if [ -z "$taken" ]; then taken=0; fi + + TOTAL_BRANCHES=$((TOTAL_BRANCHES + branches)) + TAKEN_BRANCHES=$((TAKEN_BRANCHES + taken)) + fi + fi +done + +if [ "$TOTAL_BRANCHES" -gt 0 ]; then + COVERAGE_PERCENT=$((TAKEN_BRANCHES * 100 / TOTAL_BRANCHES)) + + # Determine coverage class + if [ "$COVERAGE_PERCENT" -lt 30 ]; then + COVERAGE_CLASS="coverage-low" + elif [ "$COVERAGE_PERCENT" -lt 70 ]; then + COVERAGE_CLASS="coverage-medium" + else + COVERAGE_CLASS="coverage-high" + fi + + cat >> "$OUTPUT_DIR/index.html" < + Total Branches: $TOTAL_BRANCHES
+ Branches Taken: $TAKEN_BRANCHES
+ Coverage: ${COVERAGE_PERCENT}% +
+
+
+
${COVERAGE_PERCENT}%
+
+
+ +

Coverage by File

+ + + + + + + + +HTMLSUMMARY + + # Generate table rows for each file + for gcov_file in *.c.gcov; do + if [ -f "$gcov_file" ]; then + filename=$(echo "$gcov_file" | sed 's/.gcov$//') + branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null || echo "0") + + if [ "$branches" -gt 0 ]; then + taken=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") + taken=$(echo "$taken" | tr -d '[:space:]') + if [ -z "$taken" ]; then taken=0; fi + + percentage=$((taken * 100 / branches)) + + if [ "$percentage" -lt 30 ]; then + bar_class="coverage-low" + elif [ "$percentage" -lt 70 ]; then + bar_class="coverage-medium" + else + bar_class="coverage-high" + fi + + # Generate individual file report + file_html="${filename}.html" + echo "Generating report for $filename..." + + cat > "$OUTPUT_DIR/$file_html" < + + + + Coverage: $filename + + + +
+ ← Back to Summary +

Coverage Report: $filename

+
+ Branches Taken: $taken / $branches ($percentage%)
+
+
+FILEHEADER
+
+                # Process the gcov file and add source code with branch info
+                awk '
+                    /^branch/ {
+                        branch_info = branch_info " " $0
+                        next
+                    }
+                    /^        / {
+                        # Line with execution count
+                        line_num = $2
+                        gsub(/:/, "", line_num)
+                        exec_count = $1
+                        gsub(/^[ \t]+/, "", exec_count)
+
+                        # Get the actual source line
+                        line_content = substr($0, index($0, $3))
+
+                        # Determine background color based on branch coverage
+                        bg_class = ""
+                        if (branch_info != "") {
+                            if (branch_info ~ /taken 0/) {
+                                bg_class = "branch-not-taken"
+                            } else if (branch_info ~ /never executed/) {
+                                bg_class = "branch-not-taken"
+                            } else {
+                                bg_class = "branch-taken"
+                            }
+                        }
+
+                        # Format execution count
+                        if (exec_count == "-") {
+                            exec_str = "-"
+                        } else if (exec_count == "#####") {
+                            exec_str = "0"
+                            if (bg_class == "") bg_class = "branch-not-taken"
+                        } else {
+                            exec_str = exec_count
+                        }
+
+                        printf "%s%s%s", bg_class, line_num, exec_str, line_content
+
+                        if (branch_info != "") {
+                            printf "%s", branch_info
+                        }
+                        printf "\n"
+
+                        branch_info = ""
+                    }
+                ' "$gcov_file" >> "$OUTPUT_DIR/$file_html"
+
+                cat >> "$OUTPUT_DIR/$file_html" <<'FILEFOOTER'
+
+
+ + +FILEFOOTER + + # Add row to main index + cat >> "$OUTPUT_DIR/index.html" < +
+ + + + + +TABLEROW + fi + fi + done + + # Close HTML + cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' +
FileBranches TakenTotal BranchesCoverageVisual
$filename$taken$branches${percentage}% +
+
+
${percentage}%
+
+
+ + + +HTMLFOOTER + +else + cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' +

No branch coverage data found.

+ + + + +HTMLFOOTER +fi + +echo "HTML report generated in $OUTPUT_DIR/" diff --git a/measure_coverage.sh b/measure_coverage.sh index d513ec8..a10e1de 100755 --- a/measure_coverage.sh +++ b/measure_coverage.sh @@ -92,15 +92,20 @@ fi echo -e "${BLUE}========================================${NC}" echo -# Optionally show detailed coverage for specific files -if [ "$2" == "--detailed" ]; then - echo -e "${YELLOW}Detailed coverage reports available in *.gcov files${NC}" - echo "Use 'less .gcov' to view detailed line and branch coverage" +# Generate HTML report if requested +if [ "$2" == "--html" ]; then + echo -e "${YELLOW}Generating HTML coverage report...${NC}" + ./generate_coverage_html.sh + echo -e "${GREEN}HTML report generated in coverage_report/index.html${NC}" + echo "Open with: firefox coverage_report/index.html (or your preferred browser)" + echo fi -# Clean up gcov files -echo -e "${YELLOW}Cleaning up coverage files...${NC}" -rm -f *.gcov +# Clean up gcov files unless HTML was requested +if [ "$2" != "--html" ] && [ "$2" != "--detailed" ]; then + echo -e "${YELLOW}Cleaning up coverage files...${NC}" + rm -f *.gcov +fi echo echo -e "${GREEN}Coverage measurement complete!${NC}" From d288b2896f543d10bf5287131a569c390dd17750 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 19:58:14 +0000 Subject: [PATCH 14/21] Move coverage scripts to scripts/ directory and add Makefile targets Reorganize coverage tooling for better project structure and easier usage. Changes: - Move measure_coverage.sh and generate_coverage_html.sh to scripts/ - Add Makefile targets: - make coverage-report: Generate text coverage summary (5s simulation) - make coverage-html: Generate HTML coverage report (5s simulation) - Update measure_coverage.sh to find generate_coverage_html.sh using relative path Usage: make coverage-report # Quick text summary make coverage-html # Full HTML report with branch details The HTML report provides interactive visualization of which branches were taken during simulation execution, with color-coded source views. --- Makefile | 8 +++++++- .../generate_coverage_html.sh | 0 measure_coverage.sh => scripts/measure_coverage.sh | 4 +++- 3 files changed, 10 insertions(+), 2 deletions(-) rename generate_coverage_html.sh => scripts/generate_coverage_html.sh (100%) rename measure_coverage.sh => scripts/measure_coverage.sh (95%) diff --git a/Makefile b/Makefile index 74fdbff..5e4b4b0 100644 --- a/Makefile +++ b/Makefile @@ -15,12 +15,18 @@ CFILES = $(shell find src -name '*.c') HFILES = $(shell find src -name '*.h') OFILES = $(CFILES:.c=.o) -.PHONY: all clean coverage +.PHONY: all clean coverage coverage-report coverage-html all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a coverage: mousefs_random_test_coverage$(EXT) +coverage-report: + @./scripts/measure_coverage.sh 5 + +coverage-html: + @./scripts/measure_coverage.sh 5 --html + mousefs$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER diff --git a/generate_coverage_html.sh b/scripts/generate_coverage_html.sh similarity index 100% rename from generate_coverage_html.sh rename to scripts/generate_coverage_html.sh diff --git a/measure_coverage.sh b/scripts/measure_coverage.sh similarity index 95% rename from measure_coverage.sh rename to scripts/measure_coverage.sh index a10e1de..876ea3e 100755 --- a/measure_coverage.sh +++ b/scripts/measure_coverage.sh @@ -95,7 +95,9 @@ echo # Generate HTML report if requested if [ "$2" == "--html" ]; then echo -e "${YELLOW}Generating HTML coverage report...${NC}" - ./generate_coverage_html.sh + # Get the directory where this script is located + SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" + "$SCRIPT_DIR/generate_coverage_html.sh" echo -e "${GREEN}HTML report generated in coverage_report/index.html${NC}" echo "Open with: firefox coverage_report/index.html (or your preferred browser)" echo From 4c0b3429f0ef80ecfa2f664626572053341fbbab Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 20:07:57 +0000 Subject: [PATCH 15/21] Add .gitattributes to enforce LF line endings for shell scripts This fixes issues when running scripts in WSL where CRLF line endings cause 'No such file or directory' errors. Also add a fix_line_endings.sh script for users who already have the wrong line endings. --- .gitattributes | 2 ++ scripts/fix_line_endings.sh | 18 ++++++++++++++++++ 2 files changed, 20 insertions(+) create mode 100644 .gitattributes create mode 100755 scripts/fix_line_endings.sh diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..7c75d96 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,2 @@ +# Ensure shell scripts always use LF line endings +*.sh text eol=lf diff --git a/scripts/fix_line_endings.sh b/scripts/fix_line_endings.sh new file mode 100755 index 0000000..d2af20f --- /dev/null +++ b/scripts/fix_line_endings.sh @@ -0,0 +1,18 @@ +#!/bin/bash +# Fix line endings for coverage scripts + +echo "Fixing line endings for coverage scripts..." + +# Convert CRLF to LF +dos2unix scripts/measure_coverage.sh 2>/dev/null || sed -i 's/\r$//' scripts/measure_coverage.sh +dos2unix scripts/generate_coverage_html.sh 2>/dev/null || sed -i 's/\r$//' scripts/generate_coverage_html.sh + +# Ensure execute permissions +chmod +x scripts/measure_coverage.sh +chmod +x scripts/generate_coverage_html.sh + +echo "Done! Line endings fixed and execute permissions set." +echo "" +echo "You can now run:" +echo " make coverage-report" +echo " make coverage-html" From b6b926a659fcf41d850e37add28b10fda9f08ca8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 20:24:04 +0000 Subject: [PATCH 16/21] Simplify HTML coverage report styling to use minimal default styles Remove fancy styling (rounded corners, shadows, progress bars, colors) and use simple, clean HTML with basic tables and minimal CSS. Changes: - Use simple sans-serif font instead of Segoe UI - Remove container divs, shadows, and rounded corners - Replace progress bars with plain percentage text - Simplify color scheme (light green/red for branch coverage) - Remove emoji and decorative elements - Clean up table styling to basic borders The report is now much cleaner and follows standard HTML conventions. --- scripts/generate_coverage_html.sh | 241 +++++------------------------- 1 file changed, 41 insertions(+), 200 deletions(-) diff --git a/scripts/generate_coverage_html.sh b/scripts/generate_coverage_html.sh index 1eec3ae..423790d 100755 --- a/scripts/generate_coverage_html.sh +++ b/scripts/generate_coverage_html.sh @@ -14,136 +14,19 @@ cat > "$OUTPUT_DIR/index.html" <<'HTMLHEADER' Branch Coverage Report -
-

🎯 Branch Coverage Report

-

Generated from random simulation coverage analysis

+

Branch Coverage Report

+

Generated from random simulation coverage analysis

-
-

Overall Coverage Summary

+

Overall Coverage Summary

HTMLHEADER # Calculate totals @@ -167,36 +50,21 @@ done if [ "$TOTAL_BRANCHES" -gt 0 ]; then COVERAGE_PERCENT=$((TAKEN_BRANCHES * 100 / TOTAL_BRANCHES)) - # Determine coverage class - if [ "$COVERAGE_PERCENT" -lt 30 ]; then - COVERAGE_CLASS="coverage-low" - elif [ "$COVERAGE_PERCENT" -lt 70 ]; then - COVERAGE_CLASS="coverage-medium" - else - COVERAGE_CLASS="coverage-high" - fi - cat >> "$OUTPUT_DIR/index.html" < - Total Branches: $TOTAL_BRANCHES
- Branches Taken: $TAKEN_BRANCHES
- Coverage: ${COVERAGE_PERCENT}% -
-
-
-
${COVERAGE_PERCENT}%
-
-
+

+ Total Branches: $TOTAL_BRANCHES
+ Branches Taken: $TAKEN_BRANCHES
+ Coverage: ${COVERAGE_PERCENT}% +

-

Coverage by File

- - - - - - - - +

Coverage by File

+
FileBranches TakenTotal BranchesCoverageVisual
+ + + + + + HTMLSUMMARY # Generate table rows for each file @@ -212,14 +80,6 @@ HTMLSUMMARY percentage=$((taken * 100 / branches)) - if [ "$percentage" -lt 30 ]; then - bar_class="coverage-low" - elif [ "$percentage" -lt 70 ]; then - bar_class="coverage-medium" - else - bar_class="coverage-high" - fi - # Generate individual file report file_html="${filename}.html" echo "Generating report for $filename..." @@ -231,30 +91,21 @@ HTMLSUMMARY Coverage: $filename -
- ← Back to Summary -

Coverage Report: $filename

-
- Branches Taken: $taken / $branches ($percentage%)
-
-
+    

← Back to Summary

+

Coverage: $filename

+

Branches Taken: $taken / $branches ($percentage%)

+
 FILEHEADER
 
                 # Process the gcov file and add source code with branch info
@@ -308,25 +159,18 @@ FILEHEADER
 
                 cat >> "$OUTPUT_DIR/$file_html" <<'FILEFOOTER'
 
-
FILEFOOTER # Add row to main index cat >> "$OUTPUT_DIR/index.html" < - - - - - - + + + + + + TABLEROW fi fi @@ -334,17 +178,14 @@ TABLEROW # Close HTML cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' -
FileBranches TakenTotal BranchesCoverage %
$filename$taken$branches${percentage}% -
-
-
${percentage}%
-
-
$filename$taken$branches${percentage}%
- + HTMLFOOTER else cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' -

No branch coverage data found.

- - +

No branch coverage data found.

HTMLFOOTER From abaa12c5b35c20b60975b9edb3fe3cee2acd664e Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Nov 2025 20:26:20 +0000 Subject: [PATCH 17/21] Remove all CSS styling from HTML reports, use browser defaults Remove all custom styling and CSS to use only plain HTML with browser default rendering. Changes: - Remove all

Branch Coverage Report

@@ -90,19 +82,9 @@ HTMLSUMMARY Coverage: $filename - -

← Back to Summary

+

Back to Summary

Coverage: $filename

Branches Taken: $taken / $branches ($percentage%)

@@ -124,34 +106,22 @@ FILEHEADER
                         # Get the actual source line
                         line_content = substr($0, index($0, $3))
 
-                        # Determine background color based on branch coverage
-                        bg_class = ""
-                        if (branch_info != "") {
-                            if (branch_info ~ /taken 0/) {
-                                bg_class = "branch-not-taken"
-                            } else if (branch_info ~ /never executed/) {
-                                bg_class = "branch-not-taken"
-                            } else {
-                                bg_class = "branch-taken"
-                            }
-                        }
-
                         # Format execution count
                         if (exec_count == "-") {
                             exec_str = "-"
                         } else if (exec_count == "#####") {
                             exec_str = "0"
-                            if (bg_class == "") bg_class = "branch-not-taken"
                         } else {
                             exec_str = exec_count
                         }
 
-                        printf "%s%s%s", bg_class, line_num, exec_str, line_content
+                        # Output line number, execution count, and source
+                        printf "%6s: %5s:%s", line_num, exec_str, line_content
 
                         if (branch_info != "") {
-                            printf "%s", branch_info
+                            printf " [%s]", branch_info
                         }
-                        printf "\n"
+                        printf "\n"
 
                         branch_info = ""
                     }

From d4ecf6db17580da2e9eddc2d146cb7e1dab6fdf0 Mon Sep 17 00:00:00 2001
From: Claude 
Date: Thu, 13 Nov 2025 20:41:28 +0000
Subject: [PATCH 18/21] Replace custom HTML generation with lcov/genhtml

Use the standard lcov tools instead of custom HTML generation.

Changes:
- Completely rewrite generate_coverage_html.sh to use lcov
- Install lcov if not available
- Use lcov to capture coverage data from .gcda files
- Use genhtml to create HTML reports with branch coverage
- Add scripts/README.md with usage instructions

Benefits:
- Professional HTML reports with standard lcov styling
- Better branch coverage visualization
- Sortable tables by line/function/branch coverage
- Source code view with execution counts
- Much simpler script (30 lines vs 200+ lines)

The reports now show:
- 32.1% branch coverage (781/2431 branches)
- 43.5% line coverage
- 61.3% function coverage
---
 scripts/README.md                 |  48 ++++++++
 scripts/generate_coverage_html.sh | 177 ++++--------------------------
 2 files changed, 70 insertions(+), 155 deletions(-)
 create mode 100644 scripts/README.md

diff --git a/scripts/README.md b/scripts/README.md
new file mode 100644
index 0000000..bcaac89
--- /dev/null
+++ b/scripts/README.md
@@ -0,0 +1,48 @@
+# Branch Coverage Measurement
+
+This directory contains scripts for measuring branch coverage of the random simulation.
+
+## Prerequisites
+
+The HTML report generation requires `lcov`:
+
+```bash
+sudo apt-get install lcov
+```
+
+## Usage
+
+### Quick text summary (5 second simulation)
+```bash
+make coverage-report
+```
+
+### HTML report with branch details (5 second simulation)
+```bash
+make coverage-html
+```
+
+### Custom duration
+```bash
+./scripts/measure_coverage.sh 10          # 10 second run, text output
+./scripts/measure_coverage.sh 10 --html   # 10 second run, HTML output
+```
+
+## What gets measured
+
+The coverage tool:
+- Builds the test binary with coverage instrumentation (`--coverage` flag)
+- Runs the random simulation for the specified duration
+- Generates reports showing which branches were executed
+
+## Output
+
+**Text report**: Shows per-file and total branch coverage percentages
+
+**HTML report**: Interactive report generated by lcov/genhtml showing:
+- Overall coverage summary
+- Per-file coverage breakdown
+- Source code with execution counts
+- Branch coverage details
+
+The HTML report is generated in `coverage_report/index.html`
diff --git a/scripts/generate_coverage_html.sh b/scripts/generate_coverage_html.sh
index 04a8c33..c29788b 100755
--- a/scripts/generate_coverage_html.sh
+++ b/scripts/generate_coverage_html.sh
@@ -1,164 +1,31 @@
 #!/bin/bash
-# Generate HTML coverage report from gcov files
+# Generate HTML coverage report using lcov
 
 set -e
 
 OUTPUT_DIR="coverage_report"
-mkdir -p "$OUTPUT_DIR"
 
-# Create main index.html
-cat > "$OUTPUT_DIR/index.html" <<'HTMLHEADER'
-
-
-
-    
-    Branch Coverage Report
-
-
-    

Branch Coverage Report

-

Generated from random simulation coverage analysis

- -

Overall Coverage Summary

-HTMLHEADER - -# Calculate totals -TOTAL_BRANCHES=0 -TAKEN_BRANCHES=0 - -for gcov_file in *.c.gcov; do - if [ -f "$gcov_file" ]; then - branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null || echo "0") - if [ "$branches" -gt 0 ]; then - taken=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") - taken=$(echo "$taken" | tr -d '[:space:]') - if [ -z "$taken" ]; then taken=0; fi - - TOTAL_BRANCHES=$((TOTAL_BRANCHES + branches)) - TAKEN_BRANCHES=$((TAKEN_BRANCHES + taken)) - fi - fi -done - -if [ "$TOTAL_BRANCHES" -gt 0 ]; then - COVERAGE_PERCENT=$((TAKEN_BRANCHES * 100 / TOTAL_BRANCHES)) - - cat >> "$OUTPUT_DIR/index.html" < - Total Branches: $TOTAL_BRANCHES
- Branches Taken: $TAKEN_BRANCHES
- Coverage: ${COVERAGE_PERCENT}% -

- -

Coverage by File

- - - - - - - -HTMLSUMMARY - - # Generate table rows for each file - for gcov_file in *.c.gcov; do - if [ -f "$gcov_file" ]; then - filename=$(echo "$gcov_file" | sed 's/.gcov$//') - branches=$(grep -c "^branch" "$gcov_file" 2>/dev/null || echo "0") - - if [ "$branches" -gt 0 ]; then - taken=$(grep "^branch" "$gcov_file" 2>/dev/null | grep -c "taken [1-9]" 2>/dev/null || echo "0") - taken=$(echo "$taken" | tr -d '[:space:]') - if [ -z "$taken" ]; then taken=0; fi - - percentage=$((taken * 100 / branches)) - - # Generate individual file report - file_html="${filename}.html" - echo "Generating report for $filename..." - - cat > "$OUTPUT_DIR/$file_html" < - - - - Coverage: $filename - - -

Back to Summary

-

Coverage: $filename

-

Branches Taken: $taken / $branches ($percentage%)

-
-FILEHEADER
-
-                # Process the gcov file and add source code with branch info
-                awk '
-                    /^branch/ {
-                        branch_info = branch_info " " $0
-                        next
-                    }
-                    /^        / {
-                        # Line with execution count
-                        line_num = $2
-                        gsub(/:/, "", line_num)
-                        exec_count = $1
-                        gsub(/^[ \t]+/, "", exec_count)
-
-                        # Get the actual source line
-                        line_content = substr($0, index($0, $3))
-
-                        # Format execution count
-                        if (exec_count == "-") {
-                            exec_str = "-"
-                        } else if (exec_count == "#####") {
-                            exec_str = "0"
-                        } else {
-                            exec_str = exec_count
-                        }
-
-                        # Output line number, execution count, and source
-                        printf "%6s: %5s:%s", line_num, exec_str, line_content
-
-                        if (branch_info != "") {
-                            printf " [%s]", branch_info
-                        }
-                        printf "\n"
-
-                        branch_info = ""
-                    }
-                ' "$gcov_file" >> "$OUTPUT_DIR/$file_html"
-
-                cat >> "$OUTPUT_DIR/$file_html" <<'FILEFOOTER'
-
- - -FILEFOOTER - - # Add row to main index - cat >> "$OUTPUT_DIR/index.html" < -
- - - - -TABLEROW - fi - fi - done - - # Close HTML - cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' -
FileBranches TakenTotal BranchesCoverage %
$filename$taken$branches${percentage}%
- - -HTMLFOOTER - -else - cat >> "$OUTPUT_DIR/index.html" <<'HTMLFOOTER' -

No branch coverage data found.

- - -HTMLFOOTER +# Check if lcov is installed +if ! command -v lcov &> /dev/null; then + echo "Error: lcov is not installed" + echo "Please install it with: sudo apt-get install lcov" + exit 1 fi +# Capture coverage data with branch coverage enabled +echo "Capturing coverage data..." +lcov --capture --directory . --output-file coverage.info --rc lcov_branch_coverage=1 + +# Filter out system headers if any exist +echo "Filtering coverage data..." +lcov --remove coverage.info '/usr/*' --output-file coverage.info --ignore-errors unused --rc lcov_branch_coverage=1 || cp coverage.info coverage.info.bak + +# Generate HTML report +echo "Generating HTML report..." +genhtml coverage.info --output-directory "$OUTPUT_DIR" --branch-coverage --rc lcov_branch_coverage=1 + +# Clean up +rm -f coverage.info + echo "HTML report generated in $OUTPUT_DIR/" +echo "Open $OUTPUT_DIR/index.html in your browser" From 3048f3d66be466ff1fdfca5ec30f4d97dafbfca3 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 23:02:25 +0100 Subject: [PATCH 19/21] Bug fixes --- src/basic.c | 6 +- src/chunk_server.c | 17 +++-- src/chunk_server.h | 1 + src/hash_set.c | 24 ++++++- src/metadata_server.c | 2 +- src/system.c | 144 ++++++++++++++++++++++-------------------- 6 files changed, 115 insertions(+), 79 deletions(-) diff --git a/src/basic.c b/src/basic.c index d6a72f0..9c81fd6 100644 --- a/src/basic.c +++ b/src/basic.c @@ -30,7 +30,7 @@ Time get_current_time(void) ok = sys_QueryPerformanceFrequency((LARGE_INTEGER*) &freq); if (!ok) return INVALID_TIME; - uint64_t res = 1000 * (double) count / freq; + uint64_t res = 1000000000 * (double) count / freq; return res; } #else @@ -45,12 +45,12 @@ Time get_current_time(void) uint64_t sec = time.tv_sec; if (sec > UINT64_MAX / 1000000000) return INVALID_TIME; - res = sec * 1000; + res = sec * 1000000000; uint64_t nsec = time.tv_nsec; if (res > UINT64_MAX - nsec) return INVALID_TIME; - res += nsec / 1000000; + res += nsec; return res; } diff --git a/src/chunk_server.c b/src/chunk_server.c index 24131cd..5ea2fe0 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -812,6 +812,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts if (remote_port <= 0 || remote_port >= 1<<16) return -1; + Time current_time = get_current_time(); + state->trace = trace; tcp_context_init(&state->tcp); @@ -865,6 +867,7 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts } state->remote_addr.port = remote_port; state->disconnect_time = INVALID_TIME; + state->last_sync_time = current_time; start_connecting_to_metadata_server(state); @@ -988,11 +991,15 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } - // TODO: periodically send a SYNC message - bool should_sync = false; - if (should_sync) { - if (send_sync_message(state) < 0) { - assert(0); // TODO + if (state->disconnect_time == INVALID_TIME) { + Time next_sync_time = state->last_sync_time + (Time) SYNC_INTERVAL * 1000000000; + if (current_time >= next_sync_time) { + if (send_sync_message(state) < 0) { + assert(0); // TODO + } + state->last_sync_time = current_time; + } else { + nearest_deadline(&deadline, next_sync_time); } } diff --git a/src/chunk_server.h b/src/chunk_server.h index e3c7854..34e89b8 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -35,6 +35,7 @@ typedef struct { Address remote_addr; Time disconnect_time; + Time last_sync_time; TCP tcp; diff --git a/src/hash_set.c b/src/hash_set.c index d4075d8..a2ea582 100644 --- a/src/hash_set.c +++ b/src/hash_set.c @@ -72,12 +72,32 @@ bool hash_set_contains(HashSet *set, SHA256 hash) int hash_set_merge(HashSet *dst, HashSet src) { - assert(0); // TODO + HashSet ret; + hash_set_init(&ret); + + for (int i = 0; i < dst->count; i++) { + if (hash_set_insert(&ret, dst->items[i]) < 0) + goto error; + } + + for (int i = 0; i < src.count; i++) { + if (hash_set_insert(&ret, src.items[i]) < 0) + goto error; + } + + hash_set_free(dst); + *dst = ret; + return 0; + +error: + hash_set_free(&ret); + return -1; } void hash_set_remove_set(HashSet *dst, HashSet src) { - assert(0); // TODO + for (int i = 0; i < src.count; i++) + hash_set_remove(dst, src.items[i]); } void timed_hash_set_init(TimedHashSet *set) diff --git a/src/metadata_server.c b/src/metadata_server.c index 5533cbe..7b0688b 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -1088,7 +1088,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * Time response_timeout = chunk_server->last_response_time + (Time) RESPONSE_TIME_LIMIT * 1000000000; if (current_time > response_timeout) { - // TODO: drop the chunk server + assert(0); // TODO: drop the chunk server continue; } nearest_deadline(&next_wakeup, response_timeout); diff --git a/src/system.c b/src/system.c index be02667..6a7fecd 100644 --- a/src/system.c +++ b/src/system.c @@ -580,6 +580,77 @@ static int compare_processes(const void *p1, const void *p2) return a->wakeup_time - b->wakeup_time; } +static int setup_poll_array(void *contexts, struct pollfd *polled) +{ + int num_polled = 0; + + for (int j = 0, k = 0; k < current_process->num_desc; j++) { + + assert(j < MAX_DESCRIPTORS); + Descriptor *desc = ¤t_process->desc[j]; + if (desc->type == DESC_EMPTY) + continue; + k++; + + int revents = 0; + switch (desc->type) { + + case DESC_FILE: + assert(0); // TODO: error + break; + + case DESC_SOCKET: + // Ignore + break; + + case DESC_LISTEN_SOCKET: + if (!accept_queue_empty(&desc->accept_queue)) + revents |= POLLIN; + break; + + case DESC_CONNECTION_SOCKET: + switch (desc->connection_state) { + + case CONNECTION_DELAYED: + break; + + case CONNECTION_QUEUED: + break; + + case CONNECTION_ESTABLISHED: + { + Descriptor *peer = handle_to_desc(desc->connection_peer); + if (peer == NULL) { + revents |= POLLIN; + } else { + if (!data_queue_full(&desc->output_data)) + revents |= POLLOUT; + if (!data_queue_empty(&peer->output_data)) + revents |= POLLIN; + } + } + break; + + case CONNECTION_FAILED: + assert(0); // TODO + break; + } + break; + } + + revents &= desc->events; + if (revents) { + polled[num_polled].fd = (SOCKET) j; + polled[num_polled].events = desc->events; + polled[num_polled].revents = revents; + contexts[num_polled] = desc->context; + num_polled++; + } + } + + return num_polled; +} + void update_simulation(void) { // Order processes by wakeup time. Those with no @@ -596,76 +667,13 @@ void update_simulation(void) void *contexts[MAX_CONNS+1]; struct pollfd polled[MAX_CONNS+1]; - int num_polled = 0; - - for (int j = 0, k = 0; k < current_process->num_desc; j++) { - - assert(j < MAX_DESCRIPTORS); - Descriptor *desc = ¤t_process->desc[j]; - if (desc->type == DESC_EMPTY) - continue; - k++; - - int revents = 0; - switch (desc->type) { - - case DESC_FILE: - assert(0); // TODO: error - break; - - case DESC_SOCKET: - // Ignore - break; - - case DESC_LISTEN_SOCKET: - if (!accept_queue_empty(&desc->accept_queue)) - revents |= POLLIN; - break; - - case DESC_CONNECTION_SOCKET: - switch (desc->connection_state) { - - case CONNECTION_DELAYED: - break; - - case CONNECTION_QUEUED: - break; - - case CONNECTION_ESTABLISHED: - { - Descriptor *peer = handle_to_desc(desc->connection_peer); - if (peer == NULL) { - revents |= POLLIN; - } else { - if (!data_queue_full(&desc->output_data)) - revents |= POLLOUT; - if (!data_queue_empty(&peer->output_data)) - revents |= POLLIN; - } - } - break; - - case CONNECTION_FAILED: - assert(0); // TODO - break; - } - break; - } - - revents &= desc->events; - if (revents) { - polled[num_polled].fd = (SOCKET) j; - polled[num_polled].events = desc->events; - polled[num_polled].revents = revents; - contexts[num_polled] = desc->context; - num_polled++; - } - } + int num_polled = setup_poll_array(contexts, polled); if (num_polled == 0) { Time wakeup_time = current_process->wakeup_time; - if (wakeup_time == INVALID_TIME) continue; + if (wakeup_time == INVALID_TIME) + continue; assert(current_time <= wakeup_time); current_time = wakeup_time; @@ -703,13 +711,13 @@ void update_simulation(void) } if (num_polled < 0) { - // TODO + assert(0); // TODO } if (timeout < 0) { current_process->wakeup_time = INVALID_TIME; } else { - current_process->wakeup_time = current_time + timeout * 1000000; + current_process->wakeup_time = current_time + (Time) timeout * 1000000; } process_poll_array(current_process, contexts, polled, num_polled); From 3a2173c1547b91a4e5ab375cb7f3deff50636044 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Thu, 13 Nov 2025 23:37:42 +0100 Subject: [PATCH 20/21] Minor bug fixes --- src/metadata_server.c | 2 +- src/system.c | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/src/metadata_server.c b/src/metadata_server.c index 7b0688b..7ee59fe 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -106,7 +106,7 @@ static int find_chunk_server_by_addr(MetadataServer *state, Address addr) for (int i = 0; i < state->num_chunk_servers; i++) for (int j = 0; j < state->chunk_servers[i].num_addrs; j++) if (addr_eql(state->chunk_servers[i].addrs[j], addr)) - return j; + return i; return -1; } diff --git a/src/system.c b/src/system.c index 6a7fecd..94001ac 100644 --- a/src/system.c +++ b/src/system.c @@ -577,10 +577,12 @@ static int compare_processes(const void *p1, const void *p2) Process *b = *(Process**) p2; if (b->wakeup_time == INVALID_TIME) return -1; if (a->wakeup_time == INVALID_TIME) return +1; - return a->wakeup_time - b->wakeup_time; + if (a->wakeup_time < b->wakeup_time) return -1; + if (a->wakeup_time > b->wakeup_time) return +1; + return 0; } -static int setup_poll_array(void *contexts, struct pollfd *polled) +static int setup_poll_array(void **contexts, struct pollfd *polled) { int num_polled = 0; From 9a944feb01f097f768c9c1dbe6f86e027166baa8 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Fri, 14 Nov 2025 01:16:48 +0100 Subject: [PATCH 21/21] Rewrote simulation scheduler and fixed some bugs --- Makefile | 6 +- src/chunk_server.c | 15 ++- src/metadata_server.c | 10 +- src/system.c | 221 ++++++++++++++++++++++++++---------------- src/tcp.c | 8 +- src/tcp.h | 1 + 6 files changed, 164 insertions(+), 97 deletions(-) diff --git a/Makefile b/Makefile index 5e4b4b0..ee34697 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ -CFLAGS = -Wall -Wextra -ggdb +CFLAGS = -Wall -Wextra -ggdb -fsanitize=address,undefined COVERAGE_CFLAGS = $(CFLAGS) --coverage COVERAGE_LFLAGS = --coverage @@ -22,10 +22,10 @@ all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a coverage: mousefs_random_test_coverage$(EXT) coverage-report: - @./scripts/measure_coverage.sh 5 + @./scripts/measure_coverage.sh 60 coverage-html: - @./scripts/measure_coverage.sh 5 --html + @./scripts/measure_coverage.sh 60 --html mousefs$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER diff --git a/src/chunk_server.c b/src/chunk_server.c index 5ea2fe0..ecca8c7 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -871,6 +871,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts start_connecting_to_metadata_server(state); + // TODO: add all chunk hashes to the add list + printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n", addr.len, addr.ptr, @@ -911,14 +913,16 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled switch (events[i].type) { case EVENT_CONNECT: - if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) - state->disconnect_time = INVALID_TIME; + if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) { + assert(state->disconnect_time == INVALID_TIME); + } break; case EVENT_DISCONNECT: - switch (tcp_get_tag(&state->tcp, conn_idx)) { + switch (events[i].tag) { case TAG_METADATA_SERVER: + assert(state->disconnect_time == INVALID_TIME); state->disconnect_time = current_time; break; @@ -977,6 +981,11 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } + { + int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); + assert((conn_idx < 0) == (state->disconnect_time != INVALID_TIME)); + } + Time deadline = INVALID_TIME; // Remove items from the remove list that got too old diff --git a/src/metadata_server.c b/src/metadata_server.c index 7ee59fe..4338b7f 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -870,6 +870,8 @@ static int process_chunk_server_sync_3(MetadataServer *state, HashSet tmp_list; hash_set_init(&tmp_list); + message_write(&writer, &count, sizeof(count)); + for (uint32_t i = 0; i < count; i++) { SHA256 hash; @@ -933,6 +935,7 @@ static bool is_chunk_server_message_type(uint16_t type) switch (type) { case MESSAGE_TYPE_AUTH: case MESSAGE_TYPE_SYNC: + case MESSAGE_TYPE_SYNC_3: return true; default: @@ -1010,9 +1013,9 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * case EVENT_DISCONNECT: { - int tag = tcp_get_tag(&state->tcp, conn_idx); - if (tag >= 0) { - chunk_server_peer_free(&state->chunk_servers[tag]); + if (events[i].tag >= 0) { + chunk_server_peer_free(&state->chunk_servers[events[i].tag]); + assert(state->num_chunk_servers > 0); state->num_chunk_servers--; } } @@ -1048,6 +1051,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * } chunk_server_peer_init(&state->chunk_servers[j], current_time); + state->num_chunk_servers++; tcp_set_tag(&state->tcp, conn_idx, j, true); diff --git a/src/system.c b/src/system.c index 94001ac..9ae26c2 100644 --- a/src/system.c +++ b/src/system.c @@ -571,17 +571,6 @@ static bool data_queue_full(DataQueue *queue) return queue->used == queue->size; } -static int compare_processes(const void *p1, const void *p2) -{ - Process *a = *(Process**) p1; - Process *b = *(Process**) p2; - if (b->wakeup_time == INVALID_TIME) return -1; - if (a->wakeup_time == INVALID_TIME) return +1; - if (a->wakeup_time < b->wakeup_time) return -1; - if (a->wakeup_time > b->wakeup_time) return +1; - return 0; -} - static int setup_poll_array(void **contexts, struct pollfd *polled) { int num_polled = 0; @@ -653,80 +642,8 @@ static int setup_poll_array(void **contexts, struct pollfd *polled) return num_polled; } -void update_simulation(void) +static void update_network(void) { - // Order processes by wakeup time. Those with no - // wakeup time go last. - Process *ordered_processes[MAX_PROCESSES]; - for (int i = 0; i < num_processes; i++) - ordered_processes[i] = processes[i]; - - qsort(ordered_processes, num_processes, sizeof(*ordered_processes), compare_processes); - - for (int i = 0; i < num_processes; i++) { - - current_process = ordered_processes[i]; - - void *contexts[MAX_CONNS+1]; - struct pollfd polled[MAX_CONNS+1]; - int num_polled = setup_poll_array(contexts, polled); - - if (num_polled == 0) { - - Time wakeup_time = current_process->wakeup_time; - if (wakeup_time == INVALID_TIME) - continue; - - assert(current_time <= wakeup_time); - current_time = wakeup_time; - } - - int timeout = -1; - switch (current_process->type) { - case PROCESS_TYPE_METADATA_SERVER: - num_polled = metadata_server_step( - ¤t_process->metadata_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CHUNK_SERVER: - num_polled = chunk_server_step( - ¤t_process->chunk_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CLIENT: - num_polled = simulation_client_step( - ¤t_process->simulation_client, - contexts, - polled, - num_polled, - &timeout - ); - break; - } - - if (num_polled < 0) { - assert(0); // TODO - } - - if (timeout < 0) { - current_process->wakeup_time = INVALID_TIME; - } else { - current_process->wakeup_time = current_time + (Time) timeout * 1000000; - } - - process_poll_array(current_process, contexts, polled, num_polled); - - current_process = NULL; - } - for (int i = 0; i < num_processes; i++) { for (int j = 0, k = 0; k < processes[i]->num_desc; j++) { @@ -786,6 +703,142 @@ void update_simulation(void) } } +void update_simulation(void) +{ + for (;;) { + int num_ready = 0; + for (int i = 0; i < num_processes; i++) { + + current_process = processes[i]; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + if (num_polled > 0) { + + num_ready++; + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + } + + current_process = NULL; + + update_network(); + } + + if (num_ready == 0) + break; + } + + Process *next_process = NULL; + for (int i = 0; i < num_processes; i++) + if (processes[i]->wakeup_time != INVALID_TIME) + if (next_process == NULL || processes[i]->wakeup_time < next_process->wakeup_time) + next_process = processes[i]; + + if (next_process == NULL) { + assert(0); // Nothing to schedule next + } + + assert(current_time <= next_process->wakeup_time); + current_time = next_process->wakeup_time; + + current_process = next_process; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + + current_process = NULL; + + update_network(); +} + void *mock_malloc(size_t len) { return malloc(len); diff --git a/src/tcp.c b/src/tcp.c index 1affff1..36f16eb 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -240,8 +240,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd if (set_socket_blocking(new_fd, false) < 0) CLOSE_SOCKET(new_fd); else { - events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns }; conn_init(&tcp->conns[tcp->num_conns++], new_fd, false); + events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns-1, tcp->conns[tcp->num_conns-1].tag }; } } } @@ -266,7 +266,7 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd defer_close = true; else { conn->connecting = false; - events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns }; + events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns, conn->tag }; } } @@ -319,8 +319,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd removed[i] = defer_close; if (0) {} - else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns }; - else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns }; + else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns, conn->tag }; + else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns, conn->tag }; } } diff --git a/src/tcp.h b/src/tcp.h index 015bc7d..2a41a19 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -23,6 +23,7 @@ typedef enum { typedef struct { EventType type; int conn_idx; + int tag; } Event; typedef struct {