From 784660765181780a1c89ffb9466f348899713384 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Sat, 8 Nov 2025 11:51:24 +0100 Subject: [PATCH] Progress --- src/client.c | 62 ++++++++++++++++++++++++------------- src/config.h | 2 ++ src/metadata_server.c | 72 ++++++++++++++++++++++++++++--------------- 3 files changed, 90 insertions(+), 46 deletions(-) diff --git a/src/client.c b/src/client.c index 6ecb4cd..7d3ddf8 100644 --- a/src/client.c +++ b/src/client.c @@ -27,7 +27,6 @@ #define TAG_UPLOAD_CHUNK_MAX 2000 #define PARALLEL_LIMIT 5 -#define REPLICATION_FACTOR 3 typedef struct { SHA256 hash; @@ -1535,23 +1534,23 @@ static void process_event_for_write(TinyDFS *tdfs, } SHA256 hash; - uint16_t expected_type; - if (tdfs->operations[opidx].uploads[found].chunk_index >= tdfs->operations[opidx].num_hashes) - expected_type = MESSAGE_TYPE_CREATE_CHUNK_SUCCESS; - else { - expected_type = MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS; - - if (!binary_read(&reader, &hash, sizeof(hash))) { - // TODO - return; - } + if (!binary_read(&reader, &hash, sizeof(hash))) { + // TODO + return; } + // Check that there is nothing else to read if (binary_read(&reader, NULL, 1)) { // TODO return; } + uint16_t expected_type; + if (tdfs->operations[opidx].uploads[found].chunk_index >= tdfs->operations[opidx].num_hashes) + expected_type = MESSAGE_TYPE_CREATE_CHUNK_SUCCESS; + else + expected_type = MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS; + if (type != expected_type) { tdfs->operations[opidx].uploads[found].status = UPLOAD_FAILED; } else { @@ -1589,9 +1588,10 @@ static void process_event_for_write(TinyDFS *tdfs, // uploaded to at least N different servers typedef struct { - SHA256 old_hash; - SHA256 new_hash; - int replication; + SHA256 old_hash; + SHA256 new_hash; + int num_locations; + Address locations[REPLICATION_FACTOR]; } ChunkUploadResult; int num_upload_results = tdfs->operations[opidx].num_chunks; @@ -1602,21 +1602,24 @@ static void process_event_for_write(TinyDFS *tdfs, for (int i = 0; i < num_upload_results; i++) { upload_results[i].old_hash = tdfs->operations[opidx].hashes[i]; - upload_results[i].replication = 0; + upload_results[i].num_locations = 0; } - for (int i = 0; i < tdfs->operations[opidx].num_uploads; i++) - if (tdfs->operations[opidx].uploads[i].status == UPLOAD_COMPLETED) { - upload_results[tdfs->operations[opidx].uploads[i].chunk_index].new_hash = tdfs->operations[opidx].uploads[i].final_hash; - upload_results[tdfs->operations[opidx].uploads[i].chunk_index].replication++; + for (int i = 0; i < tdfs->operations[opidx].num_uploads; i++) { + UploadSchedule *u = &tdfs->operations[opidx].uploads[i]; + if (u->status == UPLOAD_COMPLETED) { + int n = upload_results[u->chunk_index].num_locations++; + upload_results[u->chunk_index].locations[n] = u->address; + upload_results[u->chunk_index].new_hash = u->final_hash; } + } // Now check that each chunk is replicated // at least N times - bool ok = false; + bool ok = true; for (int i = 0; i < num_upload_results; i++) { - if (upload_results[i].replication < REPLICATION_FACTOR) { + if (upload_results[i].num_locations < REPLICATION_FACTOR) { ok = false; break; } @@ -1649,9 +1652,24 @@ static void process_event_for_write(TinyDFS *tdfs, message_write(&writer, &num_chunks, sizeof(num_chunks)); for (int i = 0; i < num_upload_results; i++) { + + // TODO: newly create chunks don't have an old hash message_write(&writer, &upload_results[i].old_hash, sizeof(upload_results[i].old_hash)); message_write(&writer, &upload_results[i].new_hash, sizeof(upload_results[i].new_hash)); - // TODO + + uint32_t tmp = upload_results[i].num_locations; + message_write(&writer, &tmp, sizeof(tmp)); + + for (int j = 0; j < upload_results[i].num_locations; j++) { + + Address addr = upload_results[i].locations[j]; + + uint8_t is_ipv4 = addr.is_ipv4; + message_write(&writer, &is_ipv4, sizeof(is_ipv4)); + if (addr.is_ipv4) message_write(&writer, &addr.ipv4, sizeof(addr.ipv4)); + else message_write(&writer, &addr.ipv6, sizeof(addr.ipv6)); + message_write(&writer, &addr.port, sizeof(addr.port)); + } } free(upload_results); diff --git a/src/config.h b/src/config.h index da9fa93..7aeb7a1 100644 --- a/src/config.h +++ b/src/config.h @@ -6,4 +6,6 @@ #define MAX_OPERATIONS 128 #define MAX_REQUESTS_PER_QUEUE 128 +#define REPLICATION_FACTOR 3 + #endif // CONFIG_INCLUDED diff --git a/src/metadata_server.c b/src/metadata_server.c index d47dda4..f9b0303 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -558,9 +558,14 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) #define MAX_CHUNKS_PER_WRITE 32 - Address addrs[MAX_CHUNKS_PER_WRITE]; - SHA256 new_hashes[MAX_CHUNKS_PER_WRITE]; - SHA256 old_hashes[MAX_CHUNKS_PER_WRITE]; + typedef struct { + SHA256 old_hash; + SHA256 new_hash; + int num_addrs; + Address addrs[REPLICATION_FACTOR]; + } ChunkWriteResult; + + ChunkWriteResult results[MAX_CHUNKS_PER_WRITE]; for (uint32_t i = 0; i < num_chunks; i++) { @@ -572,27 +577,36 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, &new_hash, sizeof(new_hash))) return -1; - uint8_t is_ipv4; - if (!binary_read(&reader, &is_ipv4, sizeof(is_ipv4))) + results[i].old_hash = old_hash; + results[i].new_hash = new_hash; + + uint32_t num_locations; + if (!binary_read(&reader, &num_locations, sizeof(num_locations))) return -1; - Address addr; - addr.is_ipv4 = is_ipv4; + for (uint32_t j = 0; j < num_locations; j++) { - if (is_ipv4) { - if (!binary_read(&reader, &addr.ipv4, sizeof(addr.ipv4))) + uint8_t is_ipv4; + if (!binary_read(&reader, &is_ipv4, sizeof(is_ipv4))) return -1; - } else { - if (!binary_read(&reader, &addr.ipv6, sizeof(addr.ipv6))) + + Address addr; + addr.is_ipv4 = is_ipv4; + + if (is_ipv4) { + if (!binary_read(&reader, &addr.ipv4, sizeof(addr.ipv4))) + return -1; + } else { + if (!binary_read(&reader, &addr.ipv6, sizeof(addr.ipv6))) + return -1; + } + + if (!binary_read(&reader, &addr.port, sizeof(addr.port))) return -1; + + if (results[i].num_addrs < REPLICATION_FACTOR) + results[i].addrs[results[i].num_addrs++] = addr; } - - if (!binary_read(&reader, &addr.port, sizeof(addr.port))) - return -1; - - addrs[i] = addr; - new_hashes[i] = new_hash; - old_hashes[i] = old_hash; } // Check that there are no more bytes to read @@ -603,8 +617,15 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) SHA256 removed_hashes[MAX_CHUNKS_PER_WRITE]; int num_removed = 0; + SHA256 old_hashes[MAX_CHUNKS_PER_WRITE]; + SHA256 new_hashes[MAX_CHUNKS_PER_WRITE]; + for (int i = 0; i < num_chunks; i++) { + old_hashes[i] = results[i].old_hash; + new_hashes[i] = results[i].new_hash; + } + int ret = file_tree_write(&state->file_tree, path, offset, length, - old_hashes, new_hashes, removed_hashes, &num_removed); + old_hashes, new_hashes, removed_hashes, &num_removed); if (ret < 0) { @@ -626,12 +647,15 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) // 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; - if (!hash_list_insert(&state->chunk_servers[j].add_list, new_hashes[i])) - return -1; + for (int j = 0; j < results[i].num_addrs; j++) { + + 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) + return -1; + } } // Mark removed chunks for deletion on all chunk servers that have them