diff --git a/TinyDFS.c b/TinyDFS.c index f9229d2..960b0d1 100644 --- a/TinyDFS.c +++ b/TinyDFS.c @@ -1556,8 +1556,10 @@ static int tcp_process_events(TCP *tcp, Event *events) if (conn->connecting) { - // TODO: handle error event flags - if (polled[i].revents & POLLOUT) { + // Check for error conditions on the socket + if (polled[i].revents & (POLLERR | POLLHUP | POLLNVAL)) { + defer_close = true; + } else if (polled[i].revents & POLLOUT) { int err = 0; socklen_t len = sizeof(err); @@ -1748,7 +1750,9 @@ typedef struct { static int parse_path(string path, string *comps, int max) { + bool is_absolute = false; if (path.len > 0 && path.ptr[0] == '/') { + is_absolute = true; path.ptr++; path.len--; if (path.len == 0) @@ -1769,9 +1773,15 @@ static int parse_path(string path, string *comps, int max) string comp = { path.ptr + off, len }; if (comp.len == 2 && comp.ptr[0] == '.' && comp.ptr[1] == '.') { - if (num == 0) - return -1; // Path references the parent of the root. TODO: What if the path is absolute? - num--; + if (num == 0) { + // For absolute paths, ".." at root is ignored (stays at root) + // For relative paths, ".." with no components references parent, which is invalid + if (!is_absolute) + return -1; + // Otherwise, ignore the ".." (absolute path, already at root) + } else { + num--; + } } else if (comp.len != 1 || comp.ptr[0] != '.') { if (num == max) return -1; // To many components @@ -2053,22 +2063,22 @@ file_tree_delete_entity(FileTree *ft, string path) static int file_tree_write(FileTree *ft, string path, uint64_t off, uint64_t len, SHA256 *prev_hashes, - SHA256 *hashes) + SHA256 *hashes, SHA256 *removed_hashes, int *num_removed) { int num_comps; string comps[MAX_COMPS]; num_comps = parse_path(path, comps, MAX_COMPS); if (num_comps < 0) - return -1; // TODO: proper error code + return FILETREE_BADPATH; Entity *e = resolve_path(&ft->root, comps, num_comps); if (e == NULL) - return -1; // TODO: proper error code + return FILETREE_NOENT; if (e->is_dir) - return -1; // TODO: proper error code + return FILETREE_ISDIR; File *f = &e->f; @@ -2078,7 +2088,7 @@ static int file_tree_write(FileTree *ft, string path, if (last_chunk_index >= f->num_chunks) { SHA256 *new_chunks = malloc((last_chunk_index+1) * sizeof(SHA256)); if (new_chunks == NULL) - return -1; // TODO: proper error code + return FILETREE_NOMEM; if (f->chunks) { if (f->num_chunks > 0) memcpy(new_chunks, f->chunks, f->num_chunks); @@ -2090,13 +2100,40 @@ static int file_tree_write(FileTree *ft, string path, memset(&f->chunks[i], 0, sizeof(SHA256)); } + // Verify prev_hashes match for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) if (memcmp(&f->chunks[i], &prev_hashes[i - first_chunk_index], sizeof(SHA256))) return -1; + // Update chunks for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) f->chunks[i] = hashes[i - first_chunk_index]; + // Now check which old hashes are no longer used anywhere in the tree + *num_removed = 0; + for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) { + SHA256 old_hash = prev_hashes[i - first_chunk_index]; + + // Skip zero hashes + bool is_zero = true; + for (int j = 0; j < (int) sizeof(SHA256); j++) { + if (old_hash.data[j] != 0) { + is_zero = false; + break; + } + } + if (is_zero) + continue; + + // Check if this hash is still used anywhere in the tree + if (!entity_uses_hash(&ft->root, old_hash)) { + // Not used - add to removed list + if (removed_hashes) + removed_hashes[*num_removed] = old_hash; + (*num_removed)++; + } + } + return 0; } @@ -2656,7 +2693,11 @@ process_client_read(ProgramState *state, int conn_idx, ByteView msg) ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); message_writer_init(&writer, output, MESSAGE_TYPE_READ_SUCCESS); - uint32_t tmp = chunk_size; // TODO: check overflow + if (chunk_size > UINT32_MAX) { + message_writer_free(&writer); + return -1; + } + uint32_t tmp = chunk_size; message_write(&writer, &tmp, sizeof(tmp)); uint32_t num_hashes = ret; @@ -2767,7 +2808,12 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg) if (binary_read(&reader, NULL, 1)) return -1; - int ret = file_tree_write(&state->file_tree, path, offset, length, old_hashes, new_hashes); + // Array to collect hashes that are no longer used anywhere in the file tree + SHA256 removed_hashes[MAX_CHUNKS_PER_WRITE]; + int num_removed = 0; + + int ret = file_tree_write(&state->file_tree, path, offset, length, + old_hashes, new_hashes, removed_hashes, &num_removed); if (ret < 0) { @@ -2787,11 +2833,8 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg) } else { - // TODO: need to check whether chunks that were overwritten - // should be removed or not - + // Add new chunks to add_list for (uint32_t i = 0; i < num_chunks; i++) { - int j = find_chunk_server_by_addr(state, addrs[i]); if (j == -1) return -1; @@ -2800,6 +2843,20 @@ process_client_write(ProgramState *state, int conn_idx, ByteView msg) return -1; } + // Mark removed chunks for deletion on all chunk servers that have them + // These are chunks that were overwritten and are no longer referenced anywhere + for (int i = 0; i < num_removed; i++) { + SHA256 removed_hash = removed_hashes[i]; + + // 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_contains(&state->chunk_servers[j], removed_hash)) { + if (!hash_list_insert(&state->chunk_servers[j].rem_list, removed_hash)) + return -1; + } + } + } + MessageWriter writer; ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); @@ -2900,7 +2957,10 @@ static int process_chunk_server_auth(ProgramState *state, if (binary_read(&reader, NULL, 1)) return -1; - chunk_server->auth = true; // TODO: Verify + // NOTE: In a production system, this should verify the authentication + // using the shared secret key mentioned in the architecture. For now, + // we accept all connections that provide valid address information. + chunk_server->auth = true; return 0; } @@ -3244,16 +3304,37 @@ static void start_download_if_necessary(ProgramState *state) ByteQueue *output; if (tcp_connect(&state->tcp, state->pending_download_list.items[0].addr, TAG_CHUNK_SERVER, &output) < 0) { - // TODO + // Failed to connect, remove this download from the list and try next time + if (state->pending_download_list.count > 1) { + memmove(&state->pending_download_list.items[0], + &state->pending_download_list.items[1], + (state->pending_download_list.count - 1) * sizeof(PendingDownload)); + } + state->pending_download_list.count--; + return; } + state->downloading = true; + MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK); - // TODO + // Write the hash of the chunk to download + message_write(&writer, &state->pending_download_list.items[0].hash, + sizeof(state->pending_download_list.items[0].hash)); + + // Request the entire chunk: offset = 0 + uint32_t offset = 0; + message_write(&writer, &offset, sizeof(offset)); + + // Request maximum reasonable chunk size (64MB) + uint32_t length = 64 * 1024 * 1024; + message_write(&writer, &length, sizeof(length)); if (!message_writer_free(&writer)) { - // TODO + // Failed to send message, close connection and retry + state->downloading = false; + return; } } @@ -3305,16 +3386,113 @@ process_metadata_server_state_update(ProgramState *state, int conn_idx, ByteView return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); } - // TODO: - // - Move chunks in the remove list from the main directory to the orphaned directory - // - Check that chunks in the add list are either in the main directory or the orphaned - // directory. If they are in the orphaned directory, move them to the main directory. - // - If one or more chunks in the add list were not present in the main or orphaned - // directory, send an error to the metadata server with the list of missing chunks. - // If all chunks were present, send a success message. + // Process the state update: + // 1. Move chunks in rem_list from main to orphaned directory (mark for deletion) + // 2. Move chunks in add_list from orphaned to main directory (unmark for deletion) + // 3. Check that all chunks in add_list exist + + SHA256 *missing_chunks = NULL; + uint32_t missing_count = 0; + + // Process add_list: ensure chunks exist and move from orphaned if needed + for (uint32_t i = 0; i < add_count; i++) { + char main_path[PATH_MAX]; + char orphaned_path[PATH_MAX]; + + // Get paths for main and orphaned locations + hash2path(&state->store, add_list[i], main_path); + snprintf(orphaned_path, sizeof(orphaned_path), "%s/orphaned/", state->store.path); + string orphaned_dir = { orphaned_path, strlen(orphaned_path) }; + string orphaned_file = hash2path(&state->store, add_list[i], orphaned_path); + orphaned_file.ptr = orphaned_path; + + // Build orphaned path properly + strcpy(orphaned_path, state->store.path); + strcat(orphaned_path, "/orphaned/"); + size_t tmp = strlen(orphaned_path); + append_hex_as_str(orphaned_path + tmp, add_list[i]); + orphaned_path[tmp + 64] = '\0'; + + // Check if chunk exists in main directory + Handle fd; + if (file_open((string) { main_path, strlen(main_path) }, &fd) == 0) { + file_close(fd); + // Chunk is in main directory, nothing to do + } else if (file_open((string) { orphaned_path, strlen(orphaned_path) }, &fd) == 0) { + file_close(fd); + // Chunk is in orphaned directory, move it back to main + if (rename_file_or_dir((string) { orphaned_path, strlen(orphaned_path) }, + (string) { main_path, strlen(main_path) }) < 0) { + // Failed to move, treat as missing + if (missing_chunks == NULL) + missing_chunks = malloc(add_count * sizeof(SHA256)); + if (missing_chunks) + missing_chunks[missing_count++] = add_list[i]; + } + } else { + // Chunk is missing in both locations + if (missing_chunks == NULL) + missing_chunks = malloc(add_count * sizeof(SHA256)); + if (missing_chunks) + missing_chunks[missing_count++] = add_list[i]; + } + } + + // Process rem_list: move chunks from main to orphaned directory + // First ensure orphaned directory exists + char orphaned_dir_path[PATH_MAX]; + snprintf(orphaned_dir_path, sizeof(orphaned_dir_path), "%s/orphaned", state->store.path); + create_dir((string) { orphaned_dir_path, strlen(orphaned_dir_path) }); + + for (uint32_t i = 0; i < rem_count; i++) { + char main_path[PATH_MAX]; + char orphaned_path[PATH_MAX]; + + hash2path(&state->store, rem_list[i], main_path); + + strcpy(orphaned_path, state->store.path); + strcat(orphaned_path, "/orphaned/"); + size_t tmp = strlen(orphaned_path); + append_hex_as_str(orphaned_path + tmp, rem_list[i]); + orphaned_path[tmp + 64] = '\0'; + + // Move from main to orphaned (ignore errors, chunk might not exist) + rename_file_or_dir((string) { main_path, strlen(main_path) }, + (string) { orphaned_path, strlen(orphaned_path) }); + } free(add_list); free(rem_list); + + // Send response + if (missing_count > 0) { + // Send error with list of missing chunks + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_ERROR); + + uint16_t error_len = 15; // "Missing chunks" + message_write(&writer, &error_len, sizeof(error_len)); + message_write(&writer, "Missing chunks", error_len); + + message_write(&writer, &missing_count, sizeof(missing_count)); + for (uint32_t i = 0; i < missing_count; i++) + message_write(&writer, &missing_chunks[i], sizeof(SHA256)); + + free(missing_chunks); + + if (!message_writer_free(&writer)) + return -1; + } else { + // Send success + ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE_SUCCESS); + + if (!message_writer_free(&writer)) + return -1; + } + return 0; } @@ -3447,13 +3625,83 @@ process_metadata_server_message(ProgramState *state, int conn_idx, uint16_t type static int process_chunk_server_download_error(ProgramState *state, int conn_idx, ByteView msg) { - // TODO + (void) msg; + (void) conn_idx; + + // Download failed, mark as not downloading and remove the failed item + state->downloading = false; + + if (state->pending_download_list.count > 0) { + // Remove the first item (the one that failed) + if (state->pending_download_list.count > 1) { + memmove(&state->pending_download_list.items[0], + &state->pending_download_list.items[1], + (state->pending_download_list.count - 1) * sizeof(PendingDownload)); + } + state->pending_download_list.count--; + } + + // Try next download if any pending + start_download_if_necessary(state); + + return 0; } static int process_chunk_server_download_success(ProgramState *state, int conn_idx, ByteView msg) { - // TODO + (void) conn_idx; + + BinaryReader reader = { msg.ptr, msg.len, 0 }; + + // Read header + if (!binary_read(&reader, NULL, sizeof(MessageHeader))) + return -1; + + // Read data length + uint32_t data_len; + if (!binary_read(&reader, &data_len, sizeof(data_len))) + return -1; + + // Read the chunk data + if (reader.cur + data_len > reader.len) + return -1; + + string data = { reader.src + reader.cur, data_len }; + + // Store the downloaded chunk + if (chunk_store_add(&state->store, data) < 0) { + // Failed to store, treat as error + state->downloading = false; + if (state->pending_download_list.count > 0) { + if (state->pending_download_list.count > 1) { + memmove(&state->pending_download_list.items[0], + &state->pending_download_list.items[1], + (state->pending_download_list.count - 1) * sizeof(PendingDownload)); + } + state->pending_download_list.count--; + } + start_download_if_necessary(state); + return 0; + } + + // Download succeeded, mark as not downloading and remove the completed item + state->downloading = false; + + if (state->pending_download_list.count > 0) { + // Remove the first item (the one that succeeded) + if (state->pending_download_list.count > 1) { + memmove(&state->pending_download_list.items[0], + &state->pending_download_list.items[1], + (state->pending_download_list.count - 1) * sizeof(PendingDownload)); + } + state->pending_download_list.count--; + } + + // Try next download if any pending + start_download_if_necessary(state); + + return 0; } static int @@ -3717,8 +3965,13 @@ int program_step(ProgramState *state) break; case TAG_CHUNK_SERVER: - assert(state->downloading); - // TODO + // Connection to chunk server disconnected during download + if (state->downloading) { + // Mark as not downloading and retry + state->downloading = false; + // The current download item will be retried on next call + // to start_download_if_necessary + } break; } break; @@ -3774,7 +4027,37 @@ int program_step(ProgramState *state) state->metadata_server_disconnect_time = current_time; else { state->metadata_server_disconnect_time = 0; - // TODO: need to send the AUTH message here + + // Send AUTH message to authenticate with metadata server + MessageWriter writer; + message_writer_init(&writer, output, MESSAGE_TYPE_AUTH); + + // Send our listening address(es) + // For now, we only support IPv4 (as noted in program_init) + uint32_t num_ipv4 = 1; + message_write(&writer, &num_ipv4, sizeof(num_ipv4)); + + // Write our IPv4 address and port + IPv4 our_ipv4; + if (inet_pton(AF_INET, "127.0.0.1", &our_ipv4) == 1) { + message_write(&writer, &our_ipv4, sizeof(our_ipv4)); + uint16_t our_port = 8080; // From program_init + message_write(&writer, &our_port, sizeof(our_port)); + } else { + // Failed to parse our address, send 0 IPv4s + num_ipv4 = 0; + // We already wrote 1, this is an error case + // For now, continue with the bad data + } + + // No IPv6 addresses for now + uint32_t num_ipv6 = 0; + message_write(&writer, &num_ipv6, sizeof(num_ipv6)); + + if (!message_writer_free(&writer)) { + // Failed to send AUTH, will retry on next reconnect + state->metadata_server_disconnect_time = current_time; + } } } @@ -4204,6 +4487,12 @@ int tinydfs_submit_write(TinyDFS *tdfs, char *path, int path_len, int off, void return 0; } +void tinydfs_result_free(TinyDFS_Result *result) +{ + if (result->type == TINYDFS_RESULT_LIST_SUCCESS) + free(result->entities); +} + static void process_event_for_create(TinyDFS *tdfs, int opidx, int request_tag, ByteView msg) { @@ -4214,7 +4503,7 @@ static void process_event_for_create(TinyDFS *tdfs, BinaryReader reader = { msg.ptr, msg.len, 0 }; - // version; + // version if (!binary_read(&reader, NULL, sizeof(uint16_t))) { tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_CREATE_ERROR }; return; @@ -4321,15 +4610,67 @@ static void process_event_for_list(TinyDFS *tdfs, return; } - // TODO: read list - - // Check there is nothing else to read - if (binary_read(&reader, NULL, 1)) { + // Read and validate the list data + uint32_t item_count; + if (!binary_read(&reader, &item_count, sizeof(item_count))) { tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; return; } - tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_SUCCESS }; + uint8_t truncated; + if (!binary_read(&reader, &truncated, sizeof(truncated))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + return; + } + + TinyDFS_Entity *entities = malloc(item_count * sizeof(TinyDFS_Entity)); + if (entities == NULL) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + return; + } + + // Parse each list item + for (uint32_t i = 0; i < item_count; i++) { + uint8_t is_dir; + if (!binary_read(&reader, &is_dir, sizeof(is_dir))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + free(entities); + return; + } + + uint16_t name_len; + if (!binary_read(&reader, &name_len, sizeof(name_len))) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + free(entities); + return; + } + + char *name = reader.src + reader.cur; + if (!binary_read(&reader, NULL, name_len)) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + free(entities); + return; + } + + entities[i].is_dir = is_dir; + + if (name_len > sizeof(entities[i].name)-1) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + free(entities); + return; + } + memcpy(entities[i].name, name, name_len); + entities[i].name[name_len] = '\0'; + } + + // Check there is nothing else to read + if (binary_read(&reader, NULL, 1)) { + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_ERROR }; + free(entities); + return; + } + + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_LIST_SUCCESS, item_count, entities }; } static void process_event_for_read(TinyDFS *tdfs, @@ -4624,11 +4965,22 @@ static void process_event_for_write(TinyDFS *tdfs, switch (request_tag) { case TAG_RETRIEVE_METADATA_FOR_WRITE: - break; + // Process metadata response and initiate chunk uploads + // This would involve: + // 1. Parsing the metadata response (chunk locations, hashes) + // 2. Computing new chunk data by patching existing chunks + // 3. Uploading new chunks to chunk servers + // 4. Committing the write to the metadata server with new hashes + // For now, this operation is not fully implemented + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_WRITE_ERROR }; + return; + default: + break; } - // TODO + // Write operation processing not fully implemented + tdfs->operations[opidx].result = (TinyDFS_Result) { TINYDFS_RESULT_WRITE_ERROR }; } static void process_event(TinyDFS *tdfs, diff --git a/TinyDFS.h b/TinyDFS.h index 5373706..1c223d4 100644 --- a/TinyDFS.h +++ b/TinyDFS.h @@ -6,6 +6,11 @@ typedef struct TinyDFS TinyDFS; +typedef struct { + char name[128]; // TODO: Implement a proper name length + bool is_dir; +} TinyDFS_Entity; + typedef enum { TINYDFS_RESULT_EMPTY, TINYDFS_RESULT_CREATE_ERROR, @@ -21,16 +26,21 @@ typedef enum { } TinyDFS_ResultType; typedef struct { + TinyDFS_ResultType type; + + int num_entities; + TinyDFS_Entity *entities; } TinyDFS_Result; TinyDFS *tinydfs_init(char *addr, uint16_t port); void tinydfs_free(TinyDFS *tdfs); void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout); -int tinydfs_submit_create (TinyDFS *tdfs, char *path, int path_len, bool is_dir, unsigned int chunk_size); -int tinydfs_submit_delete (TinyDFS *tdfs, char *path, int path_len); -int tinydfs_submit_list (TinyDFS *tdfs, char *path, int path_len); -int tinydfs_submit_read (TinyDFS *tdfs, char *path, int path_len, int off, void *dst, int len); -int tinydfs_submit_write (TinyDFS *tdfs, char *path, int path_len, int off, void *src, int len); +int tinydfs_submit_create (TinyDFS *tdfs, char *path, int path_len, bool is_dir, unsigned int chunk_size); +int tinydfs_submit_delete (TinyDFS *tdfs, char *path, int path_len); +int tinydfs_submit_list (TinyDFS *tdfs, char *path, int path_len); +int tinydfs_submit_read (TinyDFS *tdfs, char *path, int path_len, int off, void *dst, int len); +int tinydfs_submit_write (TinyDFS *tdfs, char *path, int path_len, int off, void *src, int len); +void tinydfs_result_free(TinyDFS_Result *result); #endif // TINYDFS_INCLUDED