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

Test framework cleanup

Browse Source
This commit is contained in:
cozis
2025-11-01 22:44:04 +01:00
parent 7525a93035
commit eb93541c84
8 changed files with 940 additions and 1292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
inc/TinyDFS.h
+10
View File
@@ -4,6 +4,12 @@
#include <stdint.h> #include <stdint.h>
#include <stdbool.h> #include <stdbool.h>
#ifdef _WIN32
#include <winsock2.h>
#else
#include <poll.h>
#endif
typedef struct TinyDFS TinyDFS; typedef struct TinyDFS TinyDFS;
typedef struct { typedef struct {
@@ -36,6 +42,10 @@ typedef struct {
TinyDFS *tinydfs_init(char *addr, uint16_t port); TinyDFS *tinydfs_init(char *addr, uint16_t port);
void tinydfs_free(TinyDFS *tdfs); void tinydfs_free(TinyDFS *tdfs);
void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout); void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout);
bool tinydfs_isdone(TinyDFS *tdfs, int opidx, TinyDFS_Result *result);
int tinydfs_process_events(TinyDFS *tdfs, void **contexts, struct pollfd *polled, int num_polled);
int tinydfs_submit_create (TinyDFS *tdfs, char *path, int path_len, bool is_dir, unsigned int chunk_size); int tinydfs_submit_create (TinyDFS *tdfs, char *path, int path_len, bool is_dir, unsigned int chunk_size);
int tinydfs_submit_delete (TinyDFS *tdfs, char *path, int path_len); int tinydfs_submit_delete (TinyDFS *tdfs, char *path, int path_len);
int tinydfs_submit_list (TinyDFS *tdfs, char *path, int path_len); int tinydfs_submit_list (TinyDFS *tdfs, char *path, int path_len);
src/chunk_server.c
+6 -7
View File
@@ -118,6 +118,7 @@ static int chunk_store_add(ChunkStore *store, string data)
return store_chunk(store, data, &dummy); return store_chunk(store, data, &dummy);
} }
#if 0
static void chunk_store_remove(ChunkStore *store, SHA256 hash) static void chunk_store_remove(ChunkStore *store, SHA256 hash)
{ {
char buf[PATH_MAX]; char buf[PATH_MAX];
@@ -125,6 +126,7 @@ static void chunk_store_remove(ChunkStore *store, SHA256 hash)
remove_file_or_dir(path); remove_file_or_dir(path);
} }
#endif
static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk, static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk,
uint64_t patch_off, string patch, SHA256 *new_hash) uint64_t patch_off, string patch, SHA256 *new_hash)
@@ -279,9 +281,6 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
// Get paths for main and orphaned locations // Get paths for main and orphaned locations
hash2path(&state->store, add_list[i], main_path); hash2path(&state->store, add_list[i], main_path);
snprintf(orphaned_path, sizeof(orphaned_path), "%s/orphaned/", state->store.path); snprintf(orphaned_path, sizeof(orphaned_path), "%s/orphaned/", state->store.path);
string orphaned_dir = { orphaned_path, strlen(orphaned_path) };
string orphaned_file = hash2path(&state->store, add_list[i], orphaned_path);
orphaned_file.ptr = orphaned_path;
// Build orphaned path properly // Build orphaned path properly
strcpy(orphaned_path, state->store.path); strcpy(orphaned_path, state->store.path);
@@ -541,10 +540,10 @@ process_chunk_server_download_success(ChunkServer *state, int conn_idx, ByteView
return -1; return -1;
// Read the chunk data // Read the chunk data
if (reader.cur + data_len > reader.len) if (data_len > (uint32_t) (reader.len - reader.cur))
return -1; return -1;
string data = { reader.src + reader.cur, data_len }; string data = { (char*) reader.src + reader.cur, data_len };
// Store the downloaded chunk // Store the downloaded chunk
if (chunk_store_add(&state->store, data) < 0) { if (chunk_store_add(&state->store, data) < 0) {
@@ -617,7 +616,7 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg)
if (!binary_read(&reader, &target_len, sizeof(target_len))) if (!binary_read(&reader, &target_len, sizeof(target_len)))
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Invalid message"));
string data = { reader.src + reader.cur, target_len }; string data = { (char*) reader.src + reader.cur, target_len };
if (!binary_read(&reader, NULL, target_len)) if (!binary_read(&reader, NULL, target_len))
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Invalid message"));
@@ -678,7 +677,7 @@ process_client_upload_chunk(ChunkServer *state, int conn_idx, ByteView msg)
if (!binary_read(&reader, &data_len, sizeof(data_len))) if (!binary_read(&reader, &data_len, sizeof(data_len)))
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_UPLOAD_CHUNK_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_UPLOAD_CHUNK_ERROR, S("Invalid message"));
string data = { reader.src + reader.cur, data_len }; string data = { (char*) reader.src + reader.cur, data_len };
// Check that there are no more bytes to read // Check that there are no more bytes to read
if (binary_read(&reader, NULL, 1)) if (binary_read(&reader, NULL, 1))
src/client.c
+82 -62
View File
@@ -3,8 +3,10 @@
#include <stdlib.h> #include <stdlib.h>
#ifdef _WIN32 #ifdef _WIN32
#define POLL WSAPoll
#else #else
#include <arpa/inet.h> #include <arpa/inet.h>
#define POLL poll
#endif #endif
#include "tcp.h" #include "tcp.h"
@@ -560,7 +562,7 @@ static void process_event_for_list(TinyDFS *tdfs,
return; return;
} }
char *name = reader.src + reader.cur; char *name = (char*) reader.src + reader.cur;
if (!binary_read(&reader, NULL, name_len)) { if (!binary_read(&reader, NULL, name_len)) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
sys_free(entities); sys_free(entities);
@@ -922,38 +924,70 @@ translate_operation_into_result(TinyDFS *tdfs, int opidx, TinyDFS_Result *result
return true; return true;
} }
void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout) bool tinydfs_isdone(TinyDFS *tdfs, int opidx, TinyDFS_Result *result)
{ {
for (;;) { if (opidx < 0) {
for (int i = 0, j = 0; j < tdfs->num_operations; i++) {
if (opidx < 0) { if (tdfs->operations[i].type == OPERATION_TYPE_FREE)
for (int i = 0, j = 0; j < tdfs->num_operations; i++) { continue;
j++;
if (tdfs->operations[i].type == OPERATION_TYPE_FREE) if (translate_operation_into_result(tdfs, i, result))
continue; return true;
j++;
if (translate_operation_into_result(tdfs, i, result))
return;
}
} else {
if (translate_operation_into_result(tdfs, opidx, result))
return;
} }
} else {
if (translate_operation_into_result(tdfs, opidx, result))
return true;
}
int num_events; return false;
Event events[MAX_CONNS+1]; }
num_events = tcp_process_events(&tdfs->tcp, events); int tinydfs_process_events(TinyDFS *tdfs, void **contexts, struct pollfd *polled, int num_polled)
for (int i = 0; i < num_events; i++) { {
int conn_idx = events[i].conn_idx; int num_events;
switch (events[i].type) { Event events[MAX_CONNS+1];
case EVENT_CONNECT: num_events = tcp_translate_events(&tdfs->tcp, events, contexts, polled, num_polled);
break; for (int i = 0; i < num_events; i++) {
int conn_idx = events[i].conn_idx;
switch (events[i].type) {
case EVENT_CONNECT:
break;
case EVENT_DISCONNECT:
{
RequestQueue *reqs;
int tag = tcp_get_tag(&tdfs->tcp, conn_idx);
if (tag == TAG_METADATA_SERVER_TO_CLIENT)
reqs = &tdfs->metadata_server.reqs;
else {
assert(tag > -1);
reqs = &tdfs->chunk_servers[tag].reqs;
}
for (Request req; request_queue_pop(reqs, &req) == 0; )
process_event(tdfs, req.opidx, req.tag, (ByteView) { NULL, 0 });
}
break;
case EVENT_MESSAGE:
{
for (;;) {
ByteView msg;
uint16_t msg_type;
int ret = tcp_next_message(&tdfs->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
break;
if (ret < 0) {
tcp_close(&tdfs->tcp, conn_idx);
break;
}
case EVENT_DISCONNECT:
{
RequestQueue *reqs; RequestQueue *reqs;
int tag = tcp_get_tag(&tdfs->tcp, conn_idx); int tag = tcp_get_tag(&tdfs->tcp, conn_idx);
@@ -964,46 +998,32 @@ void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout)
reqs = &tdfs->chunk_servers[tag].reqs; reqs = &tdfs->chunk_servers[tag].reqs;
} }
for (Request req; request_queue_pop(reqs, &req) == 0; ) Request req;
process_event(tdfs, req.opidx, req.tag, (ByteView) { NULL, 0 }); if (request_queue_pop(reqs, &req) < 0) {
} UNREACHABLE;
break;
case EVENT_MESSAGE:
{
for (;;) {
ByteView msg;
uint16_t msg_type;
int ret = tcp_next_message(&tdfs->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
break;
if (ret < 0) {
tcp_close(&tdfs->tcp, conn_idx);
break;
}
RequestQueue *reqs;
int tag = tcp_get_tag(&tdfs->tcp, conn_idx);
if (tag == TAG_METADATA_SERVER_TO_CLIENT)
reqs = &tdfs->metadata_server.reqs;
else {
assert(tag > -1);
reqs = &tdfs->chunk_servers[tag].reqs;
}
Request req;
if (request_queue_pop(reqs, &req) < 0) {
UNREACHABLE;
}
process_event(tdfs, req.opidx, req.tag, msg);
tcp_consume_message(&tdfs->tcp, conn_idx);
} }
process_event(tdfs, req.opidx, req.tag, msg);
tcp_consume_message(&tdfs->tcp, conn_idx);
} }
break;
} }
break;
} }
} }
return tcp_register_events(&tdfs->tcp, contexts, polled);
}
void tinydfs_wait(TinyDFS *tdfs, int opidx, TinyDFS_Result *result, int timeout)
{
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled;
num_polled = tinydfs_process_events(tdfs, contexts, polled, 0);
while (!tinydfs_isdone(tdfs, opidx, result)) {
POLL(polled, num_polled, -1);
num_polled = tinydfs_process_events(tdfs, contexts, polled, num_polled);
}
} }
src/file_system.c
+4 -22
View File
@@ -1,22 +1,3 @@
#include <stdio.h>
#include <limits.h>
#include <stdlib.h>
#include <assert.h>
#ifdef __linux__
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <sys/file.h>
#include <sys/stat.h>
#endif
#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#endif
#include "system.h" #include "system.h"
#include "file_system.h" #include "file_system.h"
@@ -172,6 +153,8 @@ int file_size(Handle fd, size_t *len)
#endif #endif
} }
// TODO: port to windows
#ifndef _WIN32
// TODO: test this // TODO: test this
static string parent_path(string path) static string parent_path(string path)
{ {
@@ -189,7 +172,6 @@ static string parent_path(string path)
return path; return path;
} }
static int write_bytes(int fd, string data) static int write_bytes(int fd, string data)
{ {
size_t written = 0; size_t written = 0;
@@ -205,7 +187,6 @@ static int write_bytes(int fd, string data)
assert((size_t) data.len == written); assert((size_t) data.len == written);
return 0; return 0;
} }
int file_write_atomic(string path, string content) int file_write_atomic(string path, string content)
{ {
string parent = parent_path(path); string parent = parent_path(path);
@@ -251,6 +232,7 @@ int file_write_atomic(string path, string content)
} }
return 0; return 0;
} }
#endif
int create_dir(string path) int create_dir(string path)
{ {
@@ -261,7 +243,7 @@ int create_dir(string path)
zt[path.len] = '\0'; zt[path.len] = '\0';
#ifdef _WIN32 #ifdef _WIN32
if (sys_mkdir(zt) < 0) if (sys__mkdir(zt) < 0)
return -1; return -1;
#else #else
if (sys_mkdir(zt, 0766)) if (sys_mkdir(zt, 0766))
src/file_system.h
+13 -13
View File
@@ -7,19 +7,19 @@ typedef struct {
uint64_t data; uint64_t data;
} Handle; } Handle;
int file_open(string path, Handle *fd); int file_open(string path, Handle *fd);
void file_close(Handle fd); void file_close(Handle fd);
int file_lock(Handle fd); int file_lock(Handle fd);
int file_unlock(Handle fd); int file_unlock(Handle fd);
int file_sync(Handle fd); int file_sync(Handle fd);
int file_read(Handle fd, char *dst, int max); int file_read(Handle fd, char *dst, int max);
int file_write(Handle fd, char *src, int len); int file_write(Handle fd, char *src, int len);
int file_size(Handle fd, size_t *len); int file_size(Handle fd, size_t *len);
int file_write_atomic(string path, string content); int file_write_atomic(string path, string content);
int create_dir(string path); int create_dir(string path);
int rename_file_or_dir(string oldpath, string newpath); int rename_file_or_dir(string oldpath, string newpath);
int remove_file_or_dir(string path); int remove_file_or_dir(string path);
int get_full_path(string path, char *dst); int get_full_path(string path, char *dst);
int file_read_all(string path, string *data); int file_read_all(string path, string *data);
#endif // FILE_SYSTEM_INCLUDED #endif // FILE_SYSTEM_INCLUDED
src/main_test.c
+2 -953
View File
@@ -1,297 +1,7 @@
#ifdef BUILD_TEST #ifdef BUILD_TEST
#include "system.h"
#include <signal.h> static sig_atomic_t simulation_should_stop = false;
#include <stdio.h>
#include <string.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include "chunk_server.h"
#include "metadata_server.h"
#define MAX_PROCESSES 128
#define MAX_DESCRIPTORS 1024
#define MAX_ALLOCATIONS 128
typedef enum {
DESCRIPTOR_TYPE_EMPTY,
DESCRIPTOR_TYPE_FILE,
DESCRIPTOR_TYPE_SOCKET,
DESCRIPTOR_TYPE_LISTENER_SOCKET,
DESCRIPTOR_TYPE_CONNECTION_SOCKET,
} DescriptorType;
typedef struct {
int peer_process;
int peer_fd;
} PendingAccept;
typedef struct {
// Common
DescriptorType type;
// File
int real_fd;
// Socket
bool bound;
struct sockaddr_in addr;
// Listener Socket
PendingAccept *accept_queue;
int accept_queue_head;
int accept_queue_size;
int accept_queue_used;
// Connection Socket
bool pending;
int peer_process;
int peer_fd;
char *input;
int input_used;
int input_size;
char *output;
int output_used;
int output_size;
} Descriptor;
typedef struct {
int num_desc;
Descriptor desc[MAX_DESCRIPTORS];
} DescriptorTable;
typedef struct {
void *ptr;
size_t len;
char *file;
int line;
} Allocation;
typedef struct {
size_t mem_usage;
int num_allocs;
Allocation allocs[MAX_ALLOCATIONS];
} AllocationTable;
typedef enum {
PROCESS_TYPE_CLIENT,
PROCESS_TYPE_CHUNK_SERVER,
PROCESS_TYPE_METADATA_SERVER,
} ProcessType;
typedef struct {
ProcessType type;
union {
ChunkServer chunk_server;
MetadataServer metadata_server;
};
DescriptorTable *dt;
AllocationTable *at;
} Process;
int num_processes = 0;
Process processes[MAX_PROCESSES];
Process *current_process = NULL;
uint64_t current_time = 0;
int find_unused_descriptor(Process *process)
{
if (process->dt->num_desc == MAX_DESCRIPTORS)
return -1;
int i = 0;
while (process->dt->desc[i].type != DESCRIPTOR_TYPE_EMPTY)
i++;
return i;
}
void descriptor_init_as_file(Descriptor *desc, int real_fd)
{
desc->type = DESCRIPTOR_TYPE_FILE;
desc->real_fd = real_fd;
}
void descriptor_init_as_socket(Descriptor *desc)
{
desc->type = DESCRIPTOR_TYPE_SOCKET;
desc->bound = false;
}
void descriptor_turn_socket_into_listener(Descriptor *desc, int backlog)
{
desc->type = DESCRIPTOR_TYPE_LISTENER_SOCKET;
desc->accept_queue = malloc(backlog * sizeof(PendingAccept));
desc->accept_queue_head = 0;
desc->accept_queue_used = 0;
desc->accept_queue_size = backlog;
if (desc->accept_queue == NULL)
__builtin_trap();
}
void descriptor_turn_socket_into_connection(Descriptor *desc, int peer_process, int peer_fd)
{
desc->type = DESCRIPTOR_TYPE_CONNECTION_SOCKET;
desc->pending = true;
desc->peer_process = peer_process;
desc->peer_fd = peer_fd;
desc->input = malloc(1<<9);
desc->input_size = 1<<9;
desc->input_used = 0;
desc->output = malloc(1<<9);
desc->output_size = 1<<9;
desc->output_used = 0;
if (desc->input == NULL || desc->output == NULL)
__builtin_trap();
}
void descriptor_free(Descriptor *desc)
{
switch (desc->type) {
case DESCRIPTOR_TYPE_EMPTY:
// TODO
break;
case DESCRIPTOR_TYPE_FILE:
close(desc->real_fd);
break;
case DESCRIPTOR_TYPE_SOCKET:
break;
case DESCRIPTOR_TYPE_LISTENER_SOCKET:
free(desc->accept_queue);
break;
case DESCRIPTOR_TYPE_CONNECTION_SOCKET:
free(desc->input);
free(desc->output);
break;
}
desc->type = DESCRIPTOR_TYPE_EMPTY;
}
bool is_leader(int argc, char **argv)
{
for (int i = 1; i < argc; i++)
if (!strcmp(argv[i], "--leader") || !strcmp(argv[i], "-l"))
return true;
return false;
}
int split_args(char *args, char **out, int max)
{
int len = strlen(args);
int cur = 0;
int num = 0;
for (;;) {
while (cur < len && (args[cur] == ' ' || args[cur] == '\t'))
cur++;
int off = cur;
while (cur < len && (args[cur] != ' ' && args[cur] != '\t'))
cur++;
if (num < max)
out[num++] = args + off;
args[cur++] = '\0';
}
return num;
}
#define MAX_ARGS 128
void process_init(Process *process, char *args)
{
process->dt = malloc(sizeof(DescriptorTable));
process->at = malloc(sizeof(AllocationTable));
if (process->dt == NULL || process->at == NULL)
abort();
process->dt->num_desc = 0;
process->at->num_allocs = 0;
process->at->mem_usage = 0;
char *argv[MAX_ARGS];
int argc = split_args(args, argv, MAX_ARGS);
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled;
current_process = process;
if (is_leader(argc, argv)) {
process->type = PROCESS_TYPE_METADATA_SERVER;
num_polled = metadata_server_init(&process->metadata_server, argc, argv, contexts, polled);
} else {
process->type = PROCESS_TYPE_CHUNK_SERVER;
num_polled = chunk_server_init(&process->chunk_server, argc, argv, contexts, polled);
}
current_process = NULL;
if (num_polled < 0) {
// TODO
}
do_stuff_with_pollfds(contexts, polled, num_polled);
}
void process_step(Process *process)
{
if (process->type == PROCESS_TYPE_METADATA_SERVER)
num_polled = metadata_server_step(&process->metadata_server, xxx, yyyy, contexts, polled, num_polled);
else
num_polled = chunk_server_step(&process->chunk_server, xxx, yyyy, contexts, polled, num_polled);
if (num_polled < 0) {
// TODO
}
do_stuff_with_pollfds(contexts, polled, num_polled);
}
void process_free(Process *process)
{
if (process->type == PROCESS_TYPE_METADATA_SERVER)
metadata_server_free(&process->metadata_server);
else
chunk_server_free(&process->chunk_server);
free(process->at);
free(process->dt);
}
void spawn_simulated_process(char *args)
{
if (num_processes == MAX_PROCESSES)
abort();
process_init(&processes[num_processes++], args);
}
void update_simulation(void)
{
for (int i = 0; i < num_processes; i++)
process_step(&processes[i]);
// TODO
}
void cleanup_simulation(void)
{
for (int i = 0; i < num_processes; i++)
process_free(&processes[i]);
}
sig_atomic_t simulation_should_stop = false;
int main(int argc, char **argv) int main(int argc, char **argv)
{ {
@@ -316,665 +26,4 @@ int main(int argc, char **argv)
return 0; return 0;
} }
void *sys_malloc_(size_t len, char *file, int line)
{
if (current_process->at->num_allocs == MAX_ALLOCATIONS)
__builtin_trap();
void *ptr = malloc(len);
if (ptr == NULL)
__builtin_trap();
current_process->at->allocs[current_process->at->num_allocs++] = (Allocation) { ptr, len, file, line };
current_process->at->mem_usage += len;
return ptr;
}
void *sys_realloc_(void *ptr, size_t len, char *file, int line)
{
int found = -1;
for (int i = 0; i < current_process->at->num_allocs; i++)
if (current_process->at->allocs[i].ptr == ptr) {
found = i;
break;
}
if (found < 0)
__builtin_trap();
size_t old_len = current_process->at->allocs[found].len;
void *new_ptr = realloc(ptr, len);
if (new_ptr == NULL)
__builtin_trap();
current_process->at->allocs[found].ptr = new_ptr;
current_process->at->allocs[found].len = len;
current_process->at->allocs[found].file = file;
current_process->at->allocs[found].line = line;
current_process->at->mem_usage -= old_len;
current_process->at->mem_usage += len;
return new_ptr;
}
void sys_free_(void *ptr, char *file, int line)
{
(void) file;
(void) line;
int found = -1;
for (int i = 0; i < current_process->at->num_allocs; i++)
if (current_process->at->allocs[i].ptr == ptr) {
found = i;
break;
}
if (found < 0)
__builtin_trap();
current_process->at->mem_usage -= current_process->at->allocs[found].len;
current_process->at->allocs[found] = current_process->at->allocs[--current_process->num_allocs];
}
int sys_remove(char *path)
{
return remove(path);
}
int sys_rename(char *oldpath, char *newpath)
{
return rename(oldpath, newpath);
}
#ifdef _WIN32
SOCKET sys_socket(int domain, int type, int protocol)
{
if (domain != AF_INET) {
// TODO: errno
return INVALID_SOCKET;
}
if (type != SOCK_STREAM) {
// TODO: errno
return INVALID_SOCKET;
}
if (protocol != 0) {
// TODO: errno
return INVALID_SOCKET;
}
int fd = find_unused_descriptor(current_process);
if (fd < 0) {
// TODO: errno
return INVALID_SOCKET;
}
// TODO: maybe bind to a random address?
descriptor_init_as_socket(&current_process->dt->desc[fd]);
return (SOCKET) fd;
}
int sys_bind(SOCKET fd, void *addr, size_t addr_len)
{
Descriptor *desc = &current_process->dt->desc[(int) fd];
if (desc->type != DESCRIPTOR_TYPE_PRECONF_SOCKET) {
// TODO: errno
return -1;
}
if (addr_len != sizeof(desc->bind)) {
// TODO: errno
return -1;
}
// TODO: maybe check that no one else is listening
// on this port
desc->bound = true;
memcpy(&desc->addr, addr, addr_len);
return 0;
}
int sys_listen(SOCKET fd, int backlog)
{
Descriptor *desc = &current_process->dt->desc[(int) fd];
if (desc->type != DESCRIPTOR_TYPE_SOCKET) {
// TODO: errno
return -1;
}
descriptor_turn_socket_into_listener(desc, backlog);
return 0;
}
int sys_closesocket(SOCKET fd)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_SOCKET &&
desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET
desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) {
__builtin_trap();
}
descriptor_free(desc);
return 0;
}
SOCKET sys_accept(SOCKET fd, void *addr, int *addr_len)
{
Descriptor *desc = &current_process->desc[(int) fd];
if (desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET) {
// TODO: errno
return INVALID_SOCKET;
}
if (desc->accept_queue_used == 0) {
// TODO: would block
}
PendingAccept *pending_accept = desc->accept_queue[desc->accept_queue_head];
desc->accept_queue_head = (desc->accept_queue_head + 1) % (desc->accept_queue_size);
desc->accept_queue_used--;
int new_fd = find_unused_descriptor(current_process);
if (new_fd < 0) {
// TODO
}
descriptor_init_as_socket(&current_process->desc[new_fd]);
descriptor_turn_socket_into_connection(&current_process->desc[new_fd], peer_process, peer_fd);
return (SOCKET) new_fd;
}
int sys_getsockopt(SOCKET fd, int level, int optname, void *optval, int *optlen)
{
// TODO
}
int sys_setsockopt(SOCKET fd, int level, int optname, void *optval, int optlen)
{
// TODO
}
int sys_recv(SOCKET fd, void *dst, int len, int flags)
{
// TODO
}
int sys_send(SOCKET fd, void *src, int len, int flags)
{
// TODO
}
int sys_connect(SOCKET fd, void *addr, size_t addr_len)
{
// TODO
}
BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount)
{
// TODO
}
BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency)
{
// TODO
}
HANDLE sys_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess,
DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes,
DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes,
HANDLE hTemplateFile)
{
int fd = find_unused_descriptor(current_process);
if (fd < 0) {
// TODO: WSAGetLastError
return INVALID_HANDLE_VALUE;
}
HANDLE handle = CreateFileW(lpFileName, dwDesiredAccess,
dwShareMode, lpSecurityAttributes, dwCreationDisposition,
dwFlagsAndAttributes, hTemplateFile);
if (handle == INVALID_HANDLE_VALUE) {
// TODO
return INVALID_HANDLE_VALUE;
}
Descriptor *desc = current_process->dt->desc[fd];
descriptor_init_as_file(desc, handle);
return (HANDLE) fd;
}
BOOL sys_CloseHandle(HANDLE handle)
{
Descriptor *desc = current_process->dt->desc[(int) handle];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO
return false;
}
return CloseHandle(desc->real_fd);
}
BOOL sys_LockFile(HANDLE handle)
{
Descriptor *desc = current_process->dt->desc[(int) handle];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO
return false;
}
return LockFile(desc->real_fd);
}
BOOL sys_UnlockFile(HANDLE handle)
{
Descriptor *desc = current_process->dt->desc[(int) handle];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO
return false;
}
return UnlockFile(desc->real_fd);
}
BOOL sys_FlushFileBuffers(HANDLE handle)
{
Descriptor *desc = current_process->dt->desc[(int) handle];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO
return false;
}
return FlushFileBuffers(desc->real_fd);
}
BOOL sys_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov)
{
if (ov) {
// TODO
return false;
}
Descriptor *desc = &current_process->dt->desc[(int) handle];
switch (desc->type) {
case DESCRIPTOR_TYPE_FILE:
return ReadFile(desc->real_fd, dst, len, num, ov);
case DESCRIPTOR_TYPE_CONNECTION_SOCKET:
{
int cpy = len;
if (cpy > desc->input_used)
cpy = desc->input_used;
memcpy(dst, desc->input, cpy);
memmove(desc->input, desc->input + cpy, desc->input_used - cpy);
desc->input_used -= cpy;
*num = cpy;
return true;
}
break;
default:
// TODO: errno
return false;
}
}
BOOL sys_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov)
{
if (ov) {
// TODO
return false;
}
Descriptor *desc = &current_process->dt->desc[(int) handle];
switch (desc->type) {
case DESCRIPTOR_TYPE_FILE:
return WriteFile(desc->real_fd, src, len, num, ov);
case DESCRIPTOR_TYPE_CONNECTION_SOCKET:
{
if (desc->output_size - desc->output_used < len) {
int new_capacity = 2 * desc->output_size;
if (new_capacity - desc->output_used < len)
new_capacity = desc->output_used + len;
desc->output = realloc(desc->output, new_capacity);
if (desc->output == NULL)
__builtin_trap();
desc->output_size = new_capacity;
}
memcpy(desc->output + desc->output_used, src, len);
desc->output_used += len;
*num = len;
return true;
}
break;
default:
// TODO: errno
return false;
}
}
BOOL sys_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf)
{
Descriptor *desc = current_process->dt->desc[(int) handle];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO
return false;
}
return GetFileSizeEx(desc->real_fd, buf);
}
char *sys__fullpath(char *path, char *dst, int cap)
{
return _fullpath(path, dst, cap);
}
#else
int sys_socket(int domain, int type, int protocol)
{
if (domain != AF_INET) {
// TODO: errno
return -1;
}
if (type != SOCK_STREAM) {
// TODO: errno
return -1;
}
if (protocol != 0) {
// TODO: errno
return -1;
}
int fd = find_unused_descriptor(current_process);
if (fd < 0) {
// TODO: errno
return -1;
}
// TODO: maybe bind to a random address?
descriptor_init_as_socket(&current_process->dt->desc[fd]);
return fd;
}
int sys_bind(int fd, void *addr, size_t addr_len)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_PRECONF_SOCKET) {
// TODO: errno
return -1;
}
if (addr_len != sizeof(desc->bind)) {
// TODO: errno
return -1;
}
// TODO: maybe check that no one else is listening
// on this port
desc->bound = true;
memcpy(&desc->addr, addr, addr_len);
return 0;
}
int sys_listen(int fd, int backlog)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_SOCKET) {
// TODO: errno
return -1;
}
descriptor_turn_socket_into_listener(desc, backlog);
return 0;
}
int sys_accept(int fd, void *addr, int *addr_len)
{
Descriptor *desc = &current_process->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_LISTENER_SOCKET) {
// TODO: errno
return -1;
}
if (desc->accept_queue_used == 0) {
// TODO: would block
}
PendingAccept *pending_accept = desc->accept_queue[desc->accept_queue_head];
desc->accept_queue_head = (desc->accept_queue_head + 1) % (desc->accept_queue_size);
desc->accept_queue_used--;
int new_fd = find_unused_descriptor(current_process);
if (new_fd < 0) {
// TODO
}
descriptor_init_as_socket(&current_process->desc[new_fd]);
descriptor_turn_socket_into_connection(&current_process->desc[new_fd], peer_process, peer_fd);
return new_fd;
}
int sys_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen)
{
// TODO
}
int sys_setsockopt(int fd, int level, int optname, void *optval, socklen_t optlen)
{
// TODO
}
int sys_recv(int fd, void *dst, int len, int flags)
{
if (flags)
__builtin_trap();
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) {
// TODO: errno
return -1;
}
return sys_read(fd, dst, len);
}
int sys_send(int fd, void *src, int len, int flags)
{
if (flags)
__builtin_trap();
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET) {
// TODO: errno
return -1;
}
return sys_write(fd, src, len);
}
bool sockaddr_eql(struct sockaddr_in a, struct sockaddr_in b)
{
return a.sin_family == b.sin_family
&& a.sin_port == b.sin_port
&& a.sin_addr == b.sin_addr;
}
int sys_connect(int fd, void *addr, size_t addr_len)
{
if (addr_len != sizeof(struct sockaddr_in)) {
// TODO
}
struct sockaddr_in tmp;
memcpy(&tmp, addr, sizeof(tmp));
int peer_process = -1;
int peer_fd = -1;
for (int i = 0; i < num_processes; i++) {
for (int j = 0; j < processes[i].dt->num_desc; j++) {
Descriptor *desc = &processes[i].dt->desc[j];
if (desc->type == DESCRIPTOR_TYPE_LISTENER_SOCKET
&& sockaddr_eql(desc->addr, tmp)) {
peer_process = i;
peer_fd = j;
goto found;
}
}
}
found:
if (peer_process < 0) {
// TODO
}
int fd = find_unused_descriptor(current_process);
if (fd < 0) {
// TODO
}
Descriptor *desc = &current_process->desc[fd];
descriptor_init_as_socket(desc);
descriptor_turn_socket_into_connection(desc, peer_process, peer_fd);
return fd;
}
int sys_clock_gettime(clockid_t clockid, struct timespec *tp)
{
if (clockid != CLOCK_REALTIME) {
// TODO
}
if (tp == NULL) {
// TODO
}
tp->tv_sec = current_time / 1000000000;
tp->tv_nsec = current_time % 1000000000;
return 0;
}
int sys_open(char *path, int flags, int mode)
{
int fd = find_unused_descriptor(current_process);
if (fd < 0) {
// TODO: errno
return -1;
}
int real_fd = open(path, flags, mode);
if (real_fd < 0)
return real_fd;
descriptor_init_as_file(&current_process->dt->desc[fd], real_fd);
return fd;
}
int sys_close(int fd)
{
Descriptor *desc = &current_process->dt->desc[fd];
descriptor_free(desc);
return 0;
}
int sys_flock(int fd, int op)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO: errno
return -1;
}
return flock(desc->real_fd, op);
}
int sys_fsync(int fd)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO: errno
return -1;
}
return fsync(desc->real_fd);
}
int sys_read(int fd, char *dst, int len)
{
Descriptor *desc = &current_process->dt->desc[fd];
switch (desc->type) {
case DESCRIPTOR_TYPE_FILE:
return read(desc->real_fd, dst, len);
case DESCRIPTOR_TYPE_CONNECTION_SOCKET:
{
int cpy = len;
if (cpy > desc->input_used)
cpy = desc->input_used;
memcpy(dst, desc->input, cpy);
memmove(desc->input, desc->input + cpy, desc->input_used - cpy);
desc->input_used -= cpy;
return cpy;
}
break;
default:
// TODO: errno
return -1;
}
}
int sys_write(int fd, char *src, int len)
{
Descriptor *desc = &current_process->dt->desc[fd];
switch (desc->type) {
case DESCRIPTOR_TYPE_FILE:
return write(desc->real_fd, dst, len);
case DESCRIPTOR_TYPE_CONNECTION_SOCKET:
{
if (desc->output_size - desc->output_used < len) {
int new_capacity = 2 * desc->output_size;
if (new_capacity - desc->output_used < len)
new_capacity = desc->output_used + len;
desc->output = realloc(desc->output, new_capacity);
if (desc->output == NULL)
__builtin_trap();
desc->output_size = new_capacity;
}
memcpy(desc->output + desc->output_used, src, len);
desc->output_used += len;
return len;
}
break;
default:
// TODO: errno
return -1;
}
}
int sys_fstat(int fd, struct stat *buf)
{
Descriptor *desc = &current_process->dt->desc[fd];
if (desc->type != DESCRIPTOR_TYPE_FILE) {
// TODO: errno
return -1;
}
return stat(desc->real_fd, buf);
}
int sys_mkstemp(char *path)
{
return mkstemp(path);
}
char* sys_realpath(char *path, char *dst)
{
return realpath(path, dst);
}
int sys_mkdir(char *path, mode_t mode)
{
return mkdir(path, mode);
}
#endif #endif
#endif // BUILD_TEST
src/system.c
+707 -181
View File
@@ -1,246 +1,772 @@
#include <stdio.h> #include <assert.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/file.h>
#include "system.h" #include "system.h"
#include "chunk_server.h"
#include "metadata_server.h"
void *sys_malloc_(size_t len, char *file, int line) #ifdef _WIN32
#define NATIVE_HANDLE HANDLE
#else
#define SOCKET int
#define INVALID_SOCKET -1
#define NATIVE_HANDLE int
#endif
#define MAX_DESCRIPTORS 1024
#define MAX_ALLOCATIONS 128
#define MAX_PROCESSES 32
#define DATA_QUEUE_SIZE (1<<9)
typedef struct Process Process;
typedef enum {
DESC_EMPTY,
DESC_SOCKET,
DESC_LISTEN_SOCKET,
DESC_CONNECTION_SOCKET,
} DescriptorType;
typedef enum {
DESC_ADDR_VOID,
DESC_ADDR_IPV4,
DESC_ADDR_IPV6,
} DescriptorAddressType;
typedef struct {
DescriptorAddressType type;
union {
struct sockaddr_in ipv4;
struct sockaddr_in6 ipv6;
};
} DescriptorAddress;
typedef struct {
Process *process;
int descriptor_index;
uint32_t generation;
} DescriptorHandle;
typedef struct {
int head;
int used;
int size;
DescriptorHandle *items;
} AcceptQueue;
typedef struct {
int size;
int used;
char *data;
} DataQueue;
typedef enum {
CONNECTION_DELAYED,
CONNECTION_QUEUED,
CONNECTION_ESTABLISHED,
CONNECTION_FAILED,
} ConnectionState;
typedef struct {
// ------ Common ----------------
DescriptorType type;
uint32_t generation;
// ------ File ------------------
NATIVE_HANDLE real_fd;
// ------ Socket ----------------
// Events reported by the last "poll" call
// for this descriptor
int events;
// Events triggered since the last "poll"
// call. Note that these may include events
// not present in the "events" set.
int revents;
// Context for this descriptor, set by the
// last "poll" call.
void *context;
// Address bound to this descriptor by the
// "bind" call.
DescriptorAddress address;
// ------ Listen socket ---------
AcceptQueue accept_queue;
// ------ Connection socket -----
ConnectionState connection_state;
// When QUEUED, this refers to the peer listener
// socket. When ESTABLISHED, this refers to the
// peer connection socket.
DescriptorHandle connection_peer;
// Address of the last connect() call
// on this socket if it's still in the
// "DELAYED" state.
DescriptorAddress connect_address;
// Data written to this descriptor using "write"
// or "send".
DataQueue output_data;
// ------------------------------
} Descriptor;
typedef struct {
void *ptr;
size_t len;
char *file;
int line;
} Allocation;
struct Process {
int num_desc;
Descriptor desc[MAX_DESCRIPTORS];
int num_allocs;
Allocation allocs[MAX_ALLOCATIONS];
bool leader;
union {
ChunkServer chunk_server;
MetadataServer metadata_server;
};
};
static int num_processes = 0;
static Process *processes[MAX_PROCESSES];
static Process *current_process = NULL;
static bool is_leader(int argc, char **argv)
{ {
(void) file; for (int i = 0; i < argc; i++)
(void) line; if (!strcmp("--leader", argv[i]) || !strcmp("-l", argv[i]))
return malloc(len); return true;
return false;
} }
void *sys_realloc_(void *ptr, size_t len, char *file, int line) int spawn_simulated_process(char *args)
{ {
(void) file; if (num_processes == MAX_PROCESSES)
(void) line; return -1;
return realloc(ptr, len);
bool leader = is_leader(argc, argv);
Process *process = malloc(sizeof(Process));
if (process == NULL)
return -1;
process->leader = leader;
process->num_desc = 0;
process->num_allocs = 0;
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled;
if (leader) {
num_polled = metadata_server_init(&process->metadata_server, argc, argv, contexts, polled);
} else {
num_polled = chunk_server_init(&process->chunk_server, argc, argv, contexts, polled);
}
if (num_polled < 0) {
// TODO
}
process_poll_array(process, contexts, polled, num_polled);
processes[num_processes++] = process;
} }
void sys_free_(void *ptr, char *file, int line) static void free_process(Process *process)
{ {
(void) file; if (leader) {
(void) line; metadata_server_free(&process->metadata_server);
free(ptr); } else {
chunk_server_free(&process->chunk_server);
}
free(process);
} }
int sys_remove(char *path) void cleanup_simulation(void)
{ {
return remove(path); for (int i = 0; i < num_processes; i++)
free_process(processes[i]);
} }
int sys_rename(char *oldpath, char *newpath) static bool addr_eql_2(DescriptorAddress a, DescriptorAddress b)
{ {
return rename(oldpath, newpath); if (a.type != b.type)
return false;
if (a.type == DESC_ADDR_IPV4) {
return a.ipv4.sin_family == b.ipv4.sin_family
&& a.ipv4.sin_port == b.ipv4.sin_port
&& !memcmp(&a.ipv4.sin_addr, &a.ipv4.sin_addr, sizeof(a.ipv4.sin_addr));
} else {
return a.ipv6.sin6_family == b.ipv6.sin6_family
&& a.ipv6.sin6_port == b.ipv6.sin6_port
&& !memcmp(&a.ipv6.sin6_addr, &a.ipv6.sin6_addr, sizeof(a.ipv6.sin6_addr));
}
}
static bool find_peer_by_address(DescriptorAddress address, DescriptorHandle *handle)
{
for (int i = 0; i < num_processes; i++) {
for (int j = 0, k = 0; k < processes[i]->num_desc; j++) {
Descriptor *desc = &processes[i]->desc[j];
if (desc->type == DESC_EMPTY)
continue;
k++;
if (desc->type == DESC_LISTEN_SOCKET &&
addr_eql_2(address, desc->address)) {
*handle = (DescriptorHandle) { processes[i], j, desc->generation };
return true;
}
}
}
return false;
}
static Descriptor *handle_to_desc(DescriptorHandle handle)
{
if (handle.process == NULL
|| handle.descriptor_index < 0
|| handle.descriptor_index >= MAX_DESCRIPTORS)
return NULL;
Process *process = processes[handle.process_index];
Descriptor *desc = &process->desc[handle.descriptor_index];
if (desc->type == DESC_EMPTY || desc->generation != handle.generation)
return NULL;
return desc;
}
static void accept_queue_init(AcceptQueue *accept_queue, int size)
{
accept_queue->head = 0;
accept_queue->used = 0;
accept_queue->size = size;
accept_queue->items = malloc(size * sizeof(DescriptorHandle));
if (accept_queue->items == NULL) {
// TODO
}
}
static void accept_queue_free(AcceptQueue *accept_queue)
{
free(accept_queue->items);
}
static bool accept_queue_push(AcceptQueue *accept_queue, DescriptorHandle handle)
{
if (accept_queue->used == accept_queue->size)
return false;
int tail = (accept_queue->head + accept_queue->used) % accept_queue->size;
accept_queue->items[tail] = handle;
accept_queue->used++;
return true;
}
static bool accept_queue_pop(AcceptQueue *accept_queue, DescriptorHandle *item)
{
if (accept_queue->used == 0)
return false;
*item = accept_queue->items[accept_queue->head];
accept_queue->head = (accept_queue->head + 1) % accept_queue->size;
accept_queue->used--;
return true;
}
static void data_queue_init(DataQueue *queue, int size)
{
queue->used = 0;
queue->size = size;
queue->data = malloc(size * sizeof(char));
if (queue->data == NULL) {
// TODO
}
}
static void data_queue_free(DataQueue *queue)
{
free(queue->data);
}
static int data_queue_read(DataQueue *queue, char *dst, int max)
{
int num = max;
if (num > queue->used)
num = queue->used;
if (num > 0) {
memcpy(dst, queue->data, num);
memmove(queue->data, queue->data + num, queue->used - num);
queue->used -= num;
}
return num;
}
static int data_queue_write(DataQueue *queue, char *src, int len)
{
int num = len;
if (num > queue->size - queue->used)
num = queue->size - queue->used;
memcpy(queue->data + queue->used, src, num);
queue->used += num;
return num;
}
void update_simulation(void)
{
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;
// TODO: fill up poll array
if (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);
}
if (num_polled < 0) {
// TODO
}
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++) {
Descriptor *desc = &processes[i]->desc[j];
if (desc->type == DESC_EMPTY)
continue;
k++;
if (desc->type != DESC_CONNECTION_SOCKET)
continue;
switch (desc->connection_state) {
case CONNECTION_DELAYED:
{
DescriptorHandle peer_handle;
if (!find_peer_by_address(desc->connect_address, &peer_handle)) {
desc->connection_state = CONNECTION_FAILED;
break;
}
DescriptorHandle self_handle = { processes[i], j, desc->generation };
Descriptor *peer = handle_to_desc(peer_handle);
if (!accept_queue_push(&peer->accept_queue, self_handle)) {
desc->connection_state = CONNECTION_FAILED;
break;
}
desc->connection_state = CONNECTION_QUEUED;
desc->connection_peer = peer_handle;
}
break;
case CONNECTION_QUEUED:
{
if (handle_to_desc(desc->connection_peer) == NULL) {
// Listener closed before accepting
desc->connection_state = CONNECTION_FAILED;
break;
}
}
break;
default:
break;
}
}
}
}
SOCKET mock_socket(int domain, int type, int protocol)
{
if (domain != AF_INET || type != SOCK_STREAM || protocol != 0) {
// TODO: errno
return INVALID_SOCKET;
}
if (current_process->num_desc == MAX_DESCRIPTORS) {
// TODO: errno
return INVALID_SOCKET;
}
int idx = 0;
while (current_process->desc[idx].type != DESC_EMPTY)
idx++;
Descriptor *desc = &current_process->desc[idx];
desc->type = DESC_SOCKET;
desc->events = 0;
desc->revents = 0;
desc->context = NULL;
desc->address = (DescriptorAddress) { .type=DESC_ADDR_VOID };
return (SOCKET) idx;
}
static DescriptorAddress convert_address(void *addr, size_t addr_len)
{
int family = ((struct sockaddr*) addr)->sa_family;
if (family == AF_INET && addr_len == sizeof(struct sockaddr_in))
return (DescriptorAddress) { .type=DESC_ADDR_IPV4, .ipv4=*(struct sockaddr_in*) addr };
if (family == AF_INET6 && addr_len != sizeof(struct sockaddr_in6))
return (DescriptorAddress) { .type=DESC_ADDR_IPV6, .ipv6=*(struct sockaddr_in6*) addr };
return (DescriptorAddress) { .type=DESC_ADDR_VOID };
}
int mock_bind(SOCKET fd, void *addr, size_t addr_len)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_SOCKET) {
// TODO
return -1;
}
DescriptorAddress address = convert_address(addr, addr_len);
if (address.type == DESC_ADDR_VOID) {
// TODO
return -1;
}
// TODO: Should check that the address family matched
// the one used to create the socket
desc->address = address;
return 0;
}
int mock_listen(SOCKET fd, int backlog)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_SOCKET) {
// TODO
return -1;
}
desc->type = DESC_LISTEN_SOCKET;
accept_queue_init(&desc->accept_queue, backlog);
return 0;
}
SOCKET mock_accept(SOCKET fd, void *addr, socklen_t *addr_len)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_LISTEN_SOCKET) {
// TODO
return -1;
}
DescriptorHandle peer_handle;
if (!accept_queue_pop(&desc->accept_queue, &peer_handle)) {
// TODO
assert(0);
}
Descriptor *peer = handle_to_desc(peer_handle);
if (peer == NULL) {
// Peer closed without removing itself from the accept queue!
assert(0);
}
if (current_process->num_desc == MAX_DESCRIPTORS) {
// TODO
}
int new_idx = 0;
while (current_process->desc[new_idx].type != DESC_EMPTY)
new_idx++;
Descriptor *new_desc = &current_process->desc[new_idx];
new_desc->type = DESC_CONNECTION_SOCKET;
new_desc->events = 0;
new_desc->revents = 0;
new_desc->context = NULL;
new_desc->address = (DescriptorAddress) { .type=DESC_ADDR_VOID };
new_desc->connection_state = CONNECTION_ESTABLISHED;
new_desc->connection_peer = peer_handle;
data_queue_init(&new_desc->output_data, DATA_QUEUE_SIZE);
peer->connection_peer = (DescriptorHandle) { current_process, new_idx, new_desc->generation };
peer->connection_state = CONNECTION_ESTABLISHED;
peer->revents |= POLLOUT;
data_queue_init(&peer->output_data, DATA_QUEUE_SIZE);
return (SOCKET) new_idx;
}
int mock_getsockopt(SOCKET fd, int level, int optname, void *optval, socklen_t *optlen)
{
// TODO
return -1;
}
int mock_setsockopt(SOCKET fd, int level, int optname, void *optval, socklen_t optlen)
{
// TODO
return -1;
}
int mock_recv(SOCKET fd, void *dst, int len, int flags)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
if (flags) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_CONNECTION_SOCKET) {
// TODO
return -1;
}
if (desc->connection_state != CONNECTION_ESTABLISHED) {
// TODO
return -1;
}
Descriptor *peer = handle_to_desc(desc->connection_peer);
if (peer == NULL) {
// TODO
return -1;
}
DataQueue *input_data = &peer->output_data;
return data_queue_read(input_data, dst, len);
}
int mock_send(SOCKET fd, void *src, int len, int flags)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
if (flags) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_CONNECTION_SOCKET) {
// TODO
return -1;
}
if (desc->connection_state != CONNECTION_ESTABLISHED) {
// TODO
return -1;
}
data_queue_write(&desc->output_data, src, len);
return len;
}
int mock_connect(SOCKET fd, void *addr, size_t addr_len)
{
if (fd == INVALID_SOCKET) {
// TODO
return -1;
}
int idx = (int) fd;
Descriptor *desc = &current_process->desc[idx];
if (desc->type != DESC_SOCKET) {
// TODO
return -1;
}
desc->type = DESC_CONNECTION_SOCKET;
desc->connection_state = CONNECTION_DELAYED;
desc->connect_address = convert_address(addr, addr_len);
if (desc->connect_address.type == DESC_ADDR_VOID) {
// TODO
return -1;
}
return -1;
} }
#ifdef _WIN32 #ifdef _WIN32
SOCKET sys_socket(int domain, int type, int protocol) int mock_closesocket(SOCKET fd)
{
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)
{
return closesocket(fd);
}
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, int *optlen)
{
return getsockopt(fd, level, optname, optval);
}
int sys_setsockopt(SOCKET fd, int level, int optname, void *optval, int 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);
}
BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount)
{
return QueryPerformanceCounter(lpPerformanceCount);
}
BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency)
{
return QueryPerformanceFrequency(lpFrequency);
}
HANDLE sys_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, HANDLE hTemplateFile)
{
return CreateFileW(lpFileName, dwDesiredAccess, dwShareMode, lpSecurityAttributes, dwCreationDisposition, dwFlagsAndAttributes, hTemplateFile);
}
BOOL sys_CloseHandle(HANDLE handle)
{
return CloseHandle(handle);
}
BOOL sys_LockFile(HANDLE handle)
{
return LockFile(handle);
}
BOOL sys_UnlockFile(HANDLE handle)
{
return UnlockFile(handle);
}
BOOL sys_FlushFileBuffers(HANDLE handle)
{
return FlushFileBuffers(handle);
}
BOOL sys_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov)
{
return ReadFile(handle, dst, len, num, ov);
}
BOOL sys_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov)
{
return WriteFile(handle, src, len, num, ov);
}
BOOL sys_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf)
{ {
// TODO // TODO
} }
char *sys__fullpath(char *path, char *dst, int cap) HANDLE mock_CreateFileW(WCHAR *lpFileName, DWORD dwDesiredAccess,
DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes,
DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes,
HANDLE hTemplateFile)
{
// TODO
}
BOOL mock_CloseHandle(HANDLE handle)
{
// TODO
}
BOOL mock_LockFile(HANDLE hFile,
DWORD dwFileOffsetLow,
DWORD dwFileOffsetHigh,
DWORD nNumberOfBytesToLockLow,
DWORD nNumberOfBytesToLockHigh)
{
// TODO
}
BOOL mock_UnlockFile(
HANDLE hFile,
DWORD dwFileOffsetLow,
DWORD dwFileOffsetHigh,
DWORD nNumberOfBytesToUnlockLow,
DWORD nNumberOfBytesToUnlockHigh)
{
// TODO
}
BOOL mock_FlushFileBuffers(HANDLE handle)
{
// TODO
}
BOOL mock_ReadFile(HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov)
{
// TODO
}
BOOL mock_WriteFile(HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov)
{
// TODO
}
BOOL mock_GetFileSizeEx(HANDLE handle, LARGE_INTEGER *buf)
{
// TODO
}
char *mock__fullpath(char *path, char *dst, int cap)
{ {
return _fullpath(path, dst, cap); return _fullpath(path, dst, cap);
} }
int mock__mkdir(char *path)
{
return _mkdir(path);
}
#else #else
int sys_socket(int domain, int type, int protocol) int mock_clock_gettime(clockid_t clockid, struct timespec *tp)
{ {
return socket(domain, type, protocol); // TODO
} }
int sys_bind(int fd, void *addr, size_t addr_len) int mock_open(char *path, int flags, int mode)
{ {
return bind(fd, addr, addr_len); // TODO
} }
int sys_listen(int fd, int backlog) int mock_close(int fd)
{ {
return listen(fd, backlog); // TODO
} }
int sys_accept(int fd, void *addr, socklen_t *addr_len) int mock_flock(int fd, int op)
{ {
return accept(fd, addr, addr_len); // TODO
} }
int sys_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen) int mock_fsync(int fd)
{ {
return getsockopt(fd, level, optname, optval, optlen); // TODO
} }
int sys_setsockopt(int fd, int level, int optname, void *optval, socklen_t optlen) int mock_read(int fd, char *dst, int len)
{ {
return setsockopt(fd, level, optname, optval, optlen); // TODO
} }
int sys_recv(int fd, void *dst, int len, int flags) int mock_write(int fd, char *src, int len)
{ {
return recv(fd, dst, len, flags); // TODO
} }
int sys_send(int fd, void *src, int len, int flags) int mock_fstat(int fd, struct stat *buf)
{ {
return send(fd, src, len, flags); // TODO
} }
int sys_connect(int fd, void *addr, size_t addr_len) int mock_mkstemp(char *path)
{
return connect(fd, addr, addr_len);
}
int sys_clock_gettime(clockid_t clockid, struct timespec *tp)
{
return clock_gettime(clockid, tp);
}
int sys_open(char *path, int flags, int mode)
{
return open(path, flags, mode);
}
int sys_close(int fd)
{
return close(fd);
}
int sys_flock(int fd, int op)
{
return flock(fd, op);
}
int sys_fsync(int fd)
{
return fsync(fd);
}
int sys_read(int fd, char *dst, int len)
{
return read(fd, dst, len);
}
int sys_write(int fd, char *src, int len)
{
return write(fd, src, len);
}
int sys_fstat(int fd, struct stat *buf)
{
return fstat(fd, buf);
}
int sys_mkstemp(char *path)
{ {
return mkstemp(path); return mkstemp(path);
} }
char* sys_realpath(char *path, char *dst) char* mock_realpath(char *path, char *dst)
{ {
return realpath(path, dst); return realpath(path, dst);
} }
int sys_mkdir(char *path, mode_t mode) int mock_mkdir(char *path, mode_t mode)
{ {
return mkdir(path, mode); return mkdir(path, mode);
} }
src/system.h
+116 -54
View File
@@ -1,72 +1,134 @@
#include <stddef.h> #include <stdio.h>
void *sys_malloc_ (size_t len, char *file, int line);
void *sys_realloc_(void *ptr, size_t len, char *file, int line);
void sys_free_ (void *ptr, char *file, int line);
#define sys_malloc(len) sys_malloc_ ((len), __FILE__, __LINE__)
#define sys_realloc(ptr, len) sys_realloc_((ptr), (len), __FILE__, __LINE__)
#define sys_free(ptr) sys_free_ ((ptr), __FILE__, __LINE__)
int sys_remove(char *path);
int sys_rename(char *oldpath, char *newpath);
#ifdef _WIN32 #ifdef _WIN32
#define WIN32_LEAN_AND_MEAN #define WIN32_LEAN_AND_MEAN
#include <direct.h> // _mkdir
#include <windows.h> #include <windows.h>
#include <winsock2.h> #include <winsock2.h>
#include <ws2tcpip.h>
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);
SOCKET sys_accept (SOCKET fd, void *addr, int *addr_len);
int sys_getsockopt (SOCKET fd, int level, int optname, void *optval, int *optlen);
int sys_setsockopt (SOCKET fd, int level, int optname, void *optval, int 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);
BOOL sys_QueryPerformanceCounter(LARGE_INTEGER *lpPerformanceCount);
BOOL sys_QueryPerformanceFrequency(LARGE_INTEGER *lpFrequency);
HANDLE sys_CreateFileW (WCHAR *lpFileName, DWORD dwDesiredAccess, DWORD dwShareMode, LPSECURITY_ATTRIBUTES lpSecurityAttributes, DWORD dwCreationDisposition, DWORD dwFlagsAndAttributes, HANDLE hTemplateFile);
BOOL sys_CloseHandle (HANDLE handle);
BOOL sys_LockFile (HANDLE handle);
BOOL sys_UnlockFile (HANDLE handle);
BOOL sys_FlushFileBuffers (HANDLE handle);
BOOL sys_ReadFile (HANDLE handle, char *dst, DWORD len, DWORD *num, OVERLAPPED *ov);
BOOL sys_WriteFile (HANDLE handle, char *src, DWORD len, DWORD *num, OVERLAPPED *ov);
BOOL sys_GetFileSizeEx (HANDLE handle, LARGE_INTEGER *buf);
char* sys__fullpath (char *path, char *dst, int cap);
#else #else
#include <poll.h> #include <poll.h>
#include <time.h> #include <time.h>
#include <errno.h> #include <errno.h>
#include <fcntl.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <sys/file.h>
#include <sys/socket.h> #include <sys/socket.h>
#include <arpa/inet.h> #include <arpa/inet.h>
int sys_socket (int domain, int type, int protocol); #define SOCKET int
int sys_bind (int fd, void *addr, size_t addr_len); #define INVALID_SOCKET ((SOCKET) -1)
int sys_listen (int fd, int backlog);
int sys_accept (int fd, void *addr, socklen_t *addr_len); #endif
int sys_getsockopt (int fd, int level, int optname, void *optval, socklen_t *optlen);
int sys_setsockopt (int fd, int level, int optname, void *optval, socklen_t optlen); #ifdef BUILD_TEST
int sys_recv (int fd, void *dst, int len, int flags);
int sys_send (int fd, void *src, int len, int flags); int spawn_simulated_process(char *args);
int sys_connect (int fd, void *addr, size_t addr_len); void update_simulation(void);
int sys_clock_gettime (clockid_t clockid, struct timespec *tp); void cleanup_simulation(void);
int sys_open (char *path, int flags, int mode);
int sys_close (int fd); SOCKET mock_socket(int domain, int type, int protocol);
int sys_flock (int fd, int op); int mock_bind(SOCKET fd, void *addr, size_t addr_len);
int sys_fsync (int fd); int mock_listen(SOCKET fd, int backlog);
int sys_read (int fd, char *dst, int len); SOCKET mock_accept(SOCKET fd, void *addr, socklen_t *addr_len);
int sys_write (int fd, char *src, int len); int mock_getsockopt(SOCKET fd, int level, int optname, void *optval, socklen_t *optlen);
int sys_fstat (int fd, struct stat *buf); int mock_setsockopt(SOCKET fd, int level, int optname, void *optval, socklen_t optlen);
int sys_mkstemp (char *path); int mock_recv(SOCKET fd, void *dst, int len, int flags);
char* sys_realpath (char *path, char *dst); int mock_send(SOCKET fd, void *src, int len, int flags);
int sys_mkdir (char *path, mode_t mode); int mock_connect(SOCKET fd, void *addr, size_t addr_len);
// Common
#define sys_malloc mock_malloc
#define sys_realloc mock_realloc
#define sys_free mock_free
#define sys_remove mock_remove
#define sys_rename mock_rename
#define sys_socket mock_socket
#define sys_bind mock_bind
#define sys_listen mock_listen
#define sys_accept mock_accept
#define sys_getsockopt mock_getsockopt
#define sys_setsockopt mock_setsockopt
#define sys_recv mock_recv
#define sys_send mock_send
#define sys_connect mock_connect
// Windows
#define sys__mkdir mock__mkdir
#define sys_closesocket mock_closesocket
#define sys_CreateFileW mock_CreateFileW
#define sys_CloseHandle mock_CloseHandle
#define sys_LockFile mock_LockFile
#define sys_UnlockFile mock_UnlockFile
#define sys_FlushFileBuffers mock_FlushFileBuffers
#define sys_ReadFile mock_ReadFile
#define sys_WriteFile mock_WriteFile
#define sys_GetFileSizeEx mock_GetFileSizeEx
#define sys__fullpath mock__fullpath
#define sys_QueryPerformanceCounter mock_QueryPerformanceCounter
#define sys_QueryPerformanceFrequency mock_QueryPerformanceFrequency
// Linux
#define sys_mkdir mock_mkdir
#define sys_open mock_open
#define sys_close mock_close
#define sys_flock mock_flock
#define sys_fsync mock_fsync
#define sys_read mock_read
#define sys_write mock_write
#define sys_fstat mock_fstat
#define sys_mkstemp mock_mkstemp
#define sys_realpath mock_realpath
#define sys_clock_gettime mock_clock_gettime
#else
// Common
#define sys_malloc malloc
#define sys_realloc realloc
#define sys_free free
#define sys_remove remove
#define sys_rename rename
#define sys_socket socket
#define sys_bind bind
#define sys_listen listen
#define sys_accept accept
#define sys_getsockopt getsockopt
#define sys_setsockopt setsockopt
#define sys_recv recv
#define sys_send send
#define sys_connect connect
// Windows
#define sys__mkdir _mkdir
#define sys_closesocket closesocket
#define sys_CreateFileW CreateFileW
#define sys_CloseHandle CloseHandle
#define sys_LockFile LockFile
#define sys_UnlockFile UnlockFile
#define sys_FlushFileBuffers FlushFileBuffers
#define sys_ReadFile ReadFile
#define sys_WriteFile WriteFile
#define sys_GetFileSizeEx GetFileSizeEx
#define sys__fullpath _fullpath
#define sys_QueryPerformanceCounter QueryPerformanceCounter
#define sys_QueryPerformanceFrequency QueryPerformanceFrequency
// Linux
#define sys_mkdir mkdir
#define sys_open open
#define sys_close close
#define sys_flock flock
#define sys_fsync fsync
#define sys_read read
#define sys_write write
#define sys_fstat fstat
#define sys_mkstemp mkstemp
#define sys_realpath realpath
#define sys_clock_gettime clock_gettime
#endif #endif