From 9a944feb01f097f768c9c1dbe6f86e027166baa8 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Fri, 14 Nov 2025 01:16:48 +0100 Subject: [PATCH] Rewrote simulation scheduler and fixed some bugs --- Makefile | 6 +- src/chunk_server.c | 15 ++- src/metadata_server.c | 10 +- src/system.c | 221 ++++++++++++++++++++++++++---------------- src/tcp.c | 8 +- src/tcp.h | 1 + 6 files changed, 164 insertions(+), 97 deletions(-) diff --git a/Makefile b/Makefile index 5e4b4b0..ee34697 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ -CFLAGS = -Wall -Wextra -ggdb +CFLAGS = -Wall -Wextra -ggdb -fsanitize=address,undefined COVERAGE_CFLAGS = $(CFLAGS) --coverage COVERAGE_LFLAGS = --coverage @@ -22,10 +22,10 @@ all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a coverage: mousefs_random_test_coverage$(EXT) coverage-report: - @./scripts/measure_coverage.sh 5 + @./scripts/measure_coverage.sh 60 coverage-html: - @./scripts/measure_coverage.sh 5 --html + @./scripts/measure_coverage.sh 60 --html mousefs$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER diff --git a/src/chunk_server.c b/src/chunk_server.c index 5ea2fe0..ecca8c7 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -871,6 +871,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts start_connecting_to_metadata_server(state); + // TODO: add all chunk hashes to the add list + printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n", addr.len, addr.ptr, @@ -911,14 +913,16 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled switch (events[i].type) { case EVENT_CONNECT: - if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) - state->disconnect_time = INVALID_TIME; + if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) { + assert(state->disconnect_time == INVALID_TIME); + } break; case EVENT_DISCONNECT: - switch (tcp_get_tag(&state->tcp, conn_idx)) { + switch (events[i].tag) { case TAG_METADATA_SERVER: + assert(state->disconnect_time == INVALID_TIME); state->disconnect_time = current_time; break; @@ -977,6 +981,11 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } + { + int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); + assert((conn_idx < 0) == (state->disconnect_time != INVALID_TIME)); + } + Time deadline = INVALID_TIME; // Remove items from the remove list that got too old diff --git a/src/metadata_server.c b/src/metadata_server.c index 7ee59fe..4338b7f 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -870,6 +870,8 @@ static int process_chunk_server_sync_3(MetadataServer *state, HashSet tmp_list; hash_set_init(&tmp_list); + message_write(&writer, &count, sizeof(count)); + for (uint32_t i = 0; i < count; i++) { SHA256 hash; @@ -933,6 +935,7 @@ static bool is_chunk_server_message_type(uint16_t type) switch (type) { case MESSAGE_TYPE_AUTH: case MESSAGE_TYPE_SYNC: + case MESSAGE_TYPE_SYNC_3: return true; default: @@ -1010,9 +1013,9 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * case EVENT_DISCONNECT: { - int tag = tcp_get_tag(&state->tcp, conn_idx); - if (tag >= 0) { - chunk_server_peer_free(&state->chunk_servers[tag]); + if (events[i].tag >= 0) { + chunk_server_peer_free(&state->chunk_servers[events[i].tag]); + assert(state->num_chunk_servers > 0); state->num_chunk_servers--; } } @@ -1048,6 +1051,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * } chunk_server_peer_init(&state->chunk_servers[j], current_time); + state->num_chunk_servers++; tcp_set_tag(&state->tcp, conn_idx, j, true); diff --git a/src/system.c b/src/system.c index 94001ac..9ae26c2 100644 --- a/src/system.c +++ b/src/system.c @@ -571,17 +571,6 @@ static bool data_queue_full(DataQueue *queue) return queue->used == queue->size; } -static int compare_processes(const void *p1, const void *p2) -{ - Process *a = *(Process**) p1; - Process *b = *(Process**) p2; - if (b->wakeup_time == INVALID_TIME) return -1; - if (a->wakeup_time == INVALID_TIME) return +1; - if (a->wakeup_time < b->wakeup_time) return -1; - if (a->wakeup_time > b->wakeup_time) return +1; - return 0; -} - static int setup_poll_array(void **contexts, struct pollfd *polled) { int num_polled = 0; @@ -653,80 +642,8 @@ static int setup_poll_array(void **contexts, struct pollfd *polled) return num_polled; } -void update_simulation(void) +static void update_network(void) { - // Order processes by wakeup time. Those with no - // wakeup time go last. - Process *ordered_processes[MAX_PROCESSES]; - for (int i = 0; i < num_processes; i++) - ordered_processes[i] = processes[i]; - - qsort(ordered_processes, num_processes, sizeof(*ordered_processes), compare_processes); - - for (int i = 0; i < num_processes; i++) { - - current_process = ordered_processes[i]; - - void *contexts[MAX_CONNS+1]; - struct pollfd polled[MAX_CONNS+1]; - int num_polled = setup_poll_array(contexts, polled); - - if (num_polled == 0) { - - Time wakeup_time = current_process->wakeup_time; - if (wakeup_time == INVALID_TIME) - continue; - - assert(current_time <= wakeup_time); - current_time = wakeup_time; - } - - int timeout = -1; - switch (current_process->type) { - case PROCESS_TYPE_METADATA_SERVER: - num_polled = metadata_server_step( - ¤t_process->metadata_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CHUNK_SERVER: - num_polled = chunk_server_step( - ¤t_process->chunk_server, - contexts, - polled, - num_polled, - &timeout - ); - break; - case PROCESS_TYPE_CLIENT: - num_polled = simulation_client_step( - ¤t_process->simulation_client, - contexts, - polled, - num_polled, - &timeout - ); - break; - } - - if (num_polled < 0) { - assert(0); // TODO - } - - if (timeout < 0) { - current_process->wakeup_time = INVALID_TIME; - } else { - current_process->wakeup_time = current_time + (Time) timeout * 1000000; - } - - process_poll_array(current_process, contexts, polled, num_polled); - - current_process = NULL; - } - for (int i = 0; i < num_processes; i++) { for (int j = 0, k = 0; k < processes[i]->num_desc; j++) { @@ -786,6 +703,142 @@ void update_simulation(void) } } +void update_simulation(void) +{ + for (;;) { + int num_ready = 0; + for (int i = 0; i < num_processes; i++) { + + current_process = processes[i]; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + if (num_polled > 0) { + + num_ready++; + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + } + + current_process = NULL; + + update_network(); + } + + if (num_ready == 0) + break; + } + + Process *next_process = NULL; + for (int i = 0; i < num_processes; i++) + if (processes[i]->wakeup_time != INVALID_TIME) + if (next_process == NULL || processes[i]->wakeup_time < next_process->wakeup_time) + next_process = processes[i]; + + if (next_process == NULL) { + assert(0); // Nothing to schedule next + } + + assert(current_time <= next_process->wakeup_time); + current_time = next_process->wakeup_time; + + current_process = next_process; + + void *contexts[MAX_CONNS+1]; + struct pollfd polled[MAX_CONNS+1]; + int num_polled = setup_poll_array(contexts, polled); + + int timeout = -1; + switch (current_process->type) { + case PROCESS_TYPE_METADATA_SERVER: + num_polled = metadata_server_step( + ¤t_process->metadata_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CHUNK_SERVER: + num_polled = chunk_server_step( + ¤t_process->chunk_server, + contexts, + polled, + num_polled, + &timeout + ); + break; + case PROCESS_TYPE_CLIENT: + num_polled = simulation_client_step( + ¤t_process->simulation_client, + contexts, + polled, + num_polled, + &timeout + ); + break; + } + + if (num_polled < 0) { + assert(0); // TODO + } + + if (timeout < 0) { + current_process->wakeup_time = INVALID_TIME; + } else { + current_process->wakeup_time = current_time + (Time) timeout * 1000000; + } + + process_poll_array(current_process, contexts, polled, num_polled); + + current_process = NULL; + + update_network(); +} + void *mock_malloc(size_t len) { return malloc(len); diff --git a/src/tcp.c b/src/tcp.c index 1affff1..36f16eb 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -240,8 +240,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd if (set_socket_blocking(new_fd, false) < 0) CLOSE_SOCKET(new_fd); else { - events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns }; conn_init(&tcp->conns[tcp->num_conns++], new_fd, false); + events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns-1, tcp->conns[tcp->num_conns-1].tag }; } } } @@ -266,7 +266,7 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd defer_close = true; else { conn->connecting = false; - events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns }; + events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns, conn->tag }; } } @@ -319,8 +319,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd removed[i] = defer_close; if (0) {} - else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns }; - else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns }; + else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns, conn->tag }; + else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns, conn->tag }; } } diff --git a/src/tcp.h b/src/tcp.h index 015bc7d..2a41a19 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -23,6 +23,7 @@ typedef enum { typedef struct { EventType type; int conn_idx; + int tag; } Event; typedef struct {