From db2ef299bf011e5436e1d6a9c2c1aed30d76f432 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 25 Feb 2026 13:12:17 +0100 Subject: [PATCH] Fix chunk replication --- src/client.c | 64 ++++++++++++++++++++++++++++++++++++++++++-------- src/metadata.h | 3 +++ 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/src/client.c b/src/client.c index a85921c..50d7943 100644 --- a/src/client.c +++ b/src/client.c @@ -67,9 +67,11 @@ struct ToastyFS { Transfer transfers[MAX_TRANSFERS]; int num_transfers; - SHA256 chunks[META_CHUNKS_MAX]; - int chunk_sizes[META_CHUNKS_MAX]; - int num_chunks; + SHA256 chunks[META_CHUNKS_MAX]; + int chunk_sizes[META_CHUNKS_MAX]; + uint16_t chunk_num_holders[META_CHUNKS_MAX]; + uint16_t chunk_holders[META_CHUNKS_MAX][REPLICATION_FACTOR]; + int num_chunks; char *put_data; int put_data_len; @@ -166,6 +168,7 @@ void toastyfs_free(ToastyFS *tfs) static int find_completed_transfer_for_hash(ToastyFS *tfs, SHA256 hash); static bool all_chunk_transfers_completed(ToastyFS *tfs); +static bool all_transfers_finished(ToastyFS *tfs); static bool transfer_for_hash_already_started(ToastyFS *tfs, SHA256 hash) @@ -304,6 +307,9 @@ static void replay_request(ToastyFS *tfs) for (int i = 0; i < tfs->num_chunks; i++) { msg.oper.chunks[i].hash = tfs->chunks[i]; msg.oper.chunks[i].size = tfs->chunk_sizes[i]; + msg.oper.chunks[i].num_holders = tfs->chunk_num_holders[i]; + for (int j = 0; j < tfs->chunk_num_holders[i]; j++) + msg.oper.chunks[i].holders[j] = tfs->chunk_holders[i][j]; } send_message(&tfs->msys, leader_idx(tfs), &msg.base); } @@ -348,6 +354,16 @@ static void replay_request(ToastyFS *tfs) static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) { + // Discard messages that arrive after the operation has completed or + // before a new one has started. This handles late/stale responses + // from previous operations (e.g. a STORE_CHUNK_ACK arriving after the + // client already moved on to a GET). + if (tfs->step == STEP_IDLE + || tfs->step == STEP_PUT_DONE + || tfs->step == STEP_GET_DONE + || tfs->step == STEP_DELETE_DONE) + return 0; + switch (tfs->step) { case STEP_FETCH_CHUNK: { @@ -412,10 +428,27 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) assert(transfer_idx > -1); tfs->transfers[transfer_idx].state = TRANSFER_COMPLETED; - mark_waiting_transfers_for_hash_as_aborted(tfs, ack.hash); + // NOTE: Do NOT abort pending transfers for this hash. + // We need all REPLICATION_FACTOR copies to be stored. begin_transfers(tfs); - if (all_chunk_transfers_completed(tfs)) { + if (all_transfers_finished(tfs)) { + + // Extract holder info from completed transfers + for (int i = 0; i < tfs->num_chunks; i++) { + tfs->chunk_num_holders[i] = 0; + for (int t = 0; t < tfs->num_transfers; t++) { + if (!memcmp(&tfs->transfers[t].hash, &tfs->chunks[i], sizeof(SHA256)) + && tfs->transfers[t].state == TRANSFER_COMPLETED) + { + int h = tfs->chunk_num_holders[i]; + if (h < REPLICATION_FACTOR) { + tfs->chunk_holders[i][h] = (uint16_t)tfs->transfers[t].location; + tfs->chunk_num_holders[i]++; + } + } + } + } tfs->request_id++; CommitPutMessage msg = { @@ -438,6 +471,9 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) for (int i = 0; i < tfs->num_chunks; i++) { msg.oper.chunks[i].hash = tfs->chunks[i]; msg.oper.chunks[i].size = tfs->chunk_sizes[i]; + msg.oper.chunks[i].num_holders = tfs->chunk_num_holders[i]; + for (int j = 0; j < tfs->chunk_num_holders[i]; j++) + msg.oper.chunks[i].holders[j] = tfs->chunk_holders[i][j]; } send_message(&tfs->msys, leader_idx(tfs), &msg.base); tfs->step = STEP_COMMIT; @@ -596,12 +632,9 @@ static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg) tfs->num_transfers = 0; for (int i = 0; i < (int)resp.num_chunks; i++) { - // TODO: The server selection formula is a temporary - // solution. Figure out a proper strategy for - // picking which servers to fetch chunks from. - for (int j = 0; j < REPLICATION_FACTOR; j++) { + for (int j = 0; j < resp.chunks[i].num_holders; j++) { add_transfer(tfs, resp.chunks[i].hash, - (i + j) % tfs->num_servers, NULL, 0); + resp.chunks[i].holders[j], NULL, 0); } tfs->chunks[i] = resp.chunks[i].hash; tfs->chunk_sizes[i] = resp.chunks[i].size; @@ -925,6 +958,17 @@ static void get_result(ToastyFS *tfs, ToastyFS_Result *result) result->size = blob_size; } +static bool +all_transfers_finished(ToastyFS *tfs) +{ + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_PENDING + || tfs->transfers[i].state == TRANSFER_STARTED) + return false; + } + return true; +} + static bool all_chunk_transfers_completed(ToastyFS *tfs) { diff --git a/src/metadata.h b/src/metadata.h index 4f37b99..fdca2e2 100644 --- a/src/metadata.h +++ b/src/metadata.h @@ -5,10 +5,13 @@ #include #include +#include "config.h" typedef struct { SHA256 hash; uint32_t size; + uint16_t num_holders; + uint16_t holders[REPLICATION_FACTOR]; } ChunkRef; #define META_BUCKET_MAX 64