Progress

src/file_system.c

@@ -1,3 +1,7 @@
+#include <limits.h>
+#include <string.h>
+#include <assert.h>
+
 #include "system.h"
 #include "file_system.h"
 

src/main_test.c

@@ -1,4 +1,8 @@
 #ifdef BUILD_TEST
+
+#include <signal.h>
+#include <stdbool.h>
+
 #include "system.h"
 
 static sig_atomic_t simulation_should_stop = false;

src/system.c

@@ -1,4 +1,6 @@
 #include <assert.h>
+#include <stdlib.h>
+#include <string.h>
 
 #include "system.h"
 #include "chunk_server.h"
@@ -7,8 +9,6 @@
 #ifdef _WIN32
 #define NATIVE_HANDLE HANDLE
 #else
-#define SOCKET int
-#define INVALID_SOCKET -1
 #define NATIVE_HANDLE int
 #endif
 
@@ -22,6 +22,7 @@ typedef struct Process Process;
 
 typedef enum {
     DESC_EMPTY,
+    DESC_FILE,
     DESC_SOCKET,
     DESC_LISTEN_SOCKET,
     DESC_CONNECTION_SOCKET,
@@ -149,6 +150,34 @@ static int num_processes = 0;
 static Process *processes[MAX_PROCESSES];
 static Process *current_process = NULL;
 
+static void process_poll_array(Process *process,
+    void **contexts, struct pollfd *polled, int num_polled)
+{
+    for (int i = 0, j = 0; j < process->num_desc; i++) {
+
+        Descriptor *desc = &process->desc[i];
+        if (desc->type == DESC_EMPTY)
+            continue;
+        j++;
+
+        desc->events = 0;
+        desc->revents = 0;
+        desc->context = NULL;
+    }
+
+    for (int i = 0; i < num_polled; i++) {
+
+        SOCKET fd = polled[i].fd;
+        if (fd == INVALID_SOCKET)
+            continue;
+
+        int idx = (int) fd;
+        process->desc[idx].events = polled[i].events;
+        process->desc[idx].revents = 0;
+        process->desc[idx].context = contexts[i];
+    }
+}
+
 static bool is_leader(int argc, char **argv)
 {
     for (int i = 0; i < argc; i++)
@@ -157,11 +186,45 @@ static bool is_leader(int argc, char **argv)
     return false;
 }
 
+#define MAX_ARGS 128
+
+static bool is_space(char c)
+{
+    return c == ' ' || c == '\t' || c == '\r' || c == '\n';
+}
+
 int spawn_simulated_process(char *args)
 {
     if (num_processes == MAX_PROCESSES)
         return -1;
 
+    char mem[1<<10];
+    int args_len = strlen(args);
+    if (args_len >= (int) sizeof(mem))
+        return -1;
+    memcpy(mem, args, args_len);
+    mem[args_len] = '\0';
+
+    int argc = 0;
+    char *argv[MAX_ARGS];
+    for (int cur = 0;;) {
+
+        while (cur < args_len && is_space(args[cur]))
+            cur++;
+
+        if (cur == args_len || argc == MAX_ARGS)
+            break;
+
+        argv[argc++] = args + cur;
+
+        while (cur < args_len && !is_space(args[cur]))
+            cur++;
+
+        args[cur] = '\0';
+        if (cur < args_len)
+            cur++;
+    }
+
     bool leader = is_leader(argc, argv);
 
     Process *process = malloc(sizeof(Process));
@@ -192,7 +255,7 @@ int spawn_simulated_process(char *args)
 
 static void free_process(Process *process)
 {
-    if (leader) {
+    if (process->leader) {
         metadata_server_free(&process->metadata_server);
     } else {
         chunk_server_free(&process->chunk_server);
@@ -250,7 +313,7 @@ static Descriptor *handle_to_desc(DescriptorHandle handle)
         || handle.descriptor_index < 0
         || handle.descriptor_index >= MAX_DESCRIPTORS)
         return NULL;
-    Process *process = processes[handle.process_index];
+    Process *process = handle.process;
     Descriptor *desc = &process->desc[handle.descriptor_index];
     if (desc->type == DESC_EMPTY || desc->generation != handle.generation)
         return NULL;
@@ -344,9 +407,28 @@ void update_simulation(void)
         struct pollfd polled[MAX_CONNS+1];
         int num_polled;
 
-        // TODO: fill up poll array
+        for (int j = 0, k = 0; k < current_process->num_desc; j++) {
 
-        if (leader) {
+            Descriptor *desc = &current_process->desc[j];
+            if (desc->type == DESC_EMPTY)
+                continue;
+            k++;
+
+            if (desc->type != DESC_SOCKET &&
+                desc->type != DESC_LISTEN_SOCKET &&
+                desc->type != DESC_CONNECTION_SOCKET)
+                continue;
+
+            int revents = desc->events & desc->revents;
+            if (revents) {
+                polled[num_polled].fd = (SOCKET) j;
+                polled[num_polled].events = desc->events;
+                polled[num_polled].revents = revents;
+                num_polled++;
+            }
+        }
+
+        if (current_process->leader) {
             num_polled = metadata_server_step(&current_process->metadata_server, contexts, polled, num_polled);
         } else {
             num_polled = chunk_server_step(&current_process->chunk_server, contexts, polled, num_polled);
@@ -412,6 +494,31 @@ void update_simulation(void)
     }
 }
 
+void *mock_malloc(size_t len)
+{
+    return malloc(len);
+}
+
+void *mock_realloc(void *ptr, size_t len)
+{
+    return realloc(ptr, len);
+}
+
+void mock_free(void *ptr)
+{
+    free(ptr);
+}
+
+int mock_remove(char *path)
+{
+    return remove(path);
+}
+
+int mock_rename(char *oldpath, char *newpath)
+{
+    return rename(oldpath, newpath);
+}
+
 SOCKET mock_socket(int domain, int type, int protocol)
 {
     if (domain != AF_INET || type != SOCK_STREAM || protocol != 0) {
@@ -434,6 +541,8 @@ SOCKET mock_socket(int domain, int type, int protocol)
     desc->revents = 0;
     desc->context = NULL;
     desc->address = (DescriptorAddress) { .type=DESC_ADDR_VOID };
+
+    current_process->num_desc++;
     return (SOCKET) idx;
 }
 
@@ -543,6 +652,7 @@ SOCKET mock_accept(SOCKET fd, void *addr, socklen_t *addr_len)
     peer->revents |= POLLOUT;
     data_queue_init(&peer->output_data, DATA_QUEUE_SIZE);
 
+    current_process->num_desc++;
     return (SOCKET) new_idx;
 }
 
@@ -645,11 +755,123 @@ int mock_connect(SOCKET fd, void *addr, size_t addr_len)
     return -1;
 }
 
+static NATIVE_HANDLE
+wrap_native_file_into_desc(NATIVE_HANDLE handle)
+{
+    if (current_process->num_desc == MAX_DESCRIPTORS) {
+        // TODO
+        return -1;
+    }
+
+    int idx = 0;
+    while (current_process->desc[idx].type != DESC_EMPTY)
+        idx++;
+
+    Descriptor *desc = &current_process->desc[idx];
+
+    desc->type = DESC_FILE;
+    desc->real_fd = handle;
+
+    current_process->num_desc++;
+    return idx;
+}
+
+void accept_queue_remove(AcceptQueue *queue, DescriptorHandle handle)
+{
+    int i = 0;
+    while (i < queue->used && (
+        queue->items[i].process != handle.process ||
+        queue->items[i].descriptor_index != handle.descriptor_index ||
+        queue->items[i].generation != handle.generation))
+        i++;
+
+    if (i == queue->used)
+        return;
+
+    for (; i < queue->used-1; i++) {
+        int u = (queue->head + i + 0) % queue->size;
+        int v = (queue->head + i + 1) % queue->size;
+        queue->items[u] = queue->items[v];
+    }
+}
+
+static void close_desc(Descriptor *desc)
+{
+    switch (desc->type) {
+
+        case DESC_EMPTY:
+        // TODO
+        break;
+
+        case DESC_FILE:
+#ifdef _WIN32
+        CloseHandle(desc->real_fd);
+#else
+        close(desc->real_fd);
+#endif
+        break;
+
+        case DESC_SOCKET:
+        // TODO
+        break;
+
+        case DESC_LISTEN_SOCKET:
+        accept_queue_free(&desc->accept_queue);
+        break;
+
+        case DESC_CONNECTION_SOCKET:
+        data_queue_free(&desc->output_data);
+        switch (desc->connection_state) {
+
+            case CONNECTION_DELAYED:
+            // TODO
+            break;
+
+            case CONNECTION_QUEUED:
+            {
+                Descriptor *peer = handle_to_desc(desc->connection_peer);
+                if (peer == NULL) break;
+
+                DescriptorHandle self_handle = { current_process, desc - current_process->desc, desc->generation };
+                accept_queue_remove(&peer->accept_queue, self_handle);
+            }
+            break;
+
+            case CONNECTION_ESTABLISHED:
+            // TODO
+            break;
+
+            case CONNECTION_FAILED:
+            // TODO
+            break;
+        }
+        // TODO
+        break;
+    }
+    desc->type = DESC_EMPTY;
+    desc->generation++;
+}
+
 #ifdef _WIN32
 
 int mock_closesocket(SOCKET fd)
 {
-    // TODO
+    if (fd == INVALID_SOCKET) {
+        // TODO
+        return -1;
+    }
+    int idx = (int) fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_SOCKET &&
+        desc->type != DESC_LISTEN_SOCKET &&
+        desc->type != DESC_CONNECTION_SOCKET) {
+        // TODO
+        return -1;
+    }
+
+    close_desc(desc);
+    return 0;
 }
 
 HANDLE mock_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess,
@@ -657,12 +879,31 @@ HANDLE mock_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess,
     DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes,
     HANDLE hTemplateFile)
 {
-    // TODO
+    HANDLE handle = CreateFileW(lpFileName,
+        dwDesiredAccess, dwShareMode,
+        lpSecurityAttributes, dwCreationDisposition,
+        dwFlagsAndAttributes, hTemplateFile);
+    if (handle == INVALID_HANDLE_VALUE)
+        return INVALID_HANDLE_VALUE;
+    return wrap_native_file_into_desc(handle);
 }
 
 BOOL mock_CloseHandle(HANDLE handle)
 {
-    // TODO
+    if (fd == INVALID_SOCKET) {
+        // TODO
+        return -1;
+    }
+    int idx = (int) fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    close_desc(desc);
+    return 0;
 }
 
 BOOL mock_LockFile(HANDLE hFile,
@@ -671,37 +912,119 @@ BOOL mock_LockFile(HANDLE hFile,
     DWORD nNumberOfBytesToLockLow,
     DWORD nNumberOfBytesToLockHigh)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return LockFile(
+        desc->real_fd,
+        dwFileOffsetLow,
+        dwFileOffsetHigh,
+        nNumberOfBytesToLockLow,
+        nNumberOfBytesToLockHigh);
 }
 
 BOOL mock_UnlockFile(
-  HANDLE hFile,
+    HANDLE hFile,
-  DWORD  dwFileOffsetLow,
+    DWORD  dwFileOffsetLow,
-  DWORD  dwFileOffsetHigh,
+    DWORD  dwFileOffsetHigh,
-  DWORD  nNumberOfBytesToUnlockLow,
+    DWORD  nNumberOfBytesToUnlockLow,
-  DWORD  nNumberOfBytesToUnlockHigh)
+    DWORD  nNumberOfBytesToUnlockHigh)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return UnlockFile(
+        desc->real_fd,
+        dwFileOffsetLow,
+        dwFileOffsetHigh,
+        nNumberOfBytesToUnlockLow,
+        nNumberOfBytesToUnlockHigh);
 }
 
 BOOL mock_FlushFileBuffers(HANDLE handle)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return FlushFileBuffers(desc->real_fd);
 }
 
 BOOL mock_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return ReadFile(desc->real_fd, dst, len, num, ov);
 }
 
 BOOL mock_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return WriteFile(desc->real_fd, src, len, num ov);
 }
 
 BOOL mock_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf)
 {
-    // TODO
+    if (handle == INVALID_HANDLE_VALUE) {
+        // TODO
+        return FALSE;
+    }
+    int idx = (int) handle;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return GetFileSizeEx(desc->real_fd, buf);
 }
 
 char *mock__fullpath(char *path, char *dst, int cap)
