From 6a608051d60c5d407b66187c2c0c3065e5fd3f5c Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Mon, 10 Nov 2025 15:07:38 +0100 Subject: [PATCH] Clean up timeout code in metadata server --- src/basic.c | 13 ++++ src/basic.h | 3 + src/config.h | 6 +- src/metadata_server.c | 136 ++++++++++++++++++------------------------ src/metadata_server.h | 4 +- src/tcp.c | 2 + 6 files changed, 81 insertions(+), 83 deletions(-) diff --git a/src/basic.c b/src/basic.c index 1bb5c14..d6a72f0 100644 --- a/src/basic.c +++ b/src/basic.c @@ -57,6 +57,19 @@ Time get_current_time(void) #endif } +void nearest_deadline(Time *a, Time b) +{ + if (*a == INVALID_TIME || *a > b) + *a = b; +} + +int deadline_to_timeout(Time deadline, Time current_time) +{ + if (deadline == INVALID_TIME) + return -1; + return (deadline - current_time) / 1000000; +} + bool getargb(int argc, char **argv, char *name) { for (int i = 0; i < argc; i++) diff --git a/src/basic.h b/src/basic.h index 083212f..f17b7c7 100644 --- a/src/basic.h +++ b/src/basic.h @@ -31,7 +31,10 @@ typedef uint64_t Time; #define UNREACHABLE __builtin_trap(); bool streq(string s1, string s2); + Time get_current_time(void); +void nearest_deadline(Time *a, Time b); +int deadline_to_timeout(Time deadline, Time current_time); bool getargb(int argc, char **argv, char *name); string getargs(int argc, char **argv, char *name, char *fallback); diff --git a/src/config.h b/src/config.h index 7c95ac5..64482cb 100644 --- a/src/config.h +++ b/src/config.h @@ -8,8 +8,8 @@ #define REPLICATION_FACTOR 3 -// Health check configuration (in milliseconds) -#define HEALTH_CHECK_INTERVAL 30000 // Send STATE_UPDATE every 30 seconds -#define HEALTH_CHECK_TIMEOUT 90000 // Mark as unhealthy after 90 seconds without response + +#define SYNC_INTERVAL 10 +#define RESPONSE_TIME_LIMIT 90 #endif // CONFIG_INCLUDED diff --git a/src/metadata_server.c b/src/metadata_server.c index 4f86321..be4979e 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -54,15 +54,16 @@ static bool hash_list_contains(HashList *hash_list, SHA256 hash) return false; } -static void chunk_server_peer_init(ChunkServerPeer *chunk_server) +static void chunk_server_peer_init(ChunkServerPeer *chunk_server, Time current_time) { + 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); - chunk_server->last_sync_time = INVALID_TIME; - chunk_server->last_response_time = INVALID_TIME; + chunk_server->last_sync_time = current_time; + chunk_server->last_response_time = current_time; } static void chunk_server_peer_free(ChunkServerPeer *chunk_server) @@ -70,6 +71,7 @@ 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); + chunk_server->used = false; } static bool chunk_server_peer_contains(ChunkServerPeer *chunk_server, SHA256 hash) @@ -720,15 +722,14 @@ chunk_server_from_conn(MetadataServer *state, int conn_idx) return &state->chunk_servers[tag]; } -static int send_state_update(MetadataServer *state, int conn_idx, int chunk_server_idx) +static int send_state_update(MetadataServer *state, int chunk_server_idx) { ChunkServerPeer *chunk_server = &state->chunk_servers[chunk_server_idx]; - - if (!chunk_server->auth) - return -1; + int conn_idx = tcp_index_from_tag(&state->tcp, chunk_server_idx); // Create STATE_UPDATE message ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_STATE_UPDATE); @@ -749,14 +750,6 @@ static int send_state_update(MetadataServer *state, int conn_idx, int chunk_serv if (!message_writer_free(&writer)) return -1; - // Update last sync time - chunk_server->last_sync_time = get_current_time(); - - if (state->trace) { - printf("Sent STATE_UPDATE to chunk server %d (add=%u, rem=%u)\n", - chunk_server_idx, add_count, rem_count); - } - return 0; } @@ -974,6 +967,8 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co return -1; state->num_chunk_servers = 0; + for (int i = 0; i < MAX_CHUNK_SERVERS; i++) + state->chunk_servers[i].used = false; tcp_context_init(&state->tcp); @@ -1011,6 +1006,12 @@ 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 + } + for (int i = 0; i < num_events; i++) { int conn_idx = events[i].conn_idx; switch (events[i].type) { @@ -1048,19 +1049,33 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * if (tcp_get_tag(&state->tcp, conn_idx) == CONNECTION_TAG_UNKNOWN) { if (is_chunk_server_message_type(msg_type)) { - 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, true); + + if (state->num_chunk_servers == MAX_CHUNK_SERVERS) { + assert(0); // TODO + } + int j = 0; + while (state->chunk_servers[j].used) { + j++; + assert(j < MAX_CHUNK_SERVERS); + } + + chunk_server_peer_init(&state->chunk_servers[j], current_time); + + tcp_set_tag(&state->tcp, conn_idx, j, true); + } else { + tcp_set_tag(&state->tcp, conn_idx, CONNECTION_TAG_CLIENT, false); } } - if (tcp_get_tag(&state->tcp, conn_idx) == CONNECTION_TAG_CLIENT) + int tag = tcp_get_tag(&state->tcp, conn_idx); + if (tag == CONNECTION_TAG_CLIENT) { ret = process_client_message(state, conn_idx, msg_type, msg); - else + } else { + state->chunk_servers[tag].last_response_time = current_time; ret = process_chunk_server_message(state, conn_idx, msg_type, msg); - + } if (ret < 0) { tcp_close(&state->tcp, conn_idx); break; @@ -1073,74 +1088,37 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd * } } - // Implement periodic health checks: send STATE_UPDATE to chunk servers - Time current_time = get_current_time(); - if (current_time == INVALID_TIME) { - *timeout = -1; - return tcp_register_events(&state->tcp, contexts, polled); - } + Time next_wakeup = INVALID_TIME; - Time next_sync_time = INVALID_TIME; - bool has_authenticated_servers = false; + // Trigger chunk server timing events + for (int i = 0, j = 0; j < state->num_chunk_servers; i++) { - // Iterate through all connections to find chunk servers - for (int conn_idx = 0; conn_idx < MAX_CONNS; conn_idx++) { - int tag = tcp_get_tag(&state->tcp, conn_idx); - if (tag < 0 || tag >= state->num_chunk_servers) + ChunkServerPeer *chunk_server = &state->chunk_servers[i]; + if (!chunk_server->used) continue; + j++; - ChunkServerPeer *chunk_server = &state->chunk_servers[tag]; - if (!chunk_server->auth) + Time response_timeout = chunk_server->last_response_time + (Time) RESPONSE_TIME_LIMIT * 1000000000; + if (current_time > response_timeout) { + // TODO: drop the chunk server continue; - - has_authenticated_servers = true; - - // Check if we should send STATE_UPDATE - bool should_send = false; - - if (chunk_server->last_sync_time == INVALID_TIME) { - // Never sent STATE_UPDATE, send immediately - should_send = true; - } else { - Time time_since_sync = current_time - chunk_server->last_sync_time; - if (time_since_sync >= HEALTH_CHECK_INTERVAL) { - should_send = true; - } else { - // Calculate when next sync should happen - Time this_next_sync = chunk_server->last_sync_time + HEALTH_CHECK_INTERVAL; - if (next_sync_time == INVALID_TIME || this_next_sync < next_sync_time) - next_sync_time = this_next_sync; - } } + nearest_deadline(&next_wakeup, response_timeout); - // Send STATE_UPDATE if needed - if (should_send) { - send_state_update(state, conn_idx, tag); - - // Update next_sync_time - Time this_next_sync = chunk_server->last_sync_time + HEALTH_CHECK_INTERVAL; - if (next_sync_time == INVALID_TIME || this_next_sync < next_sync_time) - next_sync_time = this_next_sync; - } - - // Check if chunk server is unhealthy (no response for too long) - if (chunk_server->last_response_time != INVALID_TIME) { - Time time_since_response = current_time - chunk_server->last_response_time; - if (time_since_response >= HEALTH_CHECK_TIMEOUT && state->trace) { - printf("WARNING: Chunk server %d has not responded for %llu ms\n", - tag, (unsigned long long)time_since_response); + 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); } } - // Calculate timeout for next poll - if (!has_authenticated_servers) { - *timeout = -1; // No authenticated chunk servers, no timeout needed - } else { - // Set a fixed periodic timeout for health checks - // We'll check and send STATE_UPDATE as needed on each iteration - *timeout = 10000; // Wake up every 10 seconds to check - } - + *timeout = deadline_to_timeout(next_wakeup, current_time); return tcp_register_events(&state->tcp, contexts, polled); } diff --git a/src/metadata_server.h b/src/metadata_server.h index 7cb3539..7edc499 100644 --- a/src/metadata_server.h +++ b/src/metadata_server.h @@ -17,6 +17,7 @@ typedef struct { typedef struct { + bool used; bool auth; int num_addrs; @@ -36,9 +37,10 @@ typedef struct { // Time when last STATE_UPDATE was sent Time last_sync_time; + bool last_sync_done; // Time when last response was received - Time last_response_time; + Time last_response_time; // TODO: don't init to INVALID_TIME but current_time } ChunkServerPeer; diff --git a/src/tcp.c b/src/tcp.c index 04a9b9c..1affff1 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -54,6 +54,8 @@ static SOCKET create_listen_socket(string addr, uint16_t port) return INVALID_SOCKET; } + // TODO: mark address as reusable in debug builds + char tmp[1<<10]; if (addr.len >= (int) sizeof(tmp)) { CLOSE_SOCKET(fd);