diff --git a/src/basic.c b/src/basic.c index d6a72f0..9c81fd6 100644 --- a/src/basic.c +++ b/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; } diff --git a/src/chunk_server.c b/src/chunk_server.c index 24131cd..5ea2fe0 100644 --- a/src/chunk_server.c +++ b/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 - bool should_sync = false; - if (should_sync) { - if (send_sync_message(state) < 0) { - assert(0); // TODO + if (state->disconnect_time == INVALID_TIME) { + Time next_sync_time = state->last_sync_time + (Time) SYNC_INTERVAL * 1000000000; + if (current_time >= next_sync_time) { + if (send_sync_message(state) < 0) { + assert(0); // TODO + } + state->last_sync_time = current_time; + } else { + nearest_deadline(&deadline, next_sync_time); } } diff --git a/src/chunk_server.h b/src/chunk_server.h index e3c7854..34e89b8 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -35,6 +35,7 @@ typedef struct { Address remote_addr; Time disconnect_time; + Time last_sync_time; TCP tcp; diff --git a/src/hash_set.c b/src/hash_set.c index d4075d8..a2ea582 100644 --- a/src/hash_set.c +++ b/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) diff --git a/src/metadata_server.c b/src/metadata_server.c index 5533cbe..7b0688b 100644 --- a/src/metadata_server.c +++ b/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); diff --git a/src/system.c b/src/system.c index be02667..6a7fecd 100644 --- a/src/system.c +++ b/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 = ¤t_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; - - for (int j = 0, k = 0; k < current_process->num_desc; j++) { - - assert(j < MAX_DESCRIPTORS); - Descriptor *desc = ¤t_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++; - } - } + 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; + 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);