From d7cf946431211450432d142194db3f5d4dd1ecfe Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Fri, 7 Nov 2025 20:26:15 +0100 Subject: [PATCH] Progress --- src/chunk_server.c | 31 ++++++----- src/client.c | 120 ++++++++++++++++++++++++++++++++++++++---- src/file_tree.c | 15 ++++-- src/main_test.c | 1 - src/metadata_server.c | 17 +++--- src/system.c | 2 - 6 files changed, 144 insertions(+), 42 deletions(-) diff --git a/src/chunk_server.c b/src/chunk_server.c index 958cac7..641cef8 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -34,8 +34,10 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash) if (list->count == list->capacity) { int new_capacity; - if (list->capacity == 0) new_capacity = 8; - else new_capacity = 2 * list->capacity; + if (list->capacity == 0) + new_capacity = 8; + else + new_capacity = 2 * list->capacity; PendingDownload *new_items = sys_malloc(new_capacity * sizeof(PendingDownload)); if (new_items == NULL) @@ -276,6 +278,7 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView // 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]; @@ -679,6 +682,7 @@ process_client_upload_chunk(ChunkServer *state, int conn_idx, ByteView msg) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_UPLOAD_CHUNK_ERROR, S("Invalid message")); string data = { (char*) reader.src + reader.cur, data_len }; + // TODO: Shoudn't we skip the data here? binary_read(&reader, NULL, data_len) // Check that there are no more bytes to read if (binary_read(&reader, NULL, 1)) @@ -757,10 +761,17 @@ static int process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) { switch (type) { - case MESSAGE_TYPE_CREATE_CHUNK: return process_client_create_chunk(state, conn_idx, msg); - case MESSAGE_TYPE_UPLOAD_CHUNK: return process_client_upload_chunk(state, conn_idx, msg); - case MESSAGE_TYPE_DOWNLOAD_CHUNK: return process_client_download_chunk(state, conn_idx, msg); - default:break; + case MESSAGE_TYPE_CREATE_CHUNK: + return process_client_create_chunk(state, conn_idx, msg); + + case MESSAGE_TYPE_UPLOAD_CHUNK: + return process_client_upload_chunk(state, conn_idx, msg); + + case MESSAGE_TYPE_DOWNLOAD_CHUNK: + return process_client_download_chunk(state, conn_idx, msg); + + default: + break; } return -1; } @@ -768,8 +779,6 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView static void start_connecting_to_metadata_server(ChunkServer *state) { - printf("Chunk server is connecting to the metadata server\n"); - Time current_time = get_current_time(); ByteQueue *output; @@ -900,13 +909,11 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled switch (events[i].type) { case EVENT_CONNECT: - printf("New connection to chunk server\n"); if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) - state->disconnect_time = 0; + state->disconnect_time = INVALID_TIME; break; case EVENT_DISCONNECT: - printf("Dropped connection to chunk server\n"); switch (tcp_get_tag(&state->tcp, conn_idx)) { case TAG_METADATA_SERVER: @@ -939,8 +946,6 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled break; } - printf("Processing message to chunk server\n"); - switch (tcp_get_tag(&state->tcp, conn_idx)) { case TAG_METADATA_SERVER: ret = process_metadata_server_message(state, conn_idx, msg_type, msg); diff --git a/src/client.c b/src/client.c index bdebee0..e9fc5c8 100644 --- a/src/client.c +++ b/src/client.c @@ -116,6 +116,7 @@ typedef struct { uint32_t chunk_size; UploadSchedule *uploads; int num_uploads; + int cap_uploads; TinyDFS_Result result; } Operation; @@ -1107,6 +1108,37 @@ static int count_pending_uploads(TinyDFS *tdfs, int opidx) return n; } +static int schedule_upload(TinyDFS *tdfs, int opidx, UploadSchedule upload) +{ + if (tdfs->operations[opidx].num_uploads == tdfs->operations[opidx].cap_uploads) { + + int new_cap_uploads; + if (tdfs->operations[opidx].uploads == NULL) + new_cap_uploads = 0; + else + new_cap_uploads = 2 * tdfs->operations[opidx].cap_uploads; + + UploadSchedule *uploads = sys_malloc(new_cap_uploads * sizeof(UploadSchedule)); + if (uploads == NULL) + return -1; + + if (tdfs->operations[opidx].num_uploads > 0) { + memcpy( + uploads, + tdfs->operations[opidx].uploads, + tdfs->operations[opidx].num_uploads * sizeof(UploadSchedule) + ); + free(tdfs->operations[opidx].uploads); + } + + tdfs->operations[opidx].uploads = uploads; + tdfs->operations[opidx].cap_uploads = new_cap_uploads; + } + + tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload; + return 0; +} + static void process_event_for_write(TinyDFS *tdfs, int opidx, int request_tag, ByteView msg) { @@ -1153,18 +1185,18 @@ static void process_event_for_write(TinyDFS *tdfs, return; } + uint32_t num_all_hasehs = (tdfs->operations[opidx].len + tdfs->operations[opidx].chunk_size - 1) / tdfs->operations[opidx].chunk_size; + uint32_t num_new_hashes = num_all_hasehs - num_hashes; + tdfs->operations[opidx].num_hashes = num_hashes; // TODO: overflow tdfs->operations[opidx].hashes = sys_malloc(num_hashes * sizeof(SHA256)); if (tdfs->operations[opidx].hashes == NULL) { // TODO } - // TODO: !!! IMPORTANT !!! This should also account for new chunks, not patched. It does not do so at the moment - // TODO: This may overestimate by a lot the actual memory required by the array - tdfs->operations[opidx].uploads = sys_malloc(num_hashes * MAX_CHUNK_SERVERS * MAX_SERVER_ADDRS * sizeof(UploadSchedule)); - if (tdfs->operations[opidx].uploads == NULL) { - // TODO - } + tdfs->operations[opidx].uploads = NULL; + tdfs->operations[opidx].num_uploads = 0; + tdfs->operations[opidx].cap_uploads = 0; char *full_ptr = tdfs->operations[opidx].ptr; int full_off = tdfs->operations[opidx].off; @@ -1235,7 +1267,9 @@ static void process_event_for_write(TinyDFS *tdfs, upload.src = src; upload.off = off; upload.len = len; - tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload; + if (schedule_upload(tdfs, opidx, upload) < 0) { + // TODO + } } uint32_t num_ipv6; @@ -1268,8 +1302,9 @@ static void process_event_for_write(TinyDFS *tdfs, upload.src = src; upload.off = off; upload.len = len; - tdfs->operations[opidx].uploads[tdfs->operations[opidx].num_uploads++] = upload; - } + if (schedule_upload(tdfs, opidx, upload) < 0) { + // TODO + } } } } @@ -1281,6 +1316,9 @@ static void process_event_for_write(TinyDFS *tdfs, for (uint32_t i = 0; i < num_locations; i++) { + int server_lid = next_server_lid; + next_server_lid++; + uint32_t num_ipv4; if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; @@ -1301,7 +1339,37 @@ static void process_event_for_write(TinyDFS *tdfs, return; } - // TODO + int old_relative_off = relative_off; + + for (int w = 0; w < num_new_hashes; w++) { + + char *src = full_ptr + relative_off; + + int off = 0; + if (num_hashes == 0 && w == 0) + off = full_off % chunk_size; + + int len = full_len - relative_off; + if (len > chunk_size) + len = chunk_size; + + relative_off += len; + + UploadSchedule upload; + upload.status = UPLOAD_WAITING; + upload.server_lid = server_lid; + upload.address.is_ipv4 = true; + upload.address.ipv4 = ipv4; + upload.address.port = port; + upload.hash_lid = -1; + upload.src = src; + upload.off = off; + upload.len = len; + if (schedule_upload(tdfs, opidx, upload) < 0) { + // TODO + } } + + relative_off = old_relative_off; } uint32_t num_ipv6; @@ -1312,6 +1380,8 @@ static void process_event_for_write(TinyDFS *tdfs, for (uint32_t k = 0; k < num_ipv6; k++) { + char *src = full_ptr + relative_off; + IPv6 ipv6; if (!binary_read(&reader, &ipv6, sizeof(ipv6))) { tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_WRITE_ERROR }; @@ -1324,7 +1394,35 @@ static void process_event_for_write(TinyDFS *tdfs, return; } - // TODO + int old_relative_off = relative_off; + + for (int w = 0; w < num_new_hashes; w++) { + + int off = 0; + if (num_hashes == 0 && w == 0) + off = full_off % chunk_size; + + int len = full_len - relative_off; + if (len > chunk_size) + len = chunk_size; + + relative_off += len; + + UploadSchedule upload; + upload.status = UPLOAD_WAITING; + upload.server_lid = server_lid; + upload.address.is_ipv4 = false; + upload.address.ipv6 = ipv6; + upload.address.port = port; + upload.hash_lid = -1; + upload.src = src; + upload.off = off; + upload.len = len; + if (schedule_upload(tdfs, opidx, upload) < 0) { + // TODO + } } + + relative_off = old_relative_off; } } diff --git a/src/file_tree.c b/src/file_tree.c index 5d6e16a..066b1a2 100644 --- a/src/file_tree.c +++ b/src/file_tree.c @@ -423,14 +423,19 @@ int file_tree_read(FileTree *ft, string path, uint64_t first_chunk_index = off / f->chunk_size; uint64_t last_chunk_index = (off + len - 1) / f->chunk_size; + if (first_chunk_index >= f->num_chunks) + return 0; + + if (last_chunk_index >= f->num_chunks) { + if (f->num_chunks == 0) + return 0; + last_chunk_index = f->num_chunks-1; + } + int num_hashes = 0; for (uint32_t i = first_chunk_index; i <= last_chunk_index; i++) { - SHA256 hash; - if (i >= f->num_chunks) - hash = ZERO_HASH; - else - hash = f->chunks[i]; + SHA256 hash = f->chunks[i]; if (num_hashes < max_hashes) hashes[num_hashes] = hash; diff --git a/src/main_test.c b/src/main_test.c index c0a8bda..6d5ef40 100644 --- a/src/main_test.c +++ b/src/main_test.c @@ -35,7 +35,6 @@ int main(int argc, char **argv) // Spawn simulation client spawn_simulated_process("--client --remote-addr 127.0.0.1 --remote-port 8080"); - printf("Running simulation (press Ctrl+C to stop)...\n"); while (!simulation_should_stop) update_simulation(); diff --git a/src/metadata_server.c b/src/metadata_server.c index 720fa7e..887ab33 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -28,7 +28,11 @@ static int hash_list_insert(HashList *hash_list, SHA256 hash) if (hash_list->count == hash_list->capacity) { - int new_capacity = hash_list->capacity ? hash_list->capacity * 2 : 16; + 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) @@ -448,9 +452,9 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg) } else { - MessageWriter writer; - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + + MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_READ_SUCCESS); if (chunk_size > UINT32_MAX) { @@ -659,8 +663,6 @@ chunk_server_from_conn(MetadataServer *state, int conn_idx) static int process_chunk_server_auth(MetadataServer *state, int conn_idx, ByteView msg) { - printf("Received auth message from chunk server\n"); // TODO - ChunkServerPeer *chunk_server = chunk_server_from_conn(state, conn_idx); chunk_server->num_addrs = 0; @@ -810,13 +812,11 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * switch (events[i].type) { case EVENT_CONNECT: - printf("New connection to metadata server\n"); tcp_set_tag(&state->tcp, conn_idx, CONNECTION_TAG_UNKNOWN); break; case EVENT_DISCONNECT: { - printf("Dropped connection to metadata server\n"); int tag = tcp_get_tag(&state->tcp, conn_idx); if (tag >= 0) { chunk_server_peer_free(&state->chunk_servers[tag]); @@ -839,11 +839,8 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * break; } - printf("Processing message to metadata server\n"); - if (tcp_get_tag(&state->tcp, conn_idx) == CONNECTION_TAG_UNKNOWN) { if (is_chunk_server_message_type(msg_type)) { - printf("Metadata server determined peer is a chunk server\n"); // TODO int chunk_server_idx = state->num_chunk_servers++; chunk_server_peer_init(&state->chunk_servers[chunk_server_idx]); tcp_set_tag(&state->tcp, conn_idx, chunk_server_idx); diff --git a/src/system.c b/src/system.c index 8556115..23e9aed 100644 --- a/src/system.c +++ b/src/system.c @@ -636,8 +636,6 @@ void update_simulation(void) assert(current_time <= wakeup_time); current_time = wakeup_time; - - //printf("T=%2.2f ms\n", (float) current_time / 1000000); } int timeout = -1;