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;