@@ -723,37 +1046,106 @@ int mock_clock_gettime(clockid_t clockid, struct timespec *tp)
 
 int mock_open(char *path, int flags, int mode)
 {
-    // TODO
+    int fd = open(path, flags, mode);
+    if (fd < 0) return -1;
+
+    return wrap_native_file_into_desc(fd);
 }
 
 int mock_close(int fd)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = (int) fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    close_desc(desc);
+    return 0;
 }
 
 int mock_flock(int fd, int op)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return flock(desc->real_fd, op);
 }
 
 int mock_fsync(int fd)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return fsync(desc->real_fd);
 }
 
 int mock_read(int fd, char *dst, int len)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type == DESC_FILE) {
+        return read(desc->real_fd, dst, len);
+    } else {
+        return mock_recv(fd, dst, len, 0);
+    }
 }
 
 int mock_write(int fd, char *src, int len)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type == DESC_FILE) {
+        return write(desc->real_fd, src, len);
+    } else {
+        return mock_send(fd, src, len, 0);
+    }
 }
 
 int mock_fstat(int fd, struct stat *buf)
 {
-    // TODO
+    if (fd < 0) {
+        // TODO
+        return -1;
+    }
+    int idx = fd;
+
+    Descriptor *desc = &current_process->desc[idx];
+    if (desc->type != DESC_FILE) {
+        // TODO
+        return -1;
+    }
+
+    return fstat(desc->real_fd, buf);
 }
 
 int mock_mkstemp(char *path)

