From 31a4b2cc3284c337bd0ff8a33a4095c60ff1ddf1 Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Mon, 3 Nov 2025 18:32:14 +0100 Subject: [PATCH] Bug fixes --- src/client.c | 7 +++---- src/message.c | 2 +- src/metadata_server.c | 46 +++++++++++++++++++++---------------------- src/system.c | 6 ++++++ src/tcp.c | 8 +++++--- 5 files changed, 38 insertions(+), 31 deletions(-) diff --git a/src/client.c b/src/client.c index 4d745b8..98c5ef8 100644 --- a/src/client.c +++ b/src/client.c @@ -15,8 +15,7 @@ #include "message.h" #include -#define TAG_METADATA_SERVER -2 -#define TAG_METADATA_SERVER_TO_CLIENT -3 +#define TAG_METADATA_SERVER -1 #define TAG_RETRIEVE_METADATA_FOR_READ 1 #define TAG_RETRIEVE_METADATA_FOR_WRITE 2 @@ -977,7 +976,7 @@ int tinydfs_process_events(TinyDFS *tdfs, void **contexts, struct pollfd *polled RequestQueue *reqs; int tag = tcp_get_tag(&tdfs->tcp, conn_idx); - if (tag == TAG_METADATA_SERVER_TO_CLIENT) + if (tag == TAG_METADATA_SERVER) reqs = &tdfs->metadata_server.reqs; else { assert(tag > -1); @@ -1006,7 +1005,7 @@ int tinydfs_process_events(TinyDFS *tdfs, void **contexts, struct pollfd *polled RequestQueue *reqs; int tag = tcp_get_tag(&tdfs->tcp, conn_idx); - if (tag == TAG_METADATA_SERVER_TO_CLIENT) + if (tag == TAG_METADATA_SERVER) reqs = &tdfs->metadata_server.reqs; else { assert(tag > -1); diff --git a/src/message.c b/src/message.c index 32ccdcf..3462128 100644 --- a/src/message.c +++ b/src/message.c @@ -15,7 +15,7 @@ bool binary_read(BinaryReader *reader, void *dst, int len) void message_writer_init(MessageWriter *writer, ByteQueue *output, uint16_t type) { uint16_t version = MESSAGE_VERSION; - uint16_t dummy = 0; // Dummy value + uint32_t dummy = 0; // Dummy value writer->output = output; writer->start = byte_queue_offset(output); byte_queue_write(output, &version, sizeof(version)); diff --git a/src/metadata_server.c b/src/metadata_server.c index 5fa4723..7eadbf8 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -185,26 +185,26 @@ process_client_create(MetadataServer *state, int conn_idx, ByteView msg) char path_mem[1<<10]; uint16_t path_len; - if (binary_read(&reader, &path_len, sizeof(path_len))) + if (!binary_read(&reader, &path_len, sizeof(path_len))) return -1; if (path_len > sizeof(path_mem)) return -2; - if (binary_read(&reader, &path_mem, path_len)) + if (!binary_read(&reader, &path_mem, path_len)) return -1; string path = { path_mem, path_len }; uint8_t is_dir; - if (binary_read(&reader, &is_dir, sizeof(path_len))) + if (!binary_read(&reader, &is_dir, sizeof(is_dir))) return -1; uint32_t chunk_size; if (is_dir) chunk_size = 0; else { - if (binary_read(&reader, &chunk_size, sizeof(chunk_size))) + if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) return -1; } @@ -256,13 +256,13 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg) char path_mem[1<<10]; uint16_t path_len; - if (binary_read(&reader, &path_len, sizeof(path_len))) + if (!binary_read(&reader, &path_len, sizeof(path_len))) return -1; if (path_len > sizeof(path_mem)) return -2; - if (binary_read(&reader, &path_mem, path_len)) + if (!binary_read(&reader, &path_mem, path_len)) return -1; string path = { path_mem, path_len }; @@ -315,13 +315,13 @@ process_client_list(MetadataServer *state, int conn_idx, ByteView msg) char path_mem[1<<10]; uint16_t path_len; - if (binary_read(&reader, &path_len, sizeof(path_len))) + if (!binary_read(&reader, &path_len, sizeof(path_len))) return -1; if (path_len > sizeof(path_mem)) return -2; - if (binary_read(&reader, &path_mem, path_len)) + if (!binary_read(&reader, &path_mem, path_len)) return -1; string path = { path_mem, path_len }; @@ -401,23 +401,23 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg) char path_mem[1<<10]; uint16_t path_len; - if (binary_read(&reader, &path_len, sizeof(path_len))) + if (!binary_read(&reader, &path_len, sizeof(path_len))) return -1; if (path_len > sizeof(path_mem)) return -2; - if (binary_read(&reader, &path_mem, path_len)) + if (!binary_read(&reader, &path_mem, path_len)) return -1; string path = { path_mem, path_len }; uint32_t offset; - if (binary_read(&reader, &offset, sizeof(offset))) + if (!binary_read(&reader, &offset, sizeof(offset))) return -1; uint32_t length; - if (binary_read(&reader, &length, sizeof(length))) + if (!binary_read(&reader, &length, sizeof(length))) return -1; // Check that there are no more bytes to read @@ -502,27 +502,27 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) char path_mem[1<<10]; uint16_t path_len; - if (binary_read(&reader, &path_len, sizeof(path_len))) + if (!binary_read(&reader, &path_len, sizeof(path_len))) return -1; if (path_len > sizeof(path_mem)) return -2; - if (binary_read(&reader, &path_mem, path_len)) + if (!binary_read(&reader, &path_mem, path_len)) return -1; string path = { path_mem, path_len }; uint32_t offset; - if (binary_read(&reader, &offset, sizeof(offset))) + if (!binary_read(&reader, &offset, sizeof(offset))) return -1; uint32_t length; - if (binary_read(&reader, &length, sizeof(length))) + if (!binary_read(&reader, &length, sizeof(length))) return -1; uint32_t num_chunks; - if (binary_read(&reader, &num_chunks, sizeof(num_chunks))) + if (!binary_read(&reader, &num_chunks, sizeof(num_chunks))) return -1; #define MAX_CHUNKS_PER_WRITE 32 @@ -534,29 +534,29 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) for (uint32_t i = 0; i < num_chunks; i++) { SHA256 old_hash; - if (binary_read(&reader, &old_hash, sizeof(old_hash))) + if (!binary_read(&reader, &old_hash, sizeof(old_hash))) return -1; SHA256 new_hash; - if (binary_read(&reader, &new_hash, sizeof(new_hash))) + if (!binary_read(&reader, &new_hash, sizeof(new_hash))) return -1; uint8_t is_ipv4; - if (binary_read(&reader, &is_ipv4, sizeof(is_ipv4))) + if (!binary_read(&reader, &is_ipv4, sizeof(is_ipv4))) return -1; Address addr; addr.is_ipv4 = is_ipv4; if (is_ipv4) { - if (binary_read(&reader, &addr.ipv4, sizeof(addr.ipv4))) + if (!binary_read(&reader, &addr.ipv4, sizeof(addr.ipv4))) return -1; } else { - if (binary_read(&reader, &addr.ipv6, sizeof(addr.ipv6))) + if (!binary_read(&reader, &addr.ipv6, sizeof(addr.ipv6))) return -1; } - if (binary_read(&reader, &addr.port, sizeof(addr.port))) + if (!binary_read(&reader, &addr.port, sizeof(addr.port))) return -1; addrs[i] = addr; diff --git a/src/system.c b/src/system.c index 2aeb800..3664a9c 100644 --- a/src/system.c +++ b/src/system.c @@ -640,6 +640,7 @@ void update_simulation(void) polled[num_polled].fd = (SOCKET) j; polled[num_polled].events = desc->events; polled[num_polled].revents = revents; + contexts[num_polled] = desc->context; num_polled++; } } @@ -651,17 +652,22 @@ void update_simulation(void) assert(current_time <= wakeup_time); current_time = wakeup_time; + + //printf("T=%2.2f ms\n", (float) current_time / 1000000); } int timeout = -1; switch (current_process->type) { case PROCESS_TYPE_METADATA_SERVER: + //printf(" Metadata server woke up\n"); num_polled = metadata_server_step(¤t_process->metadata_server, contexts, polled, num_polled, &timeout); break; case PROCESS_TYPE_CHUNK_SERVER: + //printf(" Chunk server woke up\n"); num_polled = chunk_server_step(¤t_process->chunk_server, contexts, polled, num_polled, &timeout); break; case PROCESS_TYPE_CLIENT: + //printf(" Client woke up\n"); num_polled = simulation_client_step(¤t_process->simulation_client, contexts, polled, num_polled, &timeout); break; } diff --git a/src/tcp.c b/src/tcp.c index 98122d4..8a80577 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -224,25 +224,25 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd } else { if (polled[i].revents & POLLIN) { + byte_queue_write_setmincap(&conn->input, 1<<9); ByteView buf = byte_queue_write_buf(&conn->input); int num = sys_recv(conn->fd, (char*) buf.ptr, buf.len, 0); if (num == 0) defer_close = true; else if (num < 0) { - if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) + if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) // TODO: does Windows return these error codes or not? defer_close = true; num = 0; } byte_queue_write_ack(&conn->input, num); ByteView msg = byte_queue_read_buf(&conn->input); int ret = message_peek(msg, NULL, NULL); + byte_queue_read_ack(&conn->input, 0); if (ret < 0) { // Invalid message - byte_queue_read_ack(&conn->input, 0); defer_close = true; } else if (ret == 0) { // Still buffering - byte_queue_read_ack(&conn->input, 0); if (byte_queue_full(&conn->input)) defer_close = true; } else { @@ -266,6 +266,8 @@ int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd } } + // TODO: byte_queue_error here? + removed[i] = defer_close; if (0) {} else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns };