diff --git a/src/chunk_server.c b/src/chunk_server.c index e844dc0..920b587 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -831,11 +831,13 @@ 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->metadata_server_disconnect_time = 0; break; case EVENT_DISCONNECT: + printf("Dropped connection to chunk server\n"); switch (tcp_get_tag(&state->tcp, conn_idx)) { case TAG_METADATA_SERVER: state->metadata_server_disconnect_time = current_time; @@ -867,6 +869,8 @@ 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/main_server.c b/src/main_server.c index 2f8fcba..506a6d2 100644 --- a/src/main_server.c +++ b/src/main_server.c @@ -24,7 +24,7 @@ int metadata_server_main(int argc, char **argv) &state, argc, argv, contexts, polled, &timeout); if (num_polled < 0) return -1; for (;;) { - POLL(polled, num_polled, -1); + POLL(polled, num_polled, timeout); num_polled = metadata_server_step( &state, contexts, polled, num_polled, &timeout); if (num_polled < 0) return -1; @@ -44,7 +44,7 @@ int chunk_server_main(int argc, char **argv) &state, argc, argv, contexts, polled, &timeout); if (num_polled < 0) return -1; for (;;) { - POLL(polled, num_polled, -1); + POLL(polled, num_polled, timeout); num_polled = chunk_server_step( &state, contexts, polled, num_polled, &timeout); if (num_polled < 0) return -1; diff --git a/src/main_test.c b/src/main_test.c index 93ea61b..a42287b 100644 --- a/src/main_test.c +++ b/src/main_test.c @@ -15,26 +15,13 @@ int main(int argc, char **argv) // TODO: set simulation_should_stop=true on ctrl+C - fprintf(stderr, "=== Starting TinyDFS simulation ===\n"); - fflush(stderr); - startup_simulation(); // Spawn metadata server (leader) - fprintf(stderr, "Spawning metadata server...\n"); - fflush(stderr); - int ret = spawn_simulated_process("--addr 127.0.0.1 8080 --leader"); - fprintf(stderr, "spawn returned: %d\n", ret); - fflush(stderr); - fprintf(stderr, "Metadata server spawned\n"); - fflush(stderr); + spawn_simulated_process("--addr 127.0.0.1 8080 --leader"); // Spawn chunk servers - fprintf(stderr, "Spawning chunk server 1...\n"); - fflush(stderr); spawn_simulated_process("--addr 127.0.0.1 8081"); - fprintf(stderr, "Spawning chunk server 2...\n"); - fflush(stderr); spawn_simulated_process("--addr 127.0.0.1 8082"); spawn_simulated_process("--addr 127.0.0.1 8083"); spawn_simulated_process("--addr 127.0.0.1 8084"); diff --git a/src/metadata_server.c b/src/metadata_server.c index 0fcd75d..5fa4723 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -799,11 +799,13 @@ 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]); @@ -826,6 +828,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)) { int chunk_server_idx = state->num_chunk_servers++; diff --git a/src/simulation_client.c b/src/simulation_client.c index 645a817..0b1e2d2 100644 --- a/src/simulation_client.c +++ b/src/simulation_client.c @@ -53,9 +53,8 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv, printf("[Client] Initialized successfully\n"); - // Return no polled descriptors for now; we'll handle them in step *timeout = 0; // Wake up immediately to start processing - return 0; + return tinydfs_process_events(client->tdfs, contexts, polled, 0); } int simulation_client_step(SimulationClient *client, void **contexts, @@ -75,7 +74,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->create_dir_op < 0) { fprintf(stderr, "[Client] Failed to submit create directory operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } client->step++; break; @@ -89,7 +88,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 1: Failed to create directory\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -102,7 +101,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->create_file_op < 0) { fprintf(stderr, "[Client] Failed to submit create file operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } client->step++; break; @@ -116,7 +115,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 3: Failed to create file\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -131,7 +130,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->write_op < 0) { fprintf(stderr, "[Client] Failed to submit write operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } } client->step++; @@ -146,7 +145,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 5: Failed to write data\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -160,7 +159,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->read_op < 0) { fprintf(stderr, "[Client] Failed to submit read operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } client->step++; break; @@ -174,7 +173,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 7: Failed to read data\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -187,7 +186,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->list_op < 0) { fprintf(stderr, "[Client] Failed to submit list operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } client->step++; break; @@ -206,7 +205,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 9: Failed to list directory\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -219,7 +218,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, if (client->delete_op < 0) { fprintf(stderr, "[Client] Failed to submit delete operation\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } client->step++; break; @@ -233,7 +232,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } else { fprintf(stderr, "[Client] Step 11: Failed to delete file\n"); client->state = CLIENT_STATE_DONE; - return -1; + break; } tinydfs_result_free(&result); } @@ -260,7 +259,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, } // Return the poll array from the TinyDFS client - return num_polled; + return tinydfs_process_events(client->tdfs, contexts, polled, 0); } void simulation_client_free(SimulationClient *client) diff --git a/src/system.c b/src/system.c index 762464c..2aeb800 100644 --- a/src/system.c +++ b/src/system.c @@ -119,6 +119,9 @@ typedef struct { // "DELAYED" state. DescriptorAddress connect_address; + // Error number when the connection is FAILED + int connect_errno; + // Data written to this descriptor using "write" // or "send". DataQueue output_data; @@ -354,7 +357,7 @@ int spawn_simulated_process(char *args) if (timeout < 0) { process->wakeup_time = INVALID_TIME; } else { - process->wakeup_time = current_time + timeout; + process->wakeup_time = current_time + timeout * 1000000; } process_poll_array(process, contexts, polled, num_polled); @@ -475,6 +478,30 @@ static bool accept_queue_pop(AcceptQueue *accept_queue, DescriptorHandle *item) return true; } +static void accept_queue_remove(AcceptQueue *queue, DescriptorHandle handle) +{ + int i = 0; + while (i < queue->used && ( + queue->items[i].process != handle.process || + queue->items[i].descriptor_index != handle.descriptor_index || + queue->items[i].generation != handle.generation)) + i++; + + if (i == queue->used) + return; + + for (; i < queue->used-1; i++) { + int u = (queue->head + i + 0) % queue->size; + int v = (queue->head + i + 1) % queue->size; + queue->items[u] = queue->items[v]; + } +} + +static bool accept_queue_empty(AcceptQueue *queue) +{ + return queue->used == 0; +} + static void data_queue_init(DataQueue *queue, int size) { queue->used = 0; @@ -517,12 +544,38 @@ static int data_queue_write(DataQueue *queue, char *src, int len) return num; } +static bool data_queue_empty(DataQueue *queue) +{ + return queue->used == 0; +} + +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; + return a->wakeup_time - b->wakeup_time; +} + void update_simulation(void) { - // TODO: sort processes based on their wakeup time + // 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 = processes[i]; + + current_process = ordered_processes[i]; void *contexts[MAX_CONNS+1]; struct pollfd polled[MAX_CONNS+1]; @@ -535,12 +588,54 @@ void update_simulation(void) continue; k++; - if (desc->type != DESC_SOCKET && - desc->type != DESC_LISTEN_SOCKET && - desc->type != DESC_CONNECTION_SOCKET) - continue; + int revents = 0; + switch (desc->type) { - int revents = desc->events & desc->revents; + case DESC_FILE: + // TODO: error + break; + + case DESC_SOCKET: + // Ignore + break; + + case DESC_LISTEN_SOCKET: + if (!accept_queue_empty(&desc->accept_queue)) + revents |= POLLIN; + break; + + case DESC_CONNECTION_SOCKET: + switch (desc->connection_state) { + + case CONNECTION_DELAYED: + break; + + case CONNECTION_QUEUED: + break; + + case CONNECTION_ESTABLISHED: + { + Descriptor *peer = handle_to_desc(desc->connection_peer); + if (peer == NULL) { + // TODO + } + + if (!data_queue_full(&desc->output_data)) + revents |= POLLOUT; + + if (!data_queue_empty(&peer->output_data)) + revents |= POLLIN; + } + break; + + case CONNECTION_FAILED: + // TODO + break; + } + break; + } + + revents &= desc->events; if (revents) { polled[num_polled].fd = (SOCKET) j; polled[num_polled].events = desc->events; @@ -549,6 +644,15 @@ void update_simulation(void) } } + 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: @@ -569,7 +673,7 @@ void update_simulation(void) if (timeout < 0) { current_process->wakeup_time = INVALID_TIME; } else { - current_process->wakeup_time = current_time + timeout; + current_process->wakeup_time = current_time + timeout * 1000000; } process_poll_array(current_process, contexts, polled, num_polled); @@ -595,16 +699,22 @@ void update_simulation(void) { DescriptorHandle peer_handle; if (!find_peer_by_address(desc->connect_address, &peer_handle)) { + desc->revents |= POLLOUT; desc->connection_state = CONNECTION_FAILED; + desc->connect_errno = EHOSTUNREACH; // TODO: This only works on Linux, not Windows + break; + } + Descriptor *peer = handle_to_desc(peer_handle); + + DescriptorHandle self_handle = { processes[i], j, desc->generation }; + if (!accept_queue_push(&peer->accept_queue, self_handle)) { + desc->revents |= POLLOUT; + desc->connection_state = CONNECTION_FAILED; + desc->connect_errno = ECONNREFUSED; // TODO: This only works on Linux, not Windows break; } - DescriptorHandle self_handle = { processes[i], j, desc->generation }; - Descriptor *peer = handle_to_desc(peer_handle); - if (!accept_queue_push(&peer->accept_queue, self_handle)) { - desc->connection_state = CONNECTION_FAILED; - break; - } + peer->revents |= POLLIN; desc->connection_state = CONNECTION_QUEUED; desc->connection_peer = peer_handle; @@ -615,7 +725,9 @@ void update_simulation(void) { if (handle_to_desc(desc->connection_peer) == NULL) { // Listener closed before accepting + desc->revents |= POLLOUT; desc->connection_state = CONNECTION_FAILED; + desc->connect_errno = ECONNREFUSED; // TODO: This only works on Linux, not Windows break; } } @@ -836,7 +948,44 @@ int mock_getsockopt(SOCKET fd, int level, int optname, void *optval, socklen_t * SET_SOCKET_ERROR(SOCKET_ERROR_INVAL); return -1; } - *(int*)optval = 0; // No error + + int val; + switch (desc->type) { + + case DESC_FILE: + SET_SOCKET_ERROR(SOCKET_ERROR_NOTSOCK); + return -1; + + case DESC_SOCKET: + val = 0; // No error + break; + + case DESC_LISTEN_SOCKET: + val = 0; // No error + break; + + case DESC_CONNECTION_SOCKET: + switch (desc->connection_state) { + + case CONNECTION_DELAYED: + val = 0; // No error + break; + + case CONNECTION_QUEUED: + val = 0; // No error + break; + + case CONNECTION_ESTABLISHED: + val = 0; // No error + break; + + case CONNECTION_FAILED: + val = desc->connect_errno; + break; + } + break; + } + *(int*)optval = val; *optlen = sizeof(int); return 0; } @@ -1007,25 +1156,6 @@ wrap_native_file_into_desc(NATIVE_HANDLE handle) return idx; } -void accept_queue_remove(AcceptQueue *queue, DescriptorHandle handle) -{ - int i = 0; - while (i < queue->used && ( - queue->items[i].process != handle.process || - queue->items[i].descriptor_index != handle.descriptor_index || - queue->items[i].generation != handle.generation)) - i++; - - if (i == queue->used) - return; - - for (; i < queue->used-1; i++) { - int u = (queue->head + i + 0) % queue->size; - int v = (queue->head + i + 1) % queue->size; - queue->items[u] = queue->items[v]; - } -} - static void close_desc(Descriptor *desc) { switch (desc->type) { diff --git a/src/tcp.c b/src/tcp.c index 87e6747..98122d4 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -178,7 +178,7 @@ int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled) } } - return 0; + return num_polled; } // The "events" array must be an array of capacity MAX_CONNS+1