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

Continue work on DST framework

Browse Source
This commit is contained in:
cozis
2025-10-29 21:46:17 +01:00
parent 24c0134c03
commit 174f37c6c0
17 changed files with 1407 additions and 710 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic.c
+4 -10
View File
@@ -1,11 +1,5 @@
#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#else
#include <time.h>
#endif
#include "basic.h" #include "basic.h"
#include "system.h"
bool streq(string s1, string s2) bool streq(string s1, string s2)
{ {
@@ -28,10 +22,10 @@ Time get_current_time(void)
int64_t freq; int64_t freq;
int ok; int ok;
ok = QueryPerformanceCounter((LARGE_INTEGER*) &count); ok = sys_QueryPerformanceCounter((LARGE_INTEGER*) &count);
if (!ok) return INVALID_TIME; if (!ok) return INVALID_TIME;
ok = QueryPerformanceFrequency((LARGE_INTEGER*) &freq); ok = sys_QueryPerformanceFrequency((LARGE_INTEGER*) &freq);
if (!ok) return INVALID_TIME; if (!ok) return INVALID_TIME;
uint64_t res = 1000 * (double) count / freq; uint64_t res = 1000 * (double) count / freq;
@@ -41,7 +35,7 @@ Time get_current_time(void)
{ {
struct timespec time; struct timespec time;
if (clock_gettime(CLOCK_REALTIME, &time)) if (sys_clock_gettime(CLOCK_REALTIME, &time))
return INVALID_TIME; return INVALID_TIME;
uint64_t res; uint64_t res;
src/byte_queue.c
+6 -18
View File
@@ -2,6 +2,7 @@
#include <assert.h> #include <assert.h>
#include <stdlib.h> #include <stdlib.h>
#include "system.h"
#include "byte_queue.h" #include "byte_queue.h"
// This is the implementation of a byte queue useful // This is the implementation of a byte queue useful
@@ -13,19 +14,6 @@
// //
// Only up to 4GB of data can be stored at once. // Only up to 4GB of data can be stored at once.
static void *mymalloc(ByteQueue *queue, uint32_t len)
{
(void) queue;
return malloc(len);
}
static void myfree(ByteQueue *queue, void *ptr, uint32_t len)
{
(void) queue;
(void) len,
free(ptr);
}
// Initialize the queue // Initialize the queue
void byte_queue_init(ByteQueue *queue, uint32_t limit) void byte_queue_init(ByteQueue *queue, uint32_t limit)
{ {
@@ -44,12 +32,12 @@ void byte_queue_free(ByteQueue *queue)
{ {
if (queue->read_target) { if (queue->read_target) {
if (queue->read_target != queue->data) if (queue->read_target != queue->data)
myfree(queue, queue->read_target, queue->read_target_size); sys_free(queue->read_target);
queue->read_target = NULL; queue->read_target = NULL;
queue->read_target_size = 0; queue->read_target_size = 0;
} }
myfree(queue, queue->data, queue->size); sys_free(queue->data);
queue->data = NULL; queue->data = NULL;
} }
@@ -111,7 +99,7 @@ void byte_queue_read_ack(ByteQueue *queue, uint32_t num)
if (queue->read_target) { if (queue->read_target) {
if (queue->read_target != queue->data) if (queue->read_target != queue->data)
myfree(queue, queue->read_target, queue->read_target_size); sys_free(queue->read_target);
queue->read_target = NULL; queue->read_target = NULL;
queue->read_target_size = 0; queue->read_target_size = 0;
} }
@@ -236,7 +224,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap)
if (size > queue->limit) if (size > queue->limit)
size = queue->limit; size = queue->limit;
uint8_t *data = mymalloc(queue, size); uint8_t *data = sys_malloc(size);
if (!data) { if (!data) {
queue->flags |= BYTE_QUEUE_ERROR; queue->flags |= BYTE_QUEUE_ERROR;
return 0; return 0;
@@ -246,7 +234,7 @@ int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap)
memcpy(data, queue->data + queue->head, queue->used); memcpy(data, queue->data + queue->head, queue->used);
if (queue->read_target != queue->data) if (queue->read_target != queue->data)
myfree(queue, queue->data, queue->size); sys_free(queue->data);
queue->data = data; queue->data = data;
queue->head = 0; queue->head = 0;
src/chunk_server.c
+31 -31
View File
@@ -19,7 +19,7 @@ pending_download_list_init(PendingDownloadList *list)
static void static void
pending_download_list_free(PendingDownloadList *list) pending_download_list_free(PendingDownloadList *list)
{ {
free(list->items); sys_free(list->items);
} }
static int static int
@@ -36,13 +36,13 @@ pending_download_list_add(PendingDownloadList *list, Address addr, SHA256 hash)
if (list->capacity == 0) new_capacity = 8; if (list->capacity == 0) new_capacity = 8;
else new_capacity = 2 * list->capacity; else new_capacity = 2 * list->capacity;
PendingDownload *new_items = malloc(new_capacity * sizeof(PendingDownload)); PendingDownload *new_items = sys_malloc(new_capacity * sizeof(PendingDownload));
if (new_items == NULL) if (new_items == NULL)
return -1; return -1;
if (list->capacity > 0) { if (list->capacity > 0) {
memcpy(new_items, list->items, list->count * sizeof(list->items[0])); memcpy(new_items, list->items, list->count * sizeof(list->items[0]));
free(list->items); sys_free(list->items);
} }
list->items = new_items; list->items = new_items;
@@ -135,12 +135,12 @@ static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk,
return -1; return -1;
if (patch_off > SIZE_MAX - patch.len) { if (patch_off > SIZE_MAX - patch.len) {
free(data.ptr); sys_free(data.ptr);
return -1; return -1;
} }
if (patch_off + (size_t) patch.len > (size_t) data.len) { if (patch_off + (size_t) patch.len > (size_t) data.len) {
free(data.ptr); sys_free(data.ptr);
return -1; return -1;
} }
@@ -148,11 +148,11 @@ static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk,
ret = store_chunk(store, data, new_hash); ret = store_chunk(store, data, new_hash);
if (ret < 0) { if (ret < 0) {
free(data.ptr); sys_free(data.ptr);
return -1; return -1;
} }
free(data.ptr); sys_free(data.ptr);
return 0; return 0;
} }
@@ -233,33 +233,33 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
if (!binary_read(&reader, &rem_count, sizeof(rem_count))) if (!binary_read(&reader, &rem_count, sizeof(rem_count)))
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message"));
SHA256 *add_list = malloc(add_count * sizeof(SHA256)); SHA256 *add_list = sys_malloc(add_count * sizeof(SHA256));
SHA256 *rem_list = malloc(rem_count * sizeof(SHA256)); SHA256 *rem_list = sys_malloc(rem_count * sizeof(SHA256));
if (add_list == NULL || rem_list == NULL) { if (add_list == NULL || rem_list == NULL) {
free(add_list); sys_free(add_list);
free(rem_list); sys_free(rem_list);
return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory")); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Out of memory"));
} }
for (uint32_t i = 0; i < add_count; i++) { for (uint32_t i = 0; i < add_count; i++) {
if (!binary_read(&reader, &add_list[i], sizeof(SHA256))) { if (!binary_read(&reader, &add_list[i], sizeof(SHA256))) {
free(add_list); sys_free(add_list);
free(rem_list); sys_free(rem_list);
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message"));
} }
} }
for (uint32_t i = 0; i < rem_count; i++) { for (uint32_t i = 0; i < rem_count; i++) {
if (!binary_read(&reader, &rem_list[i], sizeof(SHA256))) { if (!binary_read(&reader, &rem_list[i], sizeof(SHA256))) {
free(add_list); sys_free(add_list);
free(rem_list); sys_free(rem_list);
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message"));
} }
} }
if (binary_read(&reader, NULL, 1)) { if (binary_read(&reader, NULL, 1)) {
free(add_list); sys_free(add_list);
free(rem_list); sys_free(rem_list);
return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message")); return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_STATE_UPDATE_ERROR, S("Invalid message"));
} }
@@ -302,14 +302,14 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
(string) { main_path, strlen(main_path) }) < 0) { (string) { main_path, strlen(main_path) }) < 0) {
// Failed to move, treat as missing // Failed to move, treat as missing
if (missing_chunks == NULL) if (missing_chunks == NULL)
missing_chunks = malloc(add_count * sizeof(SHA256)); missing_chunks = sys_malloc(add_count * sizeof(SHA256));
if (missing_chunks) if (missing_chunks)
missing_chunks[missing_count++] = add_list[i]; missing_chunks[missing_count++] = add_list[i];
} }
} else { } else {
// Chunk is missing in both locations // Chunk is missing in both locations
if (missing_chunks == NULL) if (missing_chunks == NULL)
missing_chunks = malloc(add_count * sizeof(SHA256)); missing_chunks = sys_malloc(add_count * sizeof(SHA256));
if (missing_chunks) if (missing_chunks)
missing_chunks[missing_count++] = add_list[i]; missing_chunks[missing_count++] = add_list[i];
} }
@@ -338,8 +338,8 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
(string) { orphaned_path, strlen(orphaned_path) }); (string) { orphaned_path, strlen(orphaned_path) });
} }
free(add_list); sys_free(add_list);
free(rem_list); sys_free(rem_list);
// Send response // Send response
if (missing_count > 0) { if (missing_count > 0) {
@@ -356,7 +356,7 @@ process_metadata_server_state_update(ChunkServer *state, int conn_idx, ByteView
for (uint32_t i = 0; i < missing_count; i++) for (uint32_t i = 0; i < missing_count; i++)
message_write(&writer, &missing_chunks[i], sizeof(SHA256)); message_write(&writer, &missing_chunks[i], sizeof(SHA256));
free(missing_chunks); sys_free(missing_chunks);
if (!message_writer_free(&writer)) if (!message_writer_free(&writer))
return -1; return -1;
@@ -625,7 +625,7 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg)
if (binary_read(&reader, NULL, 1)) if (binary_read(&reader, NULL, 1))
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"));
char *mem = malloc(chunk_size); char *mem = sys_malloc(chunk_size);
if (mem == NULL) if (mem == NULL)
return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Out of memory")); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("Out of memory"));
@@ -639,7 +639,7 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg)
int ret = chunk_store_add(&state->store, (string) { mem, chunk_size }); int ret = chunk_store_add(&state->store, (string) { mem, chunk_size });
free(mem); sys_free(mem);
if (ret < 0) if (ret < 0)
return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("I/O error")); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_CREATE_CHUNK_ERROR, S("I/O error"));
@@ -732,7 +732,7 @@ process_client_download_chunk(ChunkServer *state, int conn_idx, ByteView msg)
return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("I/O error")); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("I/O error"));
if (target_off >= (size_t) data.len || target_len > (size_t) data.len - target_off) { if (target_off >= (size_t) data.len || target_len > (size_t) data.len - target_off) {
free(data.ptr); sys_free(data.ptr);
return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid range")); return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid range"));
} }
string slice = { data.ptr + target_off, target_len }; string slice = { data.ptr + target_off, target_len };
@@ -746,7 +746,7 @@ process_client_download_chunk(ChunkServer *state, int conn_idx, ByteView msg)
message_write(&writer, slice.ptr, slice.len); message_write(&writer, slice.ptr, slice.len);
free(data.ptr); sys_free(data.ptr);
if (!message_writer_free(&writer)) if (!message_writer_free(&writer))
return -1; return -1;
@@ -765,7 +765,7 @@ process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView
return -1; return -1;
} }
int chunk_server_init(ChunkServer *state, int argc, char **argv) int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled)
{ {
(void) argc; (void) argc;
(void) argv; (void) argv;
@@ -806,7 +806,7 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv)
state->metadata_server_disconnect_time = 0; state->metadata_server_disconnect_time = 0;
return 0; return tcp_register_events(&state->tcp, contexts, polled);
} }
int chunk_server_free(ChunkServer *state) int chunk_server_free(ChunkServer *state)
@@ -817,10 +817,10 @@ int chunk_server_free(ChunkServer *state)
return 0; return 0;
} }
int chunk_server_step(ChunkServer *state) int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled, int num_polled)
{ {
Event events[MAX_CONNS+1]; Event events[MAX_CONNS+1];
int num_events = tcp_process_events(&state->tcp, events); int num_events = tcp_translate_events(&state->tcp, contexts, polled, num_polled, events);
Time current_time = get_current_time(); Time current_time = get_current_time();
if (current_time == INVALID_TIME) if (current_time == INVALID_TIME)
@@ -938,5 +938,5 @@ int chunk_server_step(ChunkServer *state)
} }
} }
return 0; return tcp_register_events(&state->tcp, contexts, polled, num_polled);
} }
src/chunk_server.h
+2 -2
View File
@@ -35,8 +35,8 @@ typedef struct {
PendingDownloadList pending_download_list; PendingDownloadList pending_download_list;
} ChunkServer; } ChunkServer;
int chunk_server_init(ChunkServer *state, int argc, char **argv); int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled);
int chunk_server_free(ChunkServer *state); int chunk_server_free(ChunkServer *state);
int chunk_server_step(ChunkServer *state); int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled, int num_polled);
#endif // CHUNK_SERVER_INCLUDED #endif // CHUNK_SERVER_INCLUDED
src/client.c
+24 -23
View File
@@ -8,6 +8,7 @@
#endif #endif
#include "tcp.h" #include "tcp.h"
#include "system.h"
#include "config.h" #include "config.h"
#include "message.h" #include "message.h"
#include <TinyDFS.h> #include <TinyDFS.h>
@@ -90,7 +91,7 @@ struct TinyDFS {
TinyDFS *tinydfs_init(char *addr, uint16_t port) TinyDFS *tinydfs_init(char *addr, uint16_t port)
{ {
TinyDFS *tdfs = malloc(sizeof(TinyDFS)); TinyDFS *tdfs = sys_malloc(sizeof(TinyDFS));
if (tdfs == NULL) if (tdfs == NULL)
return NULL; return NULL;
@@ -98,7 +99,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port)
addr2.is_ipv4 = true; addr2.is_ipv4 = true;
addr2.port = port; addr2.port = port;
if (inet_pton(AF_INET, addr, &addr2.ipv4) != 1) { if (inet_pton(AF_INET, addr, &addr2.ipv4) != 1) {
free(tdfs); sys_free(tdfs);
return NULL; return NULL;
} }
@@ -106,7 +107,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port)
if (tcp_connect(&tdfs->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) { if (tcp_connect(&tdfs->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) {
tcp_context_free(&tdfs->tcp); tcp_context_free(&tdfs->tcp);
free(tdfs); sys_free(tdfs);
return NULL; return NULL;
} }
@@ -121,7 +122,7 @@ TinyDFS *tinydfs_init(char *addr, uint16_t port)
void tinydfs_free(TinyDFS *tdfs) void tinydfs_free(TinyDFS *tdfs)
{ {
tcp_context_free(&tdfs->tcp); tcp_context_free(&tdfs->tcp);
free(tdfs); sys_free(tdfs);
} }
static int static int
@@ -404,7 +405,7 @@ int tinydfs_submit_write(TinyDFS *tdfs, char *path, int path_len, int off, void
void tinydfs_result_free(TinyDFS_Result *result) void tinydfs_result_free(TinyDFS_Result *result)
{ {
if (result->type == TINYDFS_RESULT_LIST_SUCCESS) if (result->type == TINYDFS_RESULT_LIST_SUCCESS)
free(result->entities); sys_free(result->entities);
} }
static void process_event_for_create(TinyDFS *tdfs, static void process_event_for_create(TinyDFS *tdfs,
@@ -537,7 +538,7 @@ static void process_event_for_list(TinyDFS *tdfs,
return; return;
} }
TinyDFS_Entity *entities = malloc(item_count * sizeof(TinyDFS_Entity)); TinyDFS_Entity *entities = sys_malloc(item_count * sizeof(TinyDFS_Entity));
if (entities == NULL) { if (entities == NULL) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
return; return;
@@ -548,21 +549,21 @@ static void process_event_for_list(TinyDFS *tdfs,
uint8_t is_dir; uint8_t is_dir;
if (!binary_read(&reader, &is_dir, sizeof(is_dir))) { if (!binary_read(&reader, &is_dir, sizeof(is_dir))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
free(entities); sys_free(entities);
return; return;
} }
uint16_t name_len; uint16_t name_len;
if (!binary_read(&reader, &name_len, sizeof(name_len))) { if (!binary_read(&reader, &name_len, sizeof(name_len))) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
free(entities); sys_free(entities);
return; return;
} }
char *name = reader.src + reader.cur; char *name = 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 };
free(entities); sys_free(entities);
return; return;
} }
@@ -570,7 +571,7 @@ static void process_event_for_list(TinyDFS *tdfs,
if (name_len > sizeof(entities[i].name)-1) { if (name_len > sizeof(entities[i].name)-1) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
free(entities); sys_free(entities);
return; return;
} }
memcpy(entities[i].name, name, name_len); memcpy(entities[i].name, name, name_len);
@@ -580,7 +581,7 @@ static void process_event_for_list(TinyDFS *tdfs,
// Check there is nothing else to read // Check there is nothing else to read
if (binary_read(&reader, NULL, 1)) { if (binary_read(&reader, NULL, 1)) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_LIST_ERROR };
free(entities); sys_free(entities);
return; return;
} }
@@ -653,7 +654,7 @@ static void process_event_for_read(TinyDFS *tdfs,
} }
// Allocate ranges // Allocate ranges
Range *ranges = malloc(num_chunks_needed * sizeof(Range)); Range *ranges = sys_malloc(num_chunks_needed * sizeof(Range));
if (ranges == NULL) { if (ranges == NULL) {
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
@@ -667,7 +668,7 @@ static void process_event_for_read(TinyDFS *tdfs,
// Read hash // Read hash
SHA256 hash; SHA256 hash;
if (!binary_read(&reader, &hash, sizeof(hash))) { if (!binary_read(&reader, &hash, sizeof(hash))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -675,7 +676,7 @@ static void process_event_for_read(TinyDFS *tdfs,
// Read number of servers // Read number of servers
uint32_t num_servers; uint32_t num_servers;
if (!binary_read(&reader, &num_servers, sizeof(num_servers))) { if (!binary_read(&reader, &num_servers, sizeof(num_servers))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -683,7 +684,7 @@ static void process_event_for_read(TinyDFS *tdfs,
// Parse IPv4 addresses // Parse IPv4 addresses
uint32_t num_ipv4; uint32_t num_ipv4;
if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) { if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -697,7 +698,7 @@ static void process_event_for_read(TinyDFS *tdfs,
uint16_t port; uint16_t port;
if (!binary_read(&reader, &ipv4, sizeof(ipv4)) || if (!binary_read(&reader, &ipv4, sizeof(ipv4)) ||
!binary_read(&reader, &port, sizeof(port))) { !binary_read(&reader, &port, sizeof(port))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -712,21 +713,21 @@ static void process_event_for_read(TinyDFS *tdfs,
// Skip IPv6 addresses // Skip IPv6 addresses
uint32_t num_ipv6; uint32_t num_ipv6;
if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) { if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
for (uint32_t j = 0; j < num_ipv6; j++) { for (uint32_t j = 0; j < num_ipv6; j++) {
if (!binary_read(&reader, NULL, sizeof(IPv6)) || if (!binary_read(&reader, NULL, sizeof(IPv6)) ||
!binary_read(&reader, NULL, sizeof(uint16_t))) { !binary_read(&reader, NULL, sizeof(uint16_t))) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
} }
if (!found) { if (!found) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -771,7 +772,7 @@ static void process_event_for_read(TinyDFS *tdfs,
Range *r = &ranges[0]; Range *r = &ranges[0];
int cs_idx = get_chunk_server_connection(tdfs, r->server_addr); int cs_idx = get_chunk_server_connection(tdfs, r->server_addr);
if (cs_idx < 0) { if (cs_idx < 0) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -779,7 +780,7 @@ static void process_event_for_read(TinyDFS *tdfs,
if (send_download_chunk(tdfs, cs_idx, r->hash, r->offset_within_chunk, if (send_download_chunk(tdfs, cs_idx, r->hash, r->offset_within_chunk,
r->length_within_chunk, opidx, 0) < 0) { r->length_within_chunk, opidx, 0) < 0) {
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_ERROR };
return; return;
} }
@@ -788,7 +789,7 @@ static void process_event_for_read(TinyDFS *tdfs,
tdfs->operations[opidx].ranges_head = 1; tdfs->operations[opidx].ranges_head = 1;
} else { } else {
// No chunks to download // No chunks to download
free(ranges); sys_free(ranges);
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS };
} }
@@ -861,7 +862,7 @@ static void process_event_for_read(TinyDFS *tdfs,
// Check if done // Check if done
if (tdfs->operations[opidx].num_pending == 0) { if (tdfs->operations[opidx].num_pending == 0) {
free(tdfs->operations[opidx].ranges); sys_free(tdfs->operations[opidx].ranges);
tdfs->operations[opidx].ranges = NULL; tdfs->operations[opidx].ranges = NULL;
tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS }; tdfs->operations[opidx].result = (TinyDFS_Result) { .type=TINYDFS_RESULT_READ_SUCCESS };
} }
src/file_system.c
+36 -34
View File
@@ -17,6 +17,7 @@
#include <windows.h> #include <windows.h>
#endif #endif
#include "system.h"
#include "file_system.h" #include "file_system.h"
int rename_file_or_dir(string oldpath, string newpath); int rename_file_or_dir(string oldpath, string newpath);
@@ -30,7 +31,7 @@ int file_open(string path, Handle *fd)
memcpy(zt, path.ptr, path.len); memcpy(zt, path.ptr, path.len);
zt[path.len] = '\0'; zt[path.len] = '\0';
int ret = open(zt, O_RDWR | O_CREAT | O_APPEND, 0644); int ret = sys_open(zt, O_RDWR | O_CREAT | O_APPEND, 0644);
if (ret < 0) if (ret < 0)
return -1; return -1;
@@ -43,7 +44,7 @@ int file_open(string path, Handle *fd)
MultiByteToWideChar(CP_UTF8, 0, path.ptr, path.len, wpath, MAX_PATH); MultiByteToWideChar(CP_UTF8, 0, path.ptr, path.len, wpath, MAX_PATH);
wpath[path.len] = L'\0'; wpath[path.len] = L'\0';
HANDLE h = CreateFileW( HANDLE h = sys_CreateFileW(
wpath, wpath,
GENERIC_WRITE | GENERIC_READ, GENERIC_WRITE | GENERIC_READ,
0, 0,
@@ -63,24 +64,24 @@ int file_open(string path, Handle *fd)
void file_close(Handle fd) void file_close(Handle fd)
{ {
#ifdef __linux__ #ifdef __linux__
close((int) fd.data); sys_close((int) fd.data);
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
CloseHandle((HANDLE) fd.data); sys_CloseHandle((HANDLE) fd.data);
#endif #endif
} }
int file_lock(Handle fd) int file_lock(Handle fd)
{ {
#ifdef __linux__ #ifdef __linux__
if (flock((int) fd.data, LOCK_EX) < 0) if (sys_flock((int) fd.data, LOCK_EX) < 0)
return -1; return -1;
return 0; return 0;
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
if (!LockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) if (!sys_LockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD))
return -1; return -1;
return 0; return 0;
#endif #endif
@@ -89,13 +90,13 @@ int file_lock(Handle fd)
int file_unlock(Handle fd) int file_unlock(Handle fd)
{ {
#ifdef __linux__ #ifdef __linux__
if (flock((int) fd.data, LOCK_UN) < 0) if (sys_flock((int) fd.data, LOCK_UN) < 0)
return -1; return -1;
return 0; return 0;
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
if (!UnlockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD)) if (!sys_UnlockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD))
return -1; return -1;
return 0; return 0;
#endif #endif
@@ -104,13 +105,13 @@ int file_unlock(Handle fd)
int file_sync(Handle fd) int file_sync(Handle fd)
{ {
#ifdef __linux__ #ifdef __linux__
if (fsync((int) fd.data) < 0) if (sys_fsync((int) fd.data) < 0)
return -1; return -1;
return 0; return 0;
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
if (!FlushFileBuffers((HANDLE) fd.data)) if (!sys_FlushFileBuffers((HANDLE) fd.data))
return -1; return -1;
return 0; return 0;
#endif #endif
@@ -119,12 +120,12 @@ int file_sync(Handle fd)
int file_read(Handle fd, char *dst, int max) int file_read(Handle fd, char *dst, int max)
{ {
#ifdef __linux__ #ifdef __linux__
return read((int) fd.data, dst, max); return sys_read((int) fd.data, dst, max);
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
DWORD num; DWORD num;
if (!ReadFile((HANDLE) fd.data, dst, max, &num, NULL)) if (!sys_ReadFile((HANDLE) fd.data, dst, max, &num, NULL))
return -1; return -1;
if (num > INT_MAX) if (num > INT_MAX)
return -1; return -1;
@@ -135,12 +136,12 @@ 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)
{ {
#ifdef __linux__ #ifdef __linux__
return write((int) fd.data, src, len); return sys_write((int) fd.data, src, len);
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
DWORD num; DWORD num;
if (!WriteFile((HANDLE) fd.data, src, len, &num, NULL)) if (!sys_WriteFile((HANDLE) fd.data, src, len, &num, NULL))
return -1; return -1;
if (num > INT_MAX) if (num > INT_MAX)
return -1; return -1;
@@ -152,7 +153,7 @@ int file_size(Handle fd, size_t *len)
{ {
#ifdef __linux__ #ifdef __linux__
struct stat buf; struct stat buf;
if (fstat((int) fd.data, &buf) < 0) if (sys_fstat((int) fd.data, &buf) < 0)
return -1; return -1;
if (buf.st_size < 0 || (uint64_t) buf.st_size > SIZE_MAX) if (buf.st_size < 0 || (uint64_t) buf.st_size > SIZE_MAX)
return -1; return -1;
@@ -162,7 +163,7 @@ int file_size(Handle fd, size_t *len)
#ifdef _WIN32 #ifdef _WIN32
LARGE_INTEGER buf; LARGE_INTEGER buf;
if (!GetFileSizeEx((HANDLE) fd.data, &buf)) if (!sys_GetFileSizeEx((HANDLE) fd.data, &buf))
return -1; return -1;
if (buf.QuadPart < 0 || (uint64_t) buf.QuadPart > SIZE_MAX) if (buf.QuadPart < 0 || (uint64_t) buf.QuadPart > SIZE_MAX)
return -1; return -1;
@@ -193,7 +194,7 @@ static int write_bytes(int fd, string data)
{ {
size_t written = 0; size_t written = 0;
while (written < (size_t) data.len) { while (written < (size_t) data.len) {
int ret = write(fd, data.ptr + written, data.len - written); int ret = sys_write(fd, data.ptr + written, data.len - written);
if (ret < 0) { if (ret < 0) {
if (errno == EINTR) if (errno == EINTR)
continue; continue;
@@ -218,26 +219,26 @@ int file_write_atomic(string path, string content)
memcpy(tmp_path + parent.len, pattern, strlen(pattern)); memcpy(tmp_path + parent.len, pattern, strlen(pattern));
tmp_path[parent.len + strlen(pattern)] = '\0'; tmp_path[parent.len + strlen(pattern)] = '\0';
int fd = mkstemp(tmp_path); int fd = sys_mkstemp(tmp_path);
if (fd < 0) if (fd < 0)
return -1; return -1;
if (write_bytes(fd, content) < 0) { if (write_bytes(fd, content) < 0) {
close(fd); sys_close(fd);
remove(tmp_path); sys_remove(tmp_path);
return -1; return -1;
} }
#ifdef _WIN32 #ifdef _WIN32
if (_commit(fd)) { if (sys__commit(fd)) {
close(fd); sys_close(fd);
remove(tmp_path); sys_remove(tmp_path);
return -1; return -1;
} }
#else #else
if (fsync(fd)) { if (sys_fsync(fd)) {
close(fd); sys_close(fd);
remove(tmp_path); sys_remove(tmp_path);
return -1; return -1;
} }
#endif #endif
@@ -245,7 +246,7 @@ int file_write_atomic(string path, string content)
close(fd); close(fd);
if (rename_file_or_dir((string) { tmp_path, strlen(tmp_path) }, path)) { if (rename_file_or_dir((string) { tmp_path, strlen(tmp_path) }, path)) {
remove(tmp_path); sys_remove(tmp_path);
return -1; return -1;
} }
return 0; return 0;
@@ -260,10 +261,10 @@ int create_dir(string path)
zt[path.len] = '\0'; zt[path.len] = '\0';
#ifdef _WIN32 #ifdef _WIN32
if (mkdir(zt) < 0) if (sys_mkdir(zt) < 0)
return -1; return -1;
#else #else
if (mkdir(zt, 0766)) if (sys_mkdir(zt, 0766))
return -1; return -1;
#endif #endif
@@ -284,7 +285,7 @@ int rename_file_or_dir(string oldpath, string newpath)
memcpy(newpath_zt, newpath.ptr, newpath.len); memcpy(newpath_zt, newpath.ptr, newpath.len);
newpath_zt[newpath.len] = '\0'; newpath_zt[newpath.len] = '\0';
if (rename(oldpath_zt, newpath_zt)) if (sys_rename(oldpath_zt, newpath_zt))
return -1; return -1;
return 0; return 0;
} }
@@ -297,7 +298,7 @@ int remove_file_or_dir(string path)
memcpy(path_zt, path.ptr, path.len); memcpy(path_zt, path.ptr, path.len);
path_zt[path.len] = '\0'; path_zt[path.len] = '\0';
if (remove(path_zt)) if (sys_remove(path_zt))
return -1; return -1;
return 0; return 0;
} }
@@ -311,12 +312,12 @@ int get_full_path(string path, char *dst)
path_zt[path.len] = '\0'; path_zt[path.len] = '\0';
#ifdef __linux__ #ifdef __linux__
if (realpath(path_zt, dst) == NULL) if (sys_realpath(path_zt, dst) == NULL)
return -1; return -1;
#endif #endif
#ifdef _WIN32 #ifdef _WIN32
if (_fullpath(path_zt, dst, PATH_MAX) == NULL) if (sys__fullpath(path_zt, dst, PATH_MAX) == NULL)
return -1; return -1;
#endif #endif
@@ -341,7 +342,7 @@ int file_read_all(string path, string *data)
return -1; return -1;
} }
char *dst = malloc(len); char *dst = sys_malloc(len);
if (dst == NULL) { if (dst == NULL) {
file_close(fd); file_close(fd);
return -1; return -1;
@@ -351,6 +352,7 @@ int file_read_all(string path, string *data)
while ((size_t) copied < len) { while ((size_t) copied < len) {
ret = file_read(fd, dst + copied, len - copied); ret = file_read(fd, dst + copied, len - copied);
if (ret < 0) { if (ret < 0) {
sys_free(dst);
file_close(fd); file_close(fd);
return -1; return -1;
} }
src/file_tree.c
+7 -6
View File
@@ -2,6 +2,7 @@
#include <string.h> #include <string.h>
#include <stdlib.h> #include <stdlib.h>
#include "system.h"
#include "file_tree.h" #include "file_tree.h"
static int parse_path(string path, string *comps, int max) static int parse_path(string path, string *comps, int max)
@@ -99,7 +100,7 @@ static void dir_free(Dir *d)
{ {
for (uint64_t i = 0; i < d->num_children; i++) for (uint64_t i = 0; i < d->num_children; i++)
entity_free(&d->children[i]); entity_free(&d->children[i]);
free(d->children); sys_free(d->children);
} }
static void dir_remove(Dir *d, int idx) static void dir_remove(Dir *d, int idx)
@@ -124,7 +125,7 @@ static void file_init(File *f, uint64_t chunk_size)
static void file_free(File *f) static void file_free(File *f)
{ {
free(f->chunks); sys_free(f->chunks);
f->chunks = NULL; f->chunks = NULL;
} }
@@ -268,14 +269,14 @@ int file_tree_create_entity(FileTree *ft, string path,
if (new_max == 0) if (new_max == 0)
new_max = 8; new_max = 8;
Entity *p = malloc(sizeof(Entity) * new_max); Entity *p = sys_malloc(sizeof(Entity) * new_max);
if (p == NULL) if (p == NULL)
return FILETREE_NOMEM; return FILETREE_NOMEM;
for (uint64_t i = 0; i < d->num_children; i++) for (uint64_t i = 0; i < d->num_children; i++)
p[i] = d->children[i]; p[i] = d->children[i];
free(d->children); sys_free(d->children);
d->children = p; d->children = p;
d->max_children = new_max; d->max_children = new_max;
} }
@@ -340,13 +341,13 @@ int file_tree_write(FileTree *ft, string path,
uint64_t last_chunk_index = (off + len - 1) / f->chunk_size; uint64_t last_chunk_index = (off + len - 1) / f->chunk_size;
if (last_chunk_index >= f->num_chunks) { if (last_chunk_index >= f->num_chunks) {
SHA256 *new_chunks = malloc((last_chunk_index+1) * sizeof(SHA256)); SHA256 *new_chunks = sys_malloc((last_chunk_index+1) * sizeof(SHA256));
if (new_chunks == NULL) if (new_chunks == NULL)
return FILETREE_NOMEM; return FILETREE_NOMEM;
if (f->chunks) { if (f->chunks) {
if (f->num_chunks > 0) if (f->num_chunks > 0)
memcpy(new_chunks, f->chunks, f->num_chunks); memcpy(new_chunks, f->chunks, f->num_chunks);
free(f->chunks); sys_free(f->chunks);
} }
f->chunks = new_chunks; f->chunks = new_chunks;
f->num_chunks = last_chunk_index+1; f->num_chunks = last_chunk_index+1;
src/main.c
-52
View File
@@ -1,52 +0,0 @@
#ifdef BUILD_SERVER
#include <string.h>
#include "chunk_server.h"
#include "metadata_server.h"
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 main(int argc, char **argv)
{
int ret;
if (is_leader(argc, argv)) {
MetadataServer state;
ret = metadata_server_init(&state, argc, argv);
if (ret)
return ret;
for (;;) {
ret = metadata_server_step(&state);
if (ret)
return ret;
}
ret = metadata_server_free(&state);
} else {
ChunkServer state;
ret = chunk_server_init(&state, argc, argv);
if (ret)
return ret;
for (;;) {
ret = chunk_server_step(&state);
if (ret)
return ret;
}
ret = chunk_server_free(&state);
}
return ret;
}
#endif // BUILD_SERVER
src/main_server.c
+62
View File
@@ -0,0 +1,62 @@
#ifdef BUILD_SERVER
#include <string.h>
#include "core/chunk_server.h"
#include "core/metadata_server.h"
int metadata_server_main(int argc, char **argv)
{
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled;
MetadataServer state;
num_polled = metadata_server_init(
&state, argc, argv, contexts, polled);
if (num_polled < 0) return -1;
for (;;) {
POLL(polled, num_polled, -1);
num_polled = metadata_server_step(
&state, contexts, polled, num_polled);
if (num_polled < 0) return -1;
}
metadata_server_free(&state);
return 0;
}
int chunk_server_main(int argc, char **argv)
{
void *contexts[MAX_CONNS+1];
struct pollfd polled[MAX_CONNS+1];
int num_polled;
ChunkServer state;
num_polled = chunk_server_init(
&state, argc, argv, contexts, polled);
if (num_polled < 0) return -1;
for (;;) {
POLL(polled, num_polled, -1);
num_polled = chunk_server_step(
&state, contexts, polled, num_polled);
if (num_polled < 0) return -1;
}
chunk_server_free(&state);
return 0;
}
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 main(int argc, char **argv)
{
if (is_leader(argc, argv))
return metadata_server_main(argc, argv);
else
return chunk_server_main(argc, argv);
}
#endif
src/main_test.c
+957
View File
@@ -0,0 +1,957 @@
#ifdef BUILD_TEST
#include "core/chunk_server.h"
#include "core/metadata_server.h"
#define MAX_PROCESSES 128
#define MAX_DESCRIPTORS 1024
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_LISTENER_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->proxy_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)
{
// TODO: set simulation_should_stop=true on ctrl+C
spawn_simulated_process("--addr 127.0.0.1 8080 --leader");
spawn_simulated_process("--addr 127.0.0.1 8081");
spawn_simulated_process("--addr 127.0.0.1 8082");
spawn_simulated_process("--addr 127.0.0.1 8083");
spawn_simulated_process("--addr 127.0.0.1 8084");
spawn_simulated_process("--addr 127.0.0.1 8085");
spawn_simulated_process("--addr 127.0.0.1 8086");
spawn_simulated_process("--addr 127.0.0.1 8087");
spawn_simulated_process("--addr 127.0.0.1 8088");
spawn_simulated_process("--addr 127.0.0.1 8089");
spawn_simulated_process("--addr 127.0.0.1 8090");
while (!simulation_should_stop)
update_simulation();
cleanup_simulation();
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];
}
#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_stat(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);
}
#endif
#endif // BUILD_TEST
src/metadata_server.c
+7 -7
View File
@@ -16,7 +16,7 @@ static void hash_list_init(HashList *hash_list)
static void hash_list_free(HashList *hash_list) static void hash_list_free(HashList *hash_list)
{ {
free(hash_list->items); sys_free(hash_list->items);
} }
static int hash_list_insert(HashList *hash_list, SHA256 hash) static int hash_list_insert(HashList *hash_list, SHA256 hash)
@@ -30,7 +30,7 @@ static int hash_list_insert(HashList *hash_list, SHA256 hash)
int new_capacity = hash_list->capacity ? hash_list->capacity * 2 : 16; int new_capacity = hash_list->capacity ? hash_list->capacity * 2 : 16;
SHA256 *new_items = realloc(hash_list->items, new_capacity * sizeof(SHA256)); SHA256 *new_items = sys_realloc(hash_list->items, new_capacity * sizeof(SHA256));
if (new_items == NULL) if (new_items == NULL)
return -1; return -1;
@@ -750,7 +750,7 @@ static bool is_chunk_server_message_type(uint16_t type)
return false; return false;
} }
int metadata_server_init(MetadataServer *state, int argc, char **argv) int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled)
{ {
(void) argc; (void) argc;
(void) argv; (void) argv;
@@ -778,7 +778,7 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv)
return -1; return -1;
} }
return 0; return tcp_register_events(&state->tcp, contexts, polled);
} }
int metadata_server_free(MetadataServer *state) int metadata_server_free(MetadataServer *state)
@@ -788,10 +788,10 @@ int metadata_server_free(MetadataServer *state)
return 0; return 0;
} }
int metadata_server_step(MetadataServer *state) int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *polled, int num_polled)
{ {
Event events[MAX_CONNS+1]; Event events[MAX_CONNS+1];
int num_events = tcp_process_events(&state->tcp, events); int num_events = tcp_translate_events(&state->tcp, events, contexts, polled, num_polled);
for (int i = 0; i < num_events; i++) { for (int i = 0; i < num_events; i++) {
int conn_idx = events[i].conn_idx; int conn_idx = events[i].conn_idx;
@@ -852,5 +852,5 @@ int metadata_server_step(MetadataServer *state)
} }
} }
return 0; return tcp_register_events(&state->tcp, contexts, polled);
} }
src/metadata_server.h
+1 -1
View File
@@ -50,6 +50,6 @@ typedef struct {
int metadata_server_init(MetadataServer *state, int argc, char **argv); int metadata_server_init(MetadataServer *state, int argc, char **argv);
int metadata_server_free(MetadataServer *state); int metadata_server_free(MetadataServer *state);
int metadata_server_step(MetadataServer *state); int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *polled, int num_polled);
#endif // METADATA_SERVER_INCLUDED #endif // METADATA_SERVER_INCLUDED
src/system.c
+194 -24
View File
@@ -1,68 +1,238 @@
#ifndef BUILD_TEST
#include "system.h" #include "system.h"
Socket sys_socket(int domain, int type, int protocol) void *sys_malloc_(size_t len, char *file, int line)
{
(void) file;
(void) line;
return malloc(len);
}
void *sys_realloc_(void *ptr, size_t len, char *file, int line)
{
(void) file;
(void) line;
return realloc(ptr, len);
}
void sys_free_(void *ptr, char *file, int line)
{
(void) file;
(void) line;
free(ptr);
}
#ifdef _WIN32
SOCKET sys_socket(int domain, int type, int protocol)
{ {
return socket(domain, type, protocol); return socket(domain, type, protocol);
} }
int sys_bind(Socket fd, void *addr, size_t addr_len) int sys_bind(SOCKET fd, void *addr, size_t addr_len)
{ {
return bind(fd, addr, addr_len); return bind(fd, addr, addr_len);
} }
int sys_listen(Socket fd, int backlog) int sys_listen(SOCKET fd, int backlog)
{ {
return listen(fd, backlog); return listen(fd, backlog);
} }
int sys_closesocket(Socket fd) int sys_closesocket(SOCKET fd)
{ {
#ifdef _WIN32 return closesocket(fd);
closesocket(fd);
#else
close(fd);
#endif
} }
int sys_poll(struct pollfd *polled, int num_polled, int timeout) int sys_WSAPoll(struct pollfd *polled, int num_polled, int timeout)
{ {
#ifdef _WIN32
return WSAPoll(polled, num_polled, timeout); return WSAPoll(polled, num_polled, timeout);
#else
return poll(polled, num_polled, timeout);
#endif
} }
Socket sys_accept(Socket fd, void *addr, int *addr_len) SOCKET sys_accept(SOCKET fd, void *addr, int *addr_len)
{ {
return accept(fd, addr, addr_len); return accept(fd, addr, addr_len);
} }
int sys_getsockopt(Socket fd, int level, int optname, void *optval, socklen_t *optlen) int sys_getsockopt(SOCKET fd, int level, int optname, void *optval, int *optlen)
{ {
return getsockopt(fd, level, optname, optval, optlen); return getsockopt(fd, level, optname, optval);
} }
int sys_setsockopt(Socket fd, int level, int optname, void *optval, socklen_t optlen) int sys_setsockopt(SOCKET fd, int level, int optname, void *optval, int optlen)
{ {
return setsockopt(fd, level, optname, optval, optlen); return setsockopt(fd, level, optname, optval, optlen);
} }
int sys_recv(Socket fd, void *dst, int len, int flags) int sys_recv(SOCKET fd, void *dst, int len, int flags)
{ {
return recv(fd, dst, len, flags); return recv(fd, dst, len, flags);
} }
int sys_send(Socket fd, void *src, int len, int flags) int sys_send(SOCKET fd, void *src, int len, int flags)
{ {
return send(fd, src, len, flags); return send(fd, src, len, flags);
} }
int sys_connect(Socket fd, void *addr, size_t addr_len) int sys_connect(SOCKET fd, void *addr, size_t addr_len)
{ {
return connect(fd, addr, addr_len); return connect(fd, addr, addr_len);
} }
#endif // BUILD_TEST 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
}
char *sys__fullpath(char *path, char *dst, int cap)
{
return _fullpath(path, dst, cap);
}
#else
int sys_socket(int domain, int type, int protocol)
{
return socket(domain, type, protocol);
}
int sys_bind(int fd, void *addr, size_t addr_len)
{
return bind(fd, addr, addr_len);
}
int sys_listen(int fd, int backlog)
{
return listen(fd, backlog);
}
int sys_close(int fd)
{
return close(fd);
}
int sys_poll(struct pollfd *polled, int num_polled, int timeout)
{
return poll(polled, num_polled, timeout);
}
int sys_accept(int fd, void *addr, int *addr_len)
{
return accept(fd, addr, addr_len);
}
int sys_getsockopt(int fd, int level, int optname, void *optval, socklen_t *optlen)
{
return getsockopt(fd, level, optname, optval, optlen);
}
int sys_setsockopt(int fd, int level, int optname, void *optval, socklen_t optlen)
{
return setsockopt(fd, level, optname, optval, optlen);
}
int sys_recv(int fd, void *dst, int len, int flags)
{
return recv(fd, dst, len, flags);
}
int sys_send(int fd, void *src, int len, int flags)
{
return send(fd, src, len, flags);
}
int sys_connect(int fd, void *addr, size_t addr_len)
{
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_stat(int fd, struct stat *buf)
{
return stat(fd, buf);
}
int sys_mkstemp(char *path)
{
return mkstemp(path);
}
char* sys_realpath(char *path, char *dst)
{
return realpath(path, dst);
}
#endif
src/system.h
+68 -32
View File
@@ -1,36 +1,72 @@
#ifndef SYSTEM_INCLUDED
#define SYSTEM_INCLUDED 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__)
#ifdef _WIN32 #ifdef _WIN32
#include <winsock2.h>
#include <ws2tcpip.h> typedef union _LARGE_INTEGER {
struct {
DWORD LowPart;
LONG HighPart;
} DUMMYSTRUCTNAME;
struct {
DWORD LowPart;
LONG HighPart;
} u;
LONGLONG QuadPart;
} LARGE_INTEGER;
#define BOOL int
#define WCHAR wchar_t
#define SOCKET void*
#define HANDLE void*
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 <errno.h> int sys_socket (int domain, int type, int protocol);
#include <unistd.h> int sys_bind (int fd, void *addr, size_t addr_len);
#include <arpa/inet.h> int sys_listen (int fd, int backlog);
#define SOCKET int int sys_accept (int fd, void *addr, int *addr_len);
#define INVALID_SOCKET -1 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);
int sys_recv (int fd, void *dst, int len, int flags);
int sys_send (int fd, void *src, int len, int flags);
int sys_connect (int fd, void *addr, size_t addr_len);
int sys_clock_gettime (clockid_t clockid, struct timespec *tp);
int sys_open (char *path, int flags, int mode);
int sys_close (int fd);
int sys_flock (int fd, int op);
int sys_fsync (int fd);
int sys_read (int fd, char *dst, int len);
int sys_stat (int fd, struct stat *buf);
int sys_mkstemp (char *path);
char* sys_realpath (char *path, char *dst);
#endif #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
+6 -5
View File
@@ -147,11 +147,8 @@ void tcp_consume_message(TCP *tcp, int conn_idx)
tcp->conns[conn_idx].msglen = 0; tcp->conns[conn_idx].msglen = 0;
} }
// The "events" array must be an array of capacity MAX_CONNS+1 int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled)
int tcp_process_events(TCP *tcp, Event *events)
{ {
struct pollfd polled[MAX_CONNS + 1];
void *contexts[MAX_CONNS + 1];
int num_polled = 0; int num_polled = 0;
if (tcp->listen_fd != INVALID_SOCKET && tcp->num_conns < MAX_CONNS) { if (tcp->listen_fd != INVALID_SOCKET && tcp->num_conns < MAX_CONNS) {
@@ -173,8 +170,12 @@ int tcp_process_events(TCP *tcp, Event *events)
} }
} }
sys_poll(polled, num_polled, -1); return 0;
}
// The "events" array must be an array of capacity MAX_CONNS+1
int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled)
{
bool removed[MAX_CONNS+1]; bool removed[MAX_CONNS+1];
int num_events = 0; int num_events = 0;
src/tcp.h
+2 -1
View File
@@ -59,7 +59,8 @@ int tcp_index_from_tag(TCP *tcp, int tag);
int tcp_listen(TCP *tcp, char *addr, uint16_t port); int tcp_listen(TCP *tcp, char *addr, uint16_t port);
int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type); int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type);
void tcp_consume_message(TCP *tcp, int conn_idx); void tcp_consume_message(TCP *tcp, int conn_idx);
int tcp_process_events(TCP *tcp, Event *events); int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled);
int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled);
ByteQueue *tcp_output_buffer(TCP *tcp, int conn_idx); ByteQueue *tcp_output_buffer(TCP *tcp, int conn_idx);
int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output); int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output);
void tcp_close(TCP *tcp, int conn_idx); void tcp_close(TCP *tcp, int conn_idx);
src/test.c
-464
View File
@@ -1,464 +0,0 @@
#ifdef BUILD_TEST
#include <string.h>
#include <stdbool.h>
#include "system.h"
#include "chunk_server.h"
#include "metadata_server.h"
#include "byte_queue.h"
#define MAX_DESCRIPTORS 1024
#define MAX_PENDING_CONNECTIONS 128
typedef enum {
DESCRIPTOR_TYPE_EMPTY,
DESCRIPTOR_TYPE_FILE,
DESCRIPTOR_TYPE_PRECONF_SOCKET,
DESCRIPTOR_TYPE_LISTEN_SOCKET,
DESCRIPTOR_TYPE_CONNECTION_SOCKET,
} DescriptorType;
typedef struct {
DescriptorType type;
// Socket fields
struct sockaddr_in bind;
bool no_bind;
// Listen socket fields
int backlog;
int pending_connections[MAX_PENDING_CONNECTIONS];
int num_pending;
// Data socket fields
int peer_process; // Index of the peer process (-1 if not connected)
int peer_fd; // Descriptor in peer process
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 (fd < MAX_DESCRIPTORS && current_process->desc[fd].type != DESCRIPTOR_TYPE_EMPTY)
fd++;
if (fd >= MAX_DESCRIPTORS) {
// TODO: errno
return BAD_SOCKET;
}
current_process->desc[fd].type = DESCRIPTOR_TYPE_PRECONF_SOCKET;
current_process->desc[fd].no_bind = true;
current_process->desc[fd].num_pending = 0;
current_process->desc[fd].peer_process = -1;
current_process->desc[fd].peer_fd = -1;
return fd;
}
int sys_bind(Socket fd, void *addr, size_t addr_len)
{
if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_SOCKET) {
// TODO: errno
return -1;
}
if (addr_len != sizeof(current_process->desc[fd].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->desc[fd].bind, addr, addr_len);
return 0;
}
int sys_listen(Socket fd, int backlog)
{
if (current_process->desc[fd].type != DESCRIPTOR_TYPE_PRECONF_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].input))
polled[i].revents |= POLLIN;
}
if (polled[i].events & POLLOUT) {
if (!byte_queue_full(&current_process->desc[fd].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;
}
if (current_process->desc[fd].num_pending == 0) {
// No pending connections - would block in real system
// TODO: errno (EAGAIN/EWOULDBLOCK)
return BAD_SOCKET;
}
// Get the first pending connection descriptor
Socket new_fd = current_process->desc[fd].pending_connections[0];
// Remove from pending queue
current_process->desc[fd].num_pending--;
for (int i = 0; i < current_process->desc[fd].num_pending; i++) {
current_process->desc[fd].pending_connections[i] =
current_process->desc[fd].pending_connections[i + 1];
}
// Fill in peer address if requested
if (addr != NULL && addr_len != NULL) {
int peer_process = current_process->desc[new_fd].peer_process;
int peer_fd = current_process->desc[new_fd].peer_fd;
if (peer_process > -1 && peer_fd > -1) {
struct sockaddr_in *peer_addr = (struct sockaddr_in *)addr;
*peer_addr = processes[peer_process].desc[peer_fd].bind;
*addr_len = sizeof(struct sockaddr_in);
}
}
return new_fd;
}
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;
}
if (optval == NULL) {
// TODO: errno
return -1;
}
// In our simulation, all connections succeed immediately
// and we don't track socket errors, so always return 0
*(int *)optval = 0;
*optlen = sizeof(int);
return 0;
}
default:
// Unsupported socket option
// TODO: errno (ENOPROTOOPT)
return -1;
}
}
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].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].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;
}
if (addr_len != sizeof(struct sockaddr_in)) {
// TODO: errno
return -1;
}
struct sockaddr_in *target_addr = (struct sockaddr_in *)addr;
// Find the process with a listen socket on this address
int target_process_idx = -1;
Socket target_listen_fd = BAD_SOCKET;
for (int i = 0; i < num_processes; i++) {
for (int j = 0; j < MAX_DESCRIPTORS; j++) {
Descriptor *desc = &processes[i].desc[j];
if (desc->type == DESCRIPTOR_TYPE_LISTEN_SOCKET && !desc->no_bind) {
if (desc->bind.sin_port == target_addr->sin_port &&
(desc->bind.sin_addr.s_addr == target_addr->sin_addr.s_addr ||
desc->bind.sin_addr.s_addr == INADDR_ANY)) {
target_process_idx = i;
target_listen_fd = j;
goto found;
}
}
}
}
found:
if (target_process_idx < 0) {
// No listener found - connection refused
// TODO: errno (ECONNREFUSED)
return -1;
}
// Create a new socket in the target process for the accepted connection
Socket accept_fd = 0;
while (accept_fd < MAX_DESCRIPTORS &&
processes[target_process_idx].desc[accept_fd].type != DESCRIPTOR_TYPE_EMPTY)
accept_fd++;
if (accept_fd >= MAX_DESCRIPTORS) {
// TODO: errno
return -1;
}
// Check if pending queue is full
if (processes[target_process_idx].desc[target_listen_fd].num_pending >= MAX_PENDING_CONNECTIONS) {
// TODO: errno (ECONNREFUSED or EAGAIN)
return -1;
}
// Initialize byte queues for both ends of the connection
byte_queue_init(&current_process->desc[fd].input, 1<<16);
byte_queue_init(&current_process->desc[fd].output, 1<<16);
byte_queue_init(&processes[target_process_idx].desc[accept_fd].input, 1<<16);
byte_queue_init(&processes[target_process_idx].desc[accept_fd].output, 1<<16);
// Set up the client socket
current_process->desc[fd].type = DESCRIPTOR_TYPE_CONNECTION_SOCKET;
int current_process_idx = current_process - processes;
current_process->desc[fd].peer_process = target_process_idx;
current_process->desc[fd].peer_fd = accept_fd;
// Set up the accepted socket
processes[target_process_idx].desc[accept_fd].type = DESCRIPTOR_TYPE_CONNECTION_SOCKET;
processes[target_process_idx].desc[accept_fd].no_bind = false;
processes[target_process_idx].desc[accept_fd].bind = *target_addr;
processes[target_process_idx].desc[accept_fd].peer_process = current_process_idx;
processes[target_process_idx].desc[accept_fd].peer_fd = fd;
processes[target_process_idx].desc[accept_fd].num_pending = 0;
// Add to pending connections queue
processes[target_process_idx].desc[target_listen_fd].pending_connections[
processes[target_process_idx].desc[target_listen_fd].num_pending++] = accept_fd;
// Connection succeeds immediately in simulation (no EINPROGRESS needed)
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 (;;) {
// Step each process
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: handle error
}
}
// Process network transit: move data from output queues to peer input queues
for (int i = 0; i < num_processes; i++) {
for (int fd = 0; fd < MAX_DESCRIPTORS; fd++) {
Descriptor *desc = &processes[i].desc[fd];
if (desc->type != DESCRIPTOR_TYPE_CONNECTION_SOCKET)
continue;
if (desc->peer_process < 0 || desc->peer_fd < 0)
continue;
// Get peer descriptor
Descriptor *peer = &processes[desc->peer_process].desc[desc->peer_fd];
// Transfer data from this socket's output to peer's input
ByteView output_data = byte_queue_read_buf(&desc->output);
if (output_data.len > 0) {
// Get available space in peer's input queue
ByteView peer_input_space = byte_queue_write_buf(&peer->input);
// Transfer as much as possible
size_t transfer_size = output_data.len;
if (transfer_size > peer_input_space.len)
transfer_size = peer_input_space.len;
if (transfer_size > 0) {
memcpy(peer_input_space.ptr, output_data.ptr, transfer_size);
byte_queue_write_ack(&peer->input, transfer_size);
byte_queue_read_ack(&desc->output, transfer_size);
}
}
}
}
}
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