From 7bc407134482e9fe385d2ee8a12de8c426caf18a Mon Sep 17 00:00:00 2001 From: Francesco Cozzuto Date: Wed, 29 Oct 2025 10:40:42 +0100 Subject: [PATCH] Start work on DST --- Makefile | 15 ++- examples/main.c | 2 +- src/main.c | 19 +-- src/system.c | 68 ++++++++++ src/system.h | 36 ++++++ src/tcp.c | 29 +++-- src/tcp.h | 17 +-- src/test.c | 330 ++++++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 472 insertions(+), 44 deletions(-) create mode 100644 src/system.c create mode 100644 src/system.h create mode 100644 src/test.c diff --git a/Makefile b/Makefile index 2d24f5c..d282153 100644 --- a/Makefile +++ b/Makefile @@ -14,17 +14,22 @@ HFILES = $(shell find src -name '*.h') .PHONY: all clean -all: tinydfs_server$(EXT) example_client$(EXT) +all: tinydfs_server$(EXT) tinydfs_test$(EXT) example_client$(EXT) tinydfs_server$(EXT): $(CFILES) $(HFILES) - gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc + gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_SERVER + +tinydfs_test$(EXT): $(CFILES) $(HFILES) + gcc -o $@ $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_TEST example_client$(EXT): examples/main.c $(CFILES) $(HFILES) - gcc -o $@ examples/main.c $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc -DBUILD_LIBRARY + gcc -o $@ examples/main.c $(CFILES) $(CFLAGS) $(LFLAGS) -Iinc clean: rm \ tinydfs_server.exe \ tinydfs_server.out \ - example.exe \ - example.out + tinydfs_test.exe \ + tinydfs_test.out \ + example_client.exe \ + example_client.out diff --git a/examples/main.c b/examples/main.c index a905332..8c7f249 100644 --- a/examples/main.c +++ b/examples/main.c @@ -23,7 +23,7 @@ int main(void) return -1; } - char buff_2[] = "This is file 1"; + char buff_2[] = "This is file 2"; if (tinydfs_submit_write(tdfs, "/my_file_1", -1, 0, buff_2, sizeof(buff_2)-1) < 0) { tinydfs_free(tdfs); return -1; diff --git a/src/main.c b/src/main.c index a9b801e..41a380d 100644 --- a/src/main.c +++ b/src/main.c @@ -1,4 +1,4 @@ -#ifndef BUILD_LIBRARY +#ifdef BUILD_SERVER #include @@ -24,10 +24,12 @@ int main(int argc, char **argv) return ret; for (;;) { - metadata_server_step(&state); + ret = metadata_server_step(&state); + if (ret) + return ret; } - metadata_server_free(&state); + ret = metadata_server_free(&state); } else { @@ -37,13 +39,14 @@ int main(int argc, char **argv) return ret; for (;;) { - chunk_server_step(&state); + ret = chunk_server_step(&state); + if (ret) + return ret; } - chunk_server_free(&state); - + ret = chunk_server_free(&state); } - return 0; + return ret; } -#endif // BUILD_LIBRARY +#endif // BUILD_SERVER diff --git a/src/system.c b/src/system.c new file mode 100644 index 0000000..26de8cf --- /dev/null +++ b/src/system.c @@ -0,0 +1,68 @@ +#ifndef BUILD_TEST + +#include "system.h" + +Socket sys_socket(int domain, int type, int protocol) +{ + return socket(domain, type, protocol); +} + +int sys_bind(Socket fd, void *addr, size_t addr_len) +{ + return bind(fd, addr, addr_len); +} + +int sys_listen(Socket fd, int backlog) +{ + return listen(fd, backlog); +} + +int sys_closesocket(Socket fd) +{ +#ifdef _WIN32 + closesocket(fd); +#else + close(fd); +#endif +} + +int sys_poll(struct pollfd *polled, int num_polled, int timeout) +{ +#ifdef _WIN32 + return WSAPoll(polled, num_polled, timeout); +#else + return poll(polled, num_polled, timeout); +#endif +} + +Socket sys_accept(Socket fd, void *addr, int *addr_len) +{ + return accept(fd, addr, addr_len); +} + +int sys_getsockopt(Socket fd, int level, int optname, void *optval, socklen_t *optlen) +{ + return getsockopt(fd, level, optname, optval, optlen); +} + +int sys_setsockopt(Socket fd, int level, int optname, void *optval, socklen_t optlen) +{ + return setsockopt(fd, level, optname, optval, optlen); +} + +int sys_recv(Socket fd, void *dst, int len, int flags) +{ + return recv(fd, dst, len, flags); +} + +int sys_send(Socket fd, void *src, int len, int flags) +{ + return send(fd, src, len, flags); +} + +int sys_connect(Socket fd, void *addr, size_t addr_len) +{ + return connect(fd, addr, addr_len); +} + +#endif // BUILD_TEST diff --git a/src/system.h b/src/system.h new file mode 100644 index 0000000..370ea4e --- /dev/null +++ b/src/system.h @@ -0,0 +1,36 @@ +#ifndef SYSTEM_INCLUDED +#define SYSTEM_INCLUDED + +#ifdef _WIN32 +#include +#include +#else +#include +#include +#include +#include +#define SOCKET int +#define INVALID_SOCKET -1 +#endif + +#if !defined(_WIN32) || defined(BUILD_TEST) +typedef int Socket; +#define BAD_SOCKET ((Socket) -1) +#else +typedef SOCKET Socket; +#define BAD_SOCKET INVALID_SOCKET +#endif + +Socket sys_socket (int domain, int type, int protocol); +int sys_bind (Socket fd, void *addr, size_t addr_len); +int sys_listen (Socket fd, int backlog); +int sys_closesocket (Socket fd); +int sys_poll (struct pollfd *polled, int num_polled, int timeout); +Socket sys_accept (Socket fd, void *addr, int *addr_len); +int sys_getsockopt (Socket fd, int level, int optname, void *optval, socklen_t *optlen); +int sys_setsockopt (Socket fd, int level, int optname, void *optval, socklen_t optlen); +int sys_recv (Socket fd, void *dst, int len, int flags); +int sys_send (Socket fd, void *src, int len, int flags); +int sys_connect (Socket fd, void *addr, size_t addr_len); + +#endif // SYSTEM_INCLUDED diff --git a/src/tcp.c b/src/tcp.c index 22d94b2..cd46e38 100644 --- a/src/tcp.c +++ b/src/tcp.c @@ -2,6 +2,7 @@ #include #include "tcp.h" +#include "system.h" #include "message.h" bool addr_eql(Address a, Address b) @@ -25,7 +26,7 @@ bool addr_eql(Address a, Address b) static SOCKET create_listen_socket(char *addr, uint16_t port) { - SOCKET fd = socket(AF_INET, SOCK_STREAM, 0); + SOCKET fd = sys_socket(AF_INET, SOCK_STREAM, 0); if (fd == INVALID_SOCKET) return INVALID_SOCKET; @@ -35,11 +36,11 @@ static SOCKET create_listen_socket(char *addr, uint16_t port) if (inet_pton(AF_INET, addr, &bind_buf.sin_addr) != 1) return INVALID_SOCKET; - if (bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf))) + if (sys_bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf))) return INVALID_SOCKET; int backlog = 32; - if (listen(fd, backlog) < 0) + if (sys_listen(fd, backlog) < 0) return INVALID_SOCKET; return fd; @@ -58,7 +59,7 @@ static void conn_init(Connection *conn, SOCKET fd, bool connecting) static void conn_free(Connection *conn) { - CLOSE_SOCKET(conn->fd); + sys_closesocket(conn->fd); byte_queue_free(&conn->input); byte_queue_free(&conn->output); } @@ -90,7 +91,7 @@ void tcp_context_init(TCP *tcp) void tcp_context_free(TCP *tcp) { if (tcp->listen_fd != INVALID_SOCKET) - CLOSE_SOCKET(tcp->listen_fd); + sys_closesocket(tcp->listen_fd); } int tcp_index_from_tag(TCP *tcp, int tag) @@ -172,7 +173,7 @@ int tcp_process_events(TCP *tcp, Event *events) } } - POLL(polled, num_polled, -1); + sys_poll(polled, num_polled, -1); bool removed[MAX_CONNS+1]; @@ -181,7 +182,7 @@ int tcp_process_events(TCP *tcp, Event *events) if (polled[i].fd == tcp->listen_fd) { - SOCKET new_fd = accept(tcp->listen_fd, NULL, NULL); + SOCKET new_fd = sys_accept(tcp->listen_fd, NULL, NULL); if (new_fd != INVALID_SOCKET) { events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns }; conn_init(&tcp->conns[tcp->num_conns++], new_fd, false); @@ -202,7 +203,7 @@ int tcp_process_events(TCP *tcp, Event *events) int err = 0; socklen_t len = sizeof(err); - if (getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, (void*) &err, &len) < 0 || err != 0) + if (sys_getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, (void*) &err, &len) < 0 || err != 0) defer_close = true; else { conn->connecting = false; @@ -214,7 +215,7 @@ int tcp_process_events(TCP *tcp, Event *events) if (polled[i].revents & POLLIN) { ByteView buf = byte_queue_write_buf(&conn->input); - int num = recv(conn->fd, (char*) buf.ptr, buf.len, 0); + int num = sys_recv(conn->fd, (char*) buf.ptr, buf.len, 0); if (num == 0) defer_close = true; else if (num < 0) { @@ -243,7 +244,7 @@ int tcp_process_events(TCP *tcp, Event *events) if (polled[i].revents & POLLOUT) { ByteView buf = byte_queue_read_buf(&conn->output); - int num = send(conn->fd, (char*) buf.ptr, buf.len, 0); + int num = sys_send(conn->fd, (char*) buf.ptr, buf.len, 0); if (num < 0) { if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) defer_close = true; @@ -281,7 +282,7 @@ int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output) return -1; int conn_idx = tcp->num_conns; - SOCKET fd = socket(AF_INET, SOCK_STREAM, 0); + SOCKET fd = sys_socket(AF_INET, SOCK_STREAM, 0); if (fd == INVALID_SOCKET) return -1; @@ -291,13 +292,13 @@ int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output) buf.sin_family = AF_INET; buf.sin_port = htons(addr.port); memcpy(&buf.sin_addr, &addr.ipv4, sizeof(IPv4)); - ret = connect(fd, (struct sockaddr*) &buf, sizeof(buf)); + ret = sys_connect(fd, (struct sockaddr*) &buf, sizeof(buf)); } else { struct sockaddr_in6 buf; buf.sin6_family = AF_INET6; buf.sin6_port = htons(addr.port); memcpy(&buf.sin6_addr, &addr.ipv6, sizeof(IPv6)); - ret = connect(fd, (struct sockaddr*) &buf, sizeof(buf)); + ret = sys_connect(fd, (struct sockaddr*) &buf, sizeof(buf)); } bool connecting; @@ -305,7 +306,7 @@ int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output) connecting = false; } else { if (errno != EINPROGRESS) { - CLOSE_SOCKET(fd); + sys_closesocket(fd); return -1; } connecting = true; diff --git a/src/tcp.h b/src/tcp.h index 771a014..d726120 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -3,22 +3,7 @@ #include -#ifdef _WIN32 -#include -#include -#define POLL WSAPoll -#define CLOSE_SOCKET closesocket -#else -#include -#include -#include -#include -#define POLL poll -#define CLOSE_SOCKET close -#define SOCKET int -#define INVALID_SOCKET -1 -#endif - +#include "system.h" #include "byte_queue.h" #define MAX_CONNS 512 diff --git a/src/test.c b/src/test.c new file mode 100644 index 0000000..50cff32 --- /dev/null +++ b/src/test.c @@ -0,0 +1,330 @@ +#ifdef BUILD_TEST + +#include "system.h" +#include "chunk_server.h" +#include "metadata_server.h" + +#define MAX_DESCRIPTORS 1024 + +typedef enum { + DESCRIPTOR_TYPE_EMPTY, + DESCRIPTOR_TYPE_FILE, + DESCRIPTOR_TYPE_PRECONF_SOCKET, + DESCRIPTOR_TYPE_LISTEN_SOCKET, + DESCRIPTOR_TYPE_CONNECTION_SOCKET, +} DescriptorType; + +typedef struct { +} DescriptorFile; + +typedef struct { + + // Generic fields + struct sockaddr_in bind; + bool no_bind; + bool is_listen; + + // Listen socket + int backlog; + + // Data socket + ByteQueue input; + ByteQueue output; + +} DescriptorSocket; + +typedef struct { + + DescriptorType type; + + // Socket fields + struct sockaddr_in bind; + bool no_bind; + + // Listen socket fields + int backlog; + + // Data socket fields + ByteQueue input; + ByteQueue output; + +} Descriptor; + +typedef struct { + int num_desc; + Descriptor desc[MAX_DESCRIPTORS]; +} Process; + +#define MAX_PROCESSES (MAX_CHUNK_SERVERS + 1) + +MetadataServer metadata_server; + +int num_chunk_servers = 0; +ChunkServer chunk_servers[MAX_CHUNK_SERVERS]; + +int num_processes = 0; +Process processes[MAX_PROCESSES]; +Process *current_process; + +Socket sys_socket(int domain, int type, int protocol) +{ + if (domain != AF_INET) { + // TODO: errno + return BAD_SOCKET; + } + + if (type != SOCK_STREAM) { + // TODO: errno + return BAD_SOCKET; + } + + if (protocol != 0) { + // TODO: errno + return BAD_SOCKET; + } + + if (num_processes == MAX_PROCESSES) { + // TODO: errno + return BAD_SOCKET; + } + + Socket fd = 0; + while (current_process->desc[fd].type != DESCRIPTOR_TYPE_EMPTY) + fd++; + + current_process->desc[fd].type = DESCRIPTOR_TYPE_PRECONF_SOCKET; + current_process->desc[fd].no_bind = true; + return fd; +} + +int sys_bind(Socket fd, void *addr, size_t addr_len) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_SOCKET) { + // TODO: errno + return -1; + } + + if (addr_len != sizeof(current_process->desc[fd].sock.bind)) { + // TODO: errno + return -1; + } + + // TODO: maybe check that no one else is listening + // on this port + + current_process->desc[fd].no_bind = false; + memcpy(¤t_process[fd].desc[fd].bind, addr, addr_len); + return 0; +} + +int sys_listen(Socket fd, int backlog) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_SOCKET) { + // TODO: errno + return -1; + } + + current_process->desc[fd].type = DESCRIPTOR_TYPE_LISTEN-SOCKET; + current_process->desc[fd].backlog = backlog; + return 0; +} + +int sys_closesocket(Socket fd) +{ + current_process->desc[fd].type = DESCRIPTOR_TYPE_EMPTY; + return 0; +} + +int sys_poll(struct pollfd *polled, int num_polled, int timeout) +{ + int num = 0; + for (int i = 0; i < num_polled; i++) { + + polled[i].revents = 0; + + Socket fd = polled[i].fd; + if (fd < 0) + continue; + + if (polled[i].events & POLLNVAL) { + // TODO + } + + if (polled[i].events & POLLERR) { + // TODO + } + + if (polled[i].events & POLLHUP) { + // TODO + } + + if (polled[i].events & POLLIN) { + if (!byte_queue_empty(¤t_process->desc[fd].data_socket.input)) + polled[i].revents |= POLLIN; + } + + if (polled[i].events & POLLOUT) { + if (!byte_queue_full(¤t_process->desc[fd].data_socket.output)) + polled[i].revents |= POLLOUT; + } + + if (polled[i].revents) + num++; + } + + return num; +} + +Socket sys_accept(Socket fd, void *addr, int *addr_len) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_LISTEN_SOCKET) { + // TODO: errno + return BAD_SOCKET; + } + + // TODO +} + +int sys_getsockopt(Socket fd, int level, int optname, void *optval, socklen_t *optlen) +{ + if (level != SOL_SOCKET) { + // TODO: errno + return -1; + } + + switch (optname) { + + case SO_ERROR: + { + if (optlen == NULL || *optlen != sizeof(int)) { + // TODO: errno + return -1; + } + + // TODO + } + break; + + default: + // TODO + break; + } + + // TODO +} + +int sys_setsockopt(Socket fd, int level, int optname, void *optval, socklen_t optlen) +{ + // TODO: errno + return -1; +} + +int sys_recv(Socket fd, void *dst, int len, int flags) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { + // TODO: errno + return -1; + } + + ByteQueue *input = ¤t_process->desc[fd].data_socket.input; + ByteView buf = byte_queue_read_buf(input); + if (buf.len > len) + buf.len = len; + memcpy(dst, buf.ptr, buf.len); + byte_queue_read_ack(input, buf.len); + + return buf.len; +} + +int sys_send(Socket fd, void *src, int len, int flags) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) { + // TODO: errno + return -1; + } + + ByteQueue *output = ¤t_process->desc[fd].data_socket.output; + ByteView buf = byte_queue_write_buf(output); + if (buf.len > len) + buf.len = len; + memcpy(buf.ptr, src, buf.len); + byte_queue_write_ack(output, buf.len); + + return buf.len; +} + +int sys_connect(Socket fd, void *addr, size_t addr_len) +{ + if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_SOCKET) { + // TODO: errno + return -1; + } + + current_process->desc[fd].type = DESCRIPTOR_TYPE_CONNECTION_SOCKET; + + // TODO + + errno = EINPROGRESS; + return 0; +} + +int main(void) +{ + num_processes = 8; + + if (num_processes == 0) { + // TODO + return -1; + } + + current_process = &processes[0]; + int ret = metadata_server_init(&metadata_server, 0, NULL); + if (ret) { + // TODO + return -1; + } + + for (int i = 0; i < num_processes-1; i++) { + current_process = &processes[i+1]; + ret = chunk_server_init(&chunk_servers[i], 0, NULL); + if (ret) { + // TODO + return -1; + } + } + + for (;;) { + for (int i = 0; i < num_processes; i++) { + + current_process = &processes[i]; + + int ret; + if (i == 0) ret = metadata_server_step(&metadata_server); + else ret = chunk_server_step(&chunk_servers[i-1]); + + if (ret) { + // TODO + } + } + + // TODO: process network transit + } + + for (int i = 0; i < num_processes-1; i++) { + ret = chunk_server_free(&chunk_servers[i]); + if (ret) { + // TODO + return -1; + } + } + + ret = metadata_server_free(&metadata_server); + if (ret) { + // TODO + return -1; + } + + return 0; +} + +#endif // BUILD_TEST