Start work on DST

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

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;

src/main.c

@@ -1,4 +1,4 @@
-#ifndef BUILD_LIBRARY
+#ifdef BUILD_SERVER
 
 #include <string.h>
 
@@ -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

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

src/system.h

@@ -0,0 +1,36 @@
+#ifndef SYSTEM_INCLUDED
+#define SYSTEM_INCLUDED
+
+#ifdef _WIN32
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#else
+#include <poll.h>
+#include <errno.h>
+#include <unistd.h>
+#include <arpa/inet.h>
+#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

src/tcp.c

@@ -2,6 +2,7 @@
 #include <string.h>
 
 #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;

src/tcp.h

@@ -3,22 +3,7 @@
 
 #include <stdbool.h>
 
-#ifdef _WIN32
-#include <winsock2.h>
-#include <ws2tcpip.h>
-#define POLL WSAPoll
-#define CLOSE_SOCKET closesocket
-#else
-#include <poll.h>
-#include <errno.h>
-#include <unistd.h>
-#include <arpa/inet.h>
-#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

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(&current_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(&current_process->desc[fd].data_socket.input))
+                polled[i].revents |= POLLIN;
+        }
+
+        if (polled[i].events & POLLOUT) {
+            if (!byte_queue_full(&current_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 = &current_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 = &current_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