Bug fixes

src/basic.c

@@ -30,7 +30,7 @@ Time get_current_time(void)
         ok = sys_QueryPerformanceFrequency((LARGE_INTEGER*) &freq);
         if (!ok) return INVALID_TIME;
 
-        uint64_t res = 1000 * (double) count / freq;
+        uint64_t res = 1000000000 * (double) count / freq;
         return res;
     }
 #else
@@ -45,12 +45,12 @@ Time get_current_time(void)
         uint64_t sec = time.tv_sec;
         if (sec > UINT64_MAX / 1000000000)
             return INVALID_TIME;
-        res = sec * 1000;
+        res = sec * 1000000000;
 
         uint64_t nsec = time.tv_nsec;
         if (res > UINT64_MAX - nsec)
             return INVALID_TIME;
-        res += nsec / 1000000;
+        res += nsec;
 
         return res;
     }

src/chunk_server.c

@@ -812,6 +812,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
     if (remote_port <= 0 || remote_port >= 1<<16)
         return -1;
 
+    Time current_time = get_current_time();
+
     state->trace = trace;
 
     tcp_context_init(&state->tcp);
@@ -865,6 +867,7 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
     }
     state->remote_addr.port = remote_port;
     state->disconnect_time = INVALID_TIME;
+    state->last_sync_time = current_time;
 
     start_connecting_to_metadata_server(state);
 
@@ -988,11 +991,15 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
         }
     }
 
-    // TODO: periodically send a SYNC message
+    if (state->disconnect_time == INVALID_TIME) {
-    bool should_sync = false;
+        Time next_sync_time = state->last_sync_time + (Time) SYNC_INTERVAL * 1000000000;
-    if (should_sync) {
+        if (current_time >= next_sync_time) {
-        if (send_sync_message(state) < 0) {
+            if (send_sync_message(state) < 0) {
-            assert(0); // TODO
+                assert(0); // TODO
+            }
+            state->last_sync_time = current_time;
+        } else {
+            nearest_deadline(&deadline, next_sync_time);
         }
     }
 

src/chunk_server.h

@@ -35,6 +35,7 @@ typedef struct {
     Address remote_addr;
 
     Time disconnect_time;
+    Time last_sync_time;
 
     TCP tcp;
 

src/hash_set.c

@@ -72,12 +72,32 @@ bool hash_set_contains(HashSet *set, SHA256 hash)
 
 int hash_set_merge(HashSet *dst, HashSet src)
 {
-    assert(0); // TODO
+    HashSet ret;
+    hash_set_init(&ret);
+
+    for (int i = 0; i < dst->count; i++) {
+        if (hash_set_insert(&ret, dst->items[i]) < 0)
+            goto error;
+    }
+
+    for (int i = 0; i < src.count; i++) {
+        if (hash_set_insert(&ret, src.items[i]) < 0)
+            goto error;
+    }
+
+    hash_set_free(dst);
+    *dst = ret;
+    return 0;
+
+error:
+    hash_set_free(&ret);
+    return -1;
 }
 
 void hash_set_remove_set(HashSet *dst, HashSet src)
 {
-    assert(0); // TODO
+    for (int i = 0; i < src.count; i++)
+        hash_set_remove(dst, src.items[i]);
 }
 
 void timed_hash_set_init(TimedHashSet *set)

src/metadata_server.c

@@ -1088,7 +1088,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
 
         Time response_timeout = chunk_server->last_response_time + (Time) RESPONSE_TIME_LIMIT * 1000000000;
         if (current_time > response_timeout) {
-            // TODO: drop the chunk server
+            assert(0); // TODO: drop the chunk server
             continue;
         }
         nearest_deadline(&next_wakeup, response_timeout);

src/system.c

@@ -580,6 +580,77 @@ static int compare_processes(const void *p1, const void *p2)
     return a->wakeup_time - b->wakeup_time;
 }
 
+static int setup_poll_array(void *contexts, struct pollfd *polled)
+{
+    int num_polled = 0;
+
+    for (int j = 0, k = 0; k < current_process->num_desc; j++) {
+
+        assert(j < MAX_DESCRIPTORS);
+        Descriptor *desc = &current_process->desc[j];
+        if (desc->type == DESC_EMPTY)
+            continue;
+        k++;
+
+        int revents = 0;
+        switch (desc->type) {
+
+            case DESC_FILE:
+            assert(0); // 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) {
+                        revents |= POLLIN;
+                    } else {
+                        if (!data_queue_full(&desc->output_data))
+                            revents |= POLLOUT;
+                        if (!data_queue_empty(&peer->output_data))
+                            revents |= POLLIN;
+                    }
+                }
+                break;
+
+                case CONNECTION_FAILED:
+                assert(0); // TODO
+                break;
+            }
+            break;
+        }
+
+        revents &= desc->events;
+        if (revents) {
+            polled[num_polled].fd = (SOCKET) j;
+            polled[num_polled].events = desc->events;
+            polled[num_polled].revents = revents;
+            contexts[num_polled] = desc->context;
+            num_polled++;
+        }
+    }
+
+    return num_polled;
+}
+
 void update_simulation(void)
 {
     // Order processes by wakeup time. Those with no
@@ -596,76 +667,13 @@ void update_simulation(void)
 
         void *contexts[MAX_CONNS+1];
         struct pollfd polled[MAX_CONNS+1];
-        int num_polled = 0;
+        int num_polled = setup_poll_array(contexts, polled);
-
-        for (int j = 0, k = 0; k < current_process->num_desc; j++) {
-
-            assert(j < MAX_DESCRIPTORS);
-            Descriptor *desc = &current_process->desc[j];
-            if (desc->type == DESC_EMPTY)
-                continue;
-            k++;
-
-            int revents = 0;
-            switch (desc->type) {
-
-                case DESC_FILE:
-                assert(0); // 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) {
-                            revents |= POLLIN;
-                        } else {
-                            if (!data_queue_full(&desc->output_data))
-                                revents |= POLLOUT;
-                            if (!data_queue_empty(&peer->output_data))
-                                revents |= POLLIN;
-                        }
-                    }
-                    break;
-
-                    case CONNECTION_FAILED:
-                    assert(0); // TODO
-                    break;
-                }
-                break;
-            }
-
-            revents &= desc->events;
-            if (revents) {
-                polled[num_polled].fd = (SOCKET) j;
-                polled[num_polled].events = desc->events;
-                polled[num_polled].revents = revents;
-                contexts[num_polled] = desc->context;
-                num_polled++;
-            }
-        }
 
         if (num_polled == 0) {
 
             Time wakeup_time = current_process->wakeup_time;
-            if (wakeup_time == INVALID_TIME) continue;
+            if (wakeup_time == INVALID_TIME)
+                continue;
 
             assert(current_time <= wakeup_time);
             current_time = wakeup_time;
@@ -703,13 +711,13 @@ void update_simulation(void)
         }
 
         if (num_polled < 0) {
-            // TODO
+            assert(0); // TODO
         }
 
         if (timeout < 0) {
             current_process->wakeup_time = INVALID_TIME;
         } else {
-            current_process->wakeup_time = current_time + timeout * 1000000;
+            current_process->wakeup_time = current_time + (Time) timeout * 1000000;
         }
 
         process_poll_array(current_process, contexts, polled, num_polled);