src/system.h

@@ -1,4 +1,8 @@
+#ifndef SYSTEM_INCLUDED
+#define SYSTEM_INCLUDED
+
 #include <stdio.h>
+#include <stdlib.h>
 
 #ifdef _WIN32
 
@@ -15,6 +19,7 @@
 #include <time.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <unistd.h>
 #include <sys/stat.h>
 #include <sys/file.h>
 #include <sys/socket.h>
@@ -31,6 +36,11 @@ int    spawn_simulated_process(char *args);
 void   update_simulation(void);
 void   cleanup_simulation(void);
 
+void*  mock_malloc(size_t len);
+void*  mock_realloc(void *ptr, size_t len);
+void   mock_free(void *ptr);
+int    mock_remove(char *path);
+int    mock_rename(char *oldpath, char *newpath);
 SOCKET mock_socket(int domain, int type, int protocol);
 int    mock_bind(SOCKET fd, void *addr, size_t addr_len);
 int    mock_listen(SOCKET fd, int backlog);
@@ -41,6 +51,32 @@ int    mock_recv(SOCKET fd, void *dst, int len, int flags);
 int    mock_send(SOCKET fd, void *src, int len, int flags);
 int    mock_connect(SOCKET fd, void *addr, size_t addr_len);
 
+#ifdef _WIN32
+int    mock_closesocket(SOCKET fd);
+HANDLE mock_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, HANDLE hTemplateFile);
+BOOL   mock_CloseHandle(HANDLE handle);
+BOOL   mock_LockFile(HANDLE hFile, DWORD dwFileOffsetLow, DWORD dwFileOffsetHigh, DWORD nNumberOfBytesToLockLow, DWORD nNumberOfBytesToLockHigh);
+BOOL   mock_UnlockFile(HANDLE hFile, DWORD dwFileOffsetLow, DWORD dwFileOffsetHigh, DWORD nNumberOfBytesToUnlockLow, DWORD nNumberOfBytesToUnlockHigh);
+BOOL   mock_FlushFileBuffers(HANDLE handle);
+BOOL   mock_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov);
+BOOL   mock_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov);
+BOOL   mock_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf);
+char*  mock__fullpath(char *path, char *dst, int cap);
+int    mock__mkdir(char *path);
+#else
+int    mock_clock_gettime(clockid_t clockid, struct timespec *tp);
+int    mock_open(char *path, int flags, int mode);
+int    mock_close(int fd);
+int    mock_flock(int fd, int op);
+int    mock_fsync(int fd);
+int    mock_read(int fd, char *dst, int len);
+int    mock_write(int fd, char *src, int len);
+int    mock_fstat(int fd, struct stat *buf);
+int    mock_mkstemp(char *path);
+char*  mock_realpath(char *path, char *dst);
+int    mock_mkdir(char *path, mode_t mode);
+#endif
+
 // Common
 #define sys_malloc           mock_malloc
 #define sys_realloc          mock_realloc
@@ -132,3 +168,4 @@ int    mock_connect(SOCKET fd, void *addr, size_t addr_len);
 #define sys_clock_gettime    clock_gettime
 
 #endif
+#endif // SYSTEM_INCLUDED

src/tcp.h

@@ -9,8 +9,6 @@
 #ifdef _WIN32
 #define CLOSE_SOCKET sys_closesocket
 #else
-#define SOCKET int
-#define INVALID_SOCKET -1
 #define CLOSE_SOCKET sys_close
 #endif