Progress

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);

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;

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");
+    spawn_simulated_process("--addr 127.0.0.1 8080 --leader");
-    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 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");

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++;

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)

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 &&
+            int revents = 0;
-                desc->type != DESC_LISTEN_SOCKET &&
+            switch (desc->type) {
-                desc->type != DESC_CONNECTION_SOCKET)
-                continue;
 
-            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 };
+                    peer->revents |= POLLIN;
-                    Descriptor *peer = handle_to_desc(peer_handle);
-                    if (!accept_queue_push(&peer->accept_queue, self_handle)) {
-                        desc->connection_state = CONNECTION_FAILED;
-                        break;
-                    }
 
                     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) {

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