cozis/ToastyFS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rewrote simulation scheduler and fixed some bugs

Browse Source
This commit is contained in:
cozis
2025-11-14 01:16:48 +01:00
parent 3a2173c154
commit 9a944feb01
6 changed files with 164 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Makefile
+3 -3
View File
@@ -1,5 +1,5 @@
CFLAGS = -Wall -Wextra -ggdb
CFLAGS = -Wall -Wextra -ggdb -fsanitize=address,undefined
COVERAGE_CFLAGS = $(CFLAGS) --coverage
COVERAGE_LFLAGS = --coverage
@@ -22,10 +22,10 @@ all: mousefs$(EXT) mousefs_random_test$(EXT) example_client$(EXT) libmousefs.a
coverage: mousefs_random_test_coverage$(EXT)
coverage-report:
@./scripts/measure_coverage.sh 5
@./scripts/measure_coverage.sh 60
coverage-html:
@./scripts/measure_coverage.sh 5 --html
@./scripts/measure_coverage.sh 60 --html
mousefs$(EXT): $(CFILES) $(HFILES)
gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER
src/chunk_server.c
+12 -3
View File
@@ -871,6 +871,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
start_connecting_to_metadata_server(state);
// TODO: add all chunk hashes to the add list
printf("Chunk server set up (local=%.*s:%d, remote=%.*s:%d, path=%.*s)\n",
addr.len,
addr.ptr,
@@ -911,14 +913,16 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
switch (events[i].type) {
case EVENT_CONNECT:
if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER)
state->disconnect_time = INVALID_TIME;
if (tcp_get_tag(&state->tcp, conn_idx) == TAG_METADATA_SERVER) {
assert(state->disconnect_time == INVALID_TIME);
}
break;
case EVENT_DISCONNECT:
switch (tcp_get_tag(&state->tcp, conn_idx)) {
switch (events[i].tag) {
case TAG_METADATA_SERVER:
assert(state->disconnect_time == INVALID_TIME);
state->disconnect_time = current_time;
break;
@@ -977,6 +981,11 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
}
}
{
int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER);
assert((conn_idx < 0) == (state->disconnect_time != INVALID_TIME));
}
Time deadline = INVALID_TIME;
// Remove items from the remove list that got too old
src/metadata_server.c
+7 -3
View File
@@ -870,6 +870,8 @@ static int process_chunk_server_sync_3(MetadataServer *state,
HashSet tmp_list;
hash_set_init(&tmp_list);
message_write(&writer, &count, sizeof(count));
for (uint32_t i = 0; i < count; i++) {
SHA256 hash;
@@ -933,6 +935,7 @@ static bool is_chunk_server_message_type(uint16_t type)
switch (type) {
case MESSAGE_TYPE_AUTH:
case MESSAGE_TYPE_SYNC:
case MESSAGE_TYPE_SYNC_3:
return true;
default:
@@ -1010,9 +1013,9 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
case EVENT_DISCONNECT:
{
int tag = tcp_get_tag(&state->tcp, conn_idx);
if (tag >= 0) {
chunk_server_peer_free(&state->chunk_servers[tag]);
if (events[i].tag >= 0) {
chunk_server_peer_free(&state->chunk_servers[events[i].tag]);
assert(state->num_chunk_servers > 0);
state->num_chunk_servers--;
}
}
@@ -1048,6 +1051,7 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
}
chunk_server_peer_init(&state->chunk_servers[j], current_time);
state->num_chunk_servers++;
tcp_set_tag(&state->tcp, conn_idx, j, true);
src/system.c
+137 -84
View File
@@ -571,17 +571,6 @@ 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;
if (a->wakeup_time < b->wakeup_time) return -1;
if (a->wakeup_time > b->wakeup_time) return +1;
return 0;
}
static int setup_poll_array(void **contexts, struct pollfd *polled)
{
int num_polled = 0;
@@ -653,80 +642,8 @@ static int setup_poll_array(void **contexts, struct pollfd *polled)
return num_polled;
}
void update_simulation(void)
static void update_network(void)
{
// 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 = ordered_processes[i];
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
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;
assert(current_time <= wakeup_time);
current_time = wakeup_time;
}
int timeout = -1;
switch (current_process->type) {
case PROCESS_TYPE_METADATA_SERVER:
num_polled = metadata_server_step(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_process->simulation_client,
contexts,
polled,
num_polled,
&timeout
);
break;
}
if (num_polled < 0) {
assert(0); // TODO
}
if (timeout < 0) {
current_process->wakeup_time = INVALID_TIME;
} else {
current_process->wakeup_time = current_time + (Time) timeout * 1000000;
}
process_poll_array(current_process, contexts, polled, num_polled);
current_process = NULL;
}
for (int i = 0; i < num_processes; i++) {
for (int j = 0, k = 0; k < processes[i]->num_desc; j++) {
@@ -786,6 +703,142 @@ void update_simulation(void)
}
}
void update_simulation(void)
{
for (;;) {
int num_ready = 0;
for (int i = 0; i < num_processes; i++) {
current_process = processes[i];
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled = setup_poll_array(contexts, polled);
if (num_polled > 0) {
num_ready++;
int timeout = -1;
switch (current_process->type) {
case PROCESS_TYPE_METADATA_SERVER:
num_polled = metadata_server_step(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_process->simulation_client,
contexts,
polled,
num_polled,
&timeout
);
break;
}
if (num_polled < 0) {
assert(0); // TODO
}
if (timeout < 0) {
current_process->wakeup_time = INVALID_TIME;
} else {
current_process->wakeup_time = current_time + (Time) timeout * 1000000;
}
process_poll_array(current_process, contexts, polled, num_polled);
}
current_process = NULL;
update_network();
}
if (num_ready == 0)
break;
}
Process *next_process = NULL;
for (int i = 0; i < num_processes; i++)
if (processes[i]->wakeup_time != INVALID_TIME)
if (next_process == NULL || processes[i]->wakeup_time < next_process->wakeup_time)
next_process = processes[i];
if (next_process == NULL) {
assert(0); // Nothing to schedule next
}
assert(current_time <= next_process->wakeup_time);
current_time = next_process->wakeup_time;
current_process = next_process;
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled = setup_poll_array(contexts, polled);
int timeout = -1;
switch (current_process->type) {
case PROCESS_TYPE_METADATA_SERVER:
num_polled = metadata_server_step(
&current_process->metadata_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CHUNK_SERVER:
num_polled = chunk_server_step(
&current_process->chunk_server,
contexts,
polled,
num_polled,
&timeout
);
break;
case PROCESS_TYPE_CLIENT:
num_polled = simulation_client_step(
&current_process->simulation_client,
contexts,
polled,
num_polled,
&timeout
);
break;
}
if (num_polled < 0) {
assert(0); // TODO
}
if (timeout < 0) {
current_process->wakeup_time = INVALID_TIME;
} else {
current_process->wakeup_time = current_time + (Time) timeout * 1000000;
}
process_poll_array(current_process, contexts, polled, num_polled);
current_process = NULL;
update_network();
}
void *mock_malloc(size_t len)
{
return malloc(len);
src/tcp.c
+4 -4
View File
@@ -240,8 +240,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd
if (set_socket_blocking(new_fd, false) < 0)
CLOSE_SOCKET(new_fd);
else {
events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns };
conn_init(&tcp->conns[tcp->num_conns++], new_fd, false);
events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns-1, tcp->conns[tcp->num_conns-1].tag };
}
}
}
@@ -266,7 +266,7 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd
defer_close = true;
else {
conn->connecting = false;
events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns };
events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns, conn->tag };
}
}
@@ -319,8 +319,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd
removed[i] = defer_close;
if (0) {}
else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns };
else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns };
else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns, conn->tag };
else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns, conn->tag };
}
}
src/tcp.h
+1
View File
@@ -23,6 +23,7 @@ typedef enum {
typedef struct {
EventType type;
int conn_idx;
int tag;
} Event;
typedef struct {