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

Major refactoring

Browse Source
This commit is contained in:
cozis
2026-02-25 10:44:43 +01:00
parent bac3477991
commit ef1d65ad2b
45 changed files with 5404 additions and 2120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic.c
-235
View File
@@ -1,235 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <stdint.h>
#include <quakey.h>
#include "basic.h"
bool streq(string s1, string s2)
{
if (s1.len != s2.len)
return false;
for (int i = 0; i < s1.len; i++)
if (s1.ptr[i] != s2.ptr[i])
return false;
return true;
}
// Returns the current time in nanoseconds since
// an unspecified time in the past (useful to calculate
// elapsed time intervals)
Time get_current_time(void)
{
#ifdef _WIN32
{
int64_t count;
int64_t freq;
int ok;
ok = QueryPerformanceCounter((LARGE_INTEGER*) &count);
if (!ok) return INVALID_TIME;
ok = QueryPerformanceFrequency((LARGE_INTEGER*) &freq);
if (!ok) return INVALID_TIME;
uint64_t res = 1000000000 * (double) count / freq;
return res;
}
#else
{
struct timespec time;
if (clock_gettime(CLOCK_REALTIME, &time))
return INVALID_TIME;
uint64_t res;
uint64_t sec = time.tv_sec;
if (sec > UINT64_MAX / 1000000000)
return INVALID_TIME;
res = sec * 1000000000;
uint64_t nsec = time.tv_nsec;
if (res > UINT64_MAX - nsec)
return INVALID_TIME;
res += nsec;
return res;
}
#endif
}
void nearest_deadline(Time *a, Time b)
{
if (*a == INVALID_TIME || *a > b)
*a = b;
}
int deadline_to_timeout(Time deadline, Time current_time)
{
if (deadline == INVALID_TIME)
return -1;
if (deadline <= current_time)
return 0;
// Round up so sub-millisecond deadlines become 1ms, not 0
return CEIL(deadline - current_time, 1000000);
}
bool getargb(int argc, char **argv, char *name)
{
for (int i = 0; i < argc; i++)
if (!strcmp(argv[i], name))
return true;
return false;
}
string getargs(int argc, char **argv, char *name, char *fallback)
{
for (int i = 0; i < argc; i++)
if (!strcmp(argv[i], name)) {
i++;
if (i == argc)
break;
return (string) { argv[i], strlen(argv[i]) };
}
return (string) { fallback, strlen(fallback) };
}
int getargi(int argc, char **argv, char *name, int fallback)
{
for (int i = 0; i < argc; i++)
if (!strcmp(argv[i], name)) {
i++;
if (i == argc)
break;
errno = 0;
char *end;
long val = strtol(argv[i], &end, 10);
if (end == argv[i] || *end != '\0' || errno == ERANGE)
break;
if (val < INT_MIN || val > INT_MAX)
break;
return (int) val;
}
return fallback;
}
void append_hex_as_str(char *out, SHA256 hash)
{
char table[] = "0123456789abcdef";
for (int i = 0; i < (int) sizeof(hash); i++) {
out[(i << 1) + 0] = table[(uint8_t) hash.data[i] >> 4];
out[(i << 1) + 1] = table[(uint8_t) hash.data[i] & 0xF];
}
}
// TODO: check this function
bool addr_lower(Address a, Address b)
{
if (a.is_ipv4) {
if (!b.is_ipv4)
return true;
if (a.ipv4.data < b.ipv4.data)
return true;
if (a.ipv4.data == b.ipv4.data &&
a.port < b.port)
return true;
return false;
} else {
if (b.is_ipv4)
return false;
for (int i = 0; i < 8; i++) {
if (a.ipv6.data[i] < b.ipv6.data[i])
return true;
if (a.ipv6.data[i] > b.ipv6.data[i])
return false;
}
if (a.port < b.port)
return true;
return false;
}
}
bool addr_eql(Address a, Address b)
{
if (a.is_ipv4 != b.is_ipv4)
return false;
if (a.port != b.port)
return false;
if (a.is_ipv4) {
if (memcmp(&a.ipv4, &b.ipv4, sizeof(a.ipv4)))
return false;
} else {
if (memcmp(&a.ipv6, &b.ipv6, sizeof(a.ipv6)))
return false;
}
return true;
}
int parse_addr_arg(char *arg, Address *out)
{
int len = strlen(arg);
int i = 0;
while (i < len && arg[i] != ':')
i++;
if (i == len)
return -1; // No ':' character.
arg[i] = '\0';
IPv4 ipv4;
int ret = inet_pton(AF_INET, arg, &ipv4);
arg[i] = ':';
if (ret != 1)
return -1;
errno = 0;
ret = atoi(arg + i + 1);
if (ret == 0 && errno != 0)
return -1;
out->ipv4 = ipv4;
out->is_ipv4 = true;
out->port = ret;
return 0;
}
void addr_sort(Address *addrs, int count)
{
for (int i = 0; i < count; i++) {
int k = i; // Index of the lowest address in [i, num_nodes-1]
for (int j = i+1; j < count; j++) {
if (addr_lower(addrs[j], addrs[k]))
k = j;
}
Address tmp = addrs[i];
addrs[i] = addrs[k];
addrs[k] = tmp;
}
}
src/basic.h
-62
View File
@@ -1,62 +0,0 @@
#ifndef BASIC_INCLUDED
#define BASIC_INCLUDED
#include <stdint.h>
#include <stdbool.h>
typedef struct {
char data[32];
} SHA256;
typedef struct {
uint32_t data;
} IPv4;
typedef struct {
uint16_t data[8];
} IPv6;
typedef struct {
union {
IPv4 ipv4;
IPv6 ipv6;
};
bool is_ipv4;
uint16_t port;
} Address;
typedef struct {
char *ptr;
int len;
} string;
typedef uint64_t Time;
#define INVALID_TIME ((Time) -1)
#define S(X) ((string) { (X), (int) sizeof(X)-1 })
#define MIN(X, Y) ((X) < (Y) ? (X) : (Y))
#define MAX(X, Y) ((X) > (Y) ? (X) : (Y))
#define CEIL(X, Y) (((X) + (Y) - 1) / (Y))
#define UNREACHABLE __builtin_trap();
bool streq(string s1, string s2);
Time get_current_time(void);
void nearest_deadline(Time *a, Time b);
int deadline_to_timeout(Time deadline, Time current_time);
bool getargb(int argc, char **argv, char *name);
string getargs(int argc, char **argv, char *name, char *fallback);
int getargi(int argc, char **argv, char *name, int fallback);
void append_hex_as_str(char *out, SHA256 hash);
bool addr_eql(Address a, Address b);
bool addr_lower(Address a, Address b);
int parse_addr_arg(char *arg, Address *out);
void addr_sort(Address *addrs, int count);
#endif // BASIC_INCLUDED
src/byte_queue.c
-306
View File
@@ -1,306 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <quakey.h>
#include <stdint.h>
#include <assert.h>
#include "byte_queue.h"
// This is the implementation of a byte queue useful
// for systems that need to process engs of bytes.
//
// It features sticky errors, a zero-copy interface,
// and a safe mechanism to patch previously written
// bytes.
//
// Only up to 4GB of data can be stored at once.
// Initialize the queue
void byte_queue_init(ByteQueue *queue, uint32_t limit)
{
queue->flags = 0;
queue->head = 0;
queue->size = 0;
queue->used = 0;
queue->curs = 0;
queue->limit = limit;
queue->data = NULL;
queue->read_target = NULL;
}
// Deinitialize the queue
void byte_queue_free(ByteQueue *queue)
{
if (queue->read_target) {
if (queue->read_target != queue->data)
free(queue->read_target);
queue->read_target = NULL;
queue->read_target_size = 0;
}
free(queue->data);
queue->data = NULL;
}
int byte_queue_error(ByteQueue *queue)
{
return queue->flags & BYTE_QUEUE_ERROR;
}
int byte_queue_empty(ByteQueue *queue)
{
return queue->used == 0;
}
int byte_queue_full(ByteQueue *queue)
{
return queue->used == queue->limit;
}
// Start a read operation on the queue.
//
// This function returnes the pointer to the memory region containing the bytes
// to read. Callers can't read more than [*len] bytes from it. To complete the
// read, the [byte_queue_read_ack] function must be called with the number of
// bytes that were acknowledged by the caller.
//
// Note:
// - You can't have more than one pending read.
ByteView byte_queue_read_buf(ByteQueue *queue)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return (ByteView) {NULL, 0};
assert((queue->flags & BYTE_QUEUE_READ) == 0);
queue->flags |= BYTE_QUEUE_READ;
queue->read_target = queue->data;
queue->read_target_size = queue->size;
if (queue->data == NULL)
return (ByteView) {NULL, 0};
return (ByteView) { queue->data + queue->head, queue->used };
}
// Complete a previously started operation on the queue.
void byte_queue_read_ack(ByteQueue *queue, uint32_t num)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return;
if ((queue->flags & BYTE_QUEUE_READ) == 0)
return;
queue->flags &= ~BYTE_QUEUE_READ;
assert((uint32_t) num <= queue->used);
queue->head += (uint32_t) num;
queue->used -= (uint32_t) num;
queue->curs += (uint32_t) num;
if (queue->read_target) {
if (queue->read_target != queue->data)
free(queue->read_target);
queue->read_target = NULL;
queue->read_target_size = 0;
}
}
ByteView byte_queue_write_buf(ByteQueue *queue)
{
if ((queue->flags & BYTE_QUEUE_ERROR) || queue->data == NULL)
return (ByteView) {NULL, 0};
assert((queue->flags & BYTE_QUEUE_WRITE) == 0);
queue->flags |= BYTE_QUEUE_WRITE;
return (ByteView) {
queue->data + (queue->head + queue->used),
queue->size - (queue->head + queue->used),
};
}
void byte_queue_write_ack(ByteQueue *queue, uint32_t num)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return;
if ((queue->flags & BYTE_QUEUE_WRITE) == 0)
return;
queue->flags &= ~BYTE_QUEUE_WRITE;
queue->used += num;
}
// Sets the minimum capacity for the next write operation
// and returns 1 if the content of the queue was moved, else
// 0 is returned.
//
// You must not call this function while a write is pending.
// In other words, you must do this:
//
// byte_queue_write_setmincap(queue, mincap);
// dst = byte_queue_write_buf(queue, &cap);
// ...
// byte_queue_write_ack(num);
//
// And NOT this:
//
// dst = byte_queue_write_buf(queue, &cap);
// byte_queue_write_setmincap(queue, mincap); <-- BAD
// ...
// byte_queue_write_ack(num);
//
int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap)
{
// Sticky error
if (queue->flags & BYTE_QUEUE_ERROR)
return 0;
// In general, the queue's contents look like this:
//
// size
// v
// [___xxxxxxxxxxxx________]
// ^ ^ ^
// 0 head head + used
//
// This function needs to make sure that at least [mincap]
// bytes are available on the right side of the content.
//
// We have 3 cases:
//
// 1) If there is enough memory already, this function doesn't
// need to do anything.
//
// 2) If there isn't enough memory on the right but there is
// enough free memory if we cound the left unused region,
// then the content is moved back to the
// start of the buffer.
//
// 3) If there isn't enough memory considering both sides, this
// function needs to allocate a new buffer.
//
// If there are pending read or write operations, the application
// is holding pointers to the buffer, so we need to make sure
// to not invalidate them. The only real problem is pending reads
// since this function can only be called before starting a write
// opearation.
//
// To avoid invalidating the read pointer when we allocate a new
// buffer, we don't free the old buffer. Instead, we store the
// pointer in the "old" field so that the read ack function can
// free it.
//
// To avoid invalidating the pointer when we are moving back the
// content since there is enough memory at the start of the buffer,
// we just avoid that. Even if there is enough memory considering
// left and right free regions, we allocate a new buffer.
assert((queue->flags & BYTE_QUEUE_WRITE) == 0);
uint32_t total_free_space = queue->size - queue->used;
uint32_t free_space_after_data = queue->size - queue->used - queue->head;
int moved = 0;
if (free_space_after_data < mincap) {
if (total_free_space < mincap || (queue->read_target == queue->data)) {
// Resize required
if (queue->used + mincap > queue->limit) {
queue->flags |= BYTE_QUEUE_ERROR;
return 0;
}
uint32_t size;
if (queue->size > UINT32_MAX / 2)
size = UINT32_MAX;
else
size = 2 * queue->size;
if (size < queue->used + mincap)
size = queue->used + mincap;
if (size > queue->limit)
size = queue->limit;
uint8_t *data = malloc(size);
if (!data) {
queue->flags |= BYTE_QUEUE_ERROR;
return 0;
}
if (queue->used > 0)
memcpy(data, queue->data + queue->head, queue->used);
if (queue->read_target != queue->data)
free(queue->data);
queue->data = data;
queue->head = 0;
queue->size = size;
} else {
// Move required
memmove(queue->data, queue->data + queue->head, queue->used);
queue->head = 0;
}
moved = 1;
}
return moved;
}
void byte_queue_write(ByteQueue *queue, void *ptr, uint32_t len)
{
byte_queue_write_setmincap(queue, len);
ByteView dst = byte_queue_write_buf(queue);
if (dst.ptr) {
memcpy(dst.ptr, ptr, len);
byte_queue_write_ack(queue, len);
}
}
ByteQueueOffset byte_queue_offset(ByteQueue *queue)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return (ByteQueueOffset) { 0 };
return (ByteQueueOffset) { queue->curs + queue->used };
}
void byte_queue_patch(ByteQueue *queue, ByteQueueOffset off,
void *src, uint32_t len)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return;
// Check that the offset is in range
assert(off >= queue->curs && off - queue->curs < queue->used);
// Check that the length is in range
assert(len <= queue->used - (off - queue->curs));
// Perform the patch
uint8_t *dst = queue->data + queue->head + (off - queue->curs);
memcpy(dst, src, len);
}
uint32_t byte_queue_size_from_offset(ByteQueue *queue, ByteQueueOffset off)
{
return queue->curs + queue->used - off;
}
void byte_queue_remove_from_offset(ByteQueue *queue, ByteQueueOffset offset)
{
if (queue->flags & BYTE_QUEUE_ERROR)
return;
uint64_t num = (queue->curs + queue->used) - offset;
assert(num <= queue->used);
queue->used -= num;
}
src/byte_queue.h
-53
View File
@@ -1,53 +0,0 @@
#ifndef BYTE_QUEUE_INCLUDED
#define BYTE_QUEUE_INCLUDED
#include <stddef.h>
#include "basic.h"
typedef struct {
uint8_t *ptr;
size_t len;
} ByteView;
typedef struct {
uint64_t curs;
uint8_t* data;
uint32_t head;
uint32_t size;
uint32_t used;
uint32_t limit;
uint8_t* read_target;
uint32_t read_target_size;
int flags;
} ByteQueue;
typedef uint64_t ByteQueueOffset;
enum {
BYTE_QUEUE_ERROR = 1 << 0,
BYTE_QUEUE_READ = 1 << 1,
BYTE_QUEUE_WRITE = 1 << 2,
};
void byte_queue_init(ByteQueue *queue, uint32_t limit);
void byte_queue_free(ByteQueue *queue);
int byte_queue_error(ByteQueue *queue);
int byte_queue_empty(ByteQueue *queue);
int byte_queue_full(ByteQueue *queue);
ByteView byte_queue_read_buf(ByteQueue *queue);
void byte_queue_read_ack(ByteQueue *queue, uint32_t num);
ByteView byte_queue_write_buf(ByteQueue *queue);
void byte_queue_write_ack(ByteQueue *queue, uint32_t num);
int byte_queue_write_setmincap(ByteQueue *queue, uint32_t mincap);
void byte_queue_write(ByteQueue *queue, void *ptr, uint32_t len);
ByteQueueOffset byte_queue_offset(ByteQueue *queue);
void byte_queue_patch(ByteQueue *queue, ByteQueueOffset off, void *src, uint32_t len);
uint32_t byte_queue_size_from_offset(ByteQueue *queue, ByteQueueOffset off);
void byte_queue_remove_from_offset(ByteQueue *queue, ByteQueueOffset offset);
#endif // BYTE_QUEUE_INCLUDED
src/chunk_store.c
+1 -1
View File
@@ -6,7 +6,7 @@
#include <stdio.h>
#include "chunk_store.h"
#include "file_system.h"
#include <lib/file_system.h>
// Build the full path for a chunk: "base_path/HEX_HASH"
// SHA256 hex = 64 chars. Returns string pointing into buf.
src/chunk_store.h
+1 -1
View File
@@ -4,7 +4,7 @@
#include <stdint.h>
#include <stdbool.h>
#include "basic.h"
#include <lib/basic.h>
typedef struct {
char base_path[256];
src/client.c
+37 -98
View File
@@ -6,16 +6,17 @@
#include <stdint.h>
#include <stdlib.h>
#include <assert.h>
#include <poll.h>
#include "tcp.h"
#include "basic.h"
#include <lib/basic.h>
#include "config.h"
#include "metadata.h"
#include "server.h"
#include <toastyfs.h>
#include <stdio.h>
#define POLL_CAPACITY (NODE_LIMIT * 2 + 4)
typedef enum {
STEP_IDLE,
STEP_STORE_CHUNK,
@@ -45,7 +46,7 @@ typedef struct {
struct ToastyFS {
TCP tcp;
MessageSystem msys;
Address server_addrs[NODE_LIMIT];
int num_servers;
@@ -74,8 +75,6 @@ struct ToastyFS {
int put_data_len;
};
// ---- Client logging infrastructure (mirrors server node_log) ----
#define TIME_FMT "%7.3fs"
#define TIME_VAL(t) ((double)(t) / 1000000000.0)
@@ -147,7 +146,7 @@ ToastyFS *toastyfs_init(uint64_t client_id, char **addrs, int num_addrs)
tfs->num_servers = num_addrs;
addr_sort(tfs->server_addrs, tfs->num_servers);
if (tcp_context_init(&tfs->tcp) < 0) {
if (message_system_init(&tfs->msys, tfs->server_addrs, num_addrs) < 0) {
free(tfs);
return NULL;
}
@@ -161,7 +160,7 @@ ToastyFS *toastyfs_init(uint64_t client_id, char **addrs, int num_addrs)
void toastyfs_free(ToastyFS *tfs)
{
tcp_context_free(&tfs->tcp);
message_system_free(&tfs->msys);
free(tfs->put_data);
}
@@ -203,40 +202,6 @@ static int leader_idx(ToastyFS *tfs)
return tfs->view_number % tfs->num_servers;
}
static void send_message_to_server(ToastyFS *tfs, int server_idx, MessageHeader *msg)
{
ByteQueue *output;
int conn_idx = tcp_index_from_tag(&tfs->tcp, server_idx);
if (conn_idx < 0) {
if (tcp_connect(&tfs->tcp, tfs->server_addrs[server_idx], server_idx, &output) < 0)
return;
} else {
output = tcp_output_buffer(&tfs->tcp, conn_idx);
if (output == NULL)
return;
}
byte_queue_write(output, msg, msg->length);
}
static void send_message_to_server_ex(ToastyFS *tfs, int server_idx,
MessageHeader *msg, void *extra, int extra_len)
{
ByteQueue *output;
int conn_idx = tcp_index_from_tag(&tfs->tcp, server_idx);
if (conn_idx < 0) {
if (tcp_connect(&tfs->tcp, tfs->server_addrs[server_idx], server_idx, &output) < 0)
return;
} else {
output = tcp_output_buffer(&tfs->tcp, conn_idx);
if (output == NULL)
return;
}
byte_queue_write(output, msg, msg->length - extra_len);
byte_queue_write(output, extra, extra_len);
}
static int begin_transfers(ToastyFS *tfs)
{
// Count started transfers
@@ -264,7 +229,7 @@ static int begin_transfers(ToastyFS *tfs)
.hash = tfs->transfers[i].hash,
.size = tfs->transfers[i].size,
};
send_message_to_server_ex(tfs, tfs->transfers[i].location,
send_message_ex(&tfs->msys, tfs->transfers[i].location,
&msg.base, tfs->transfers[i].data, tfs->transfers[i].size);
} else {
FetchChunkMessage msg = {
@@ -276,7 +241,7 @@ static int begin_transfers(ToastyFS *tfs)
.hash = tfs->transfers[i].hash,
.sender_idx = -1, // Client (not a peer server)
};
send_message_to_server(tfs, tfs->transfers[i].location, &msg.base);
send_message(&tfs->msys, tfs->transfers[i].location, &msg.base);
}
tfs->transfers[i].state = TRANSFER_STARTED;
@@ -340,7 +305,7 @@ static void replay_request(ToastyFS *tfs)
msg.oper.chunks[i].hash = tfs->chunks[i];
msg.oper.chunks[i].size = tfs->chunk_sizes[i];
}
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
}
break;
case STEP_DELETE:
@@ -359,7 +324,7 @@ static void replay_request(ToastyFS *tfs)
};
memcpy(msg.oper.bucket, tfs->bucket, META_BUCKET_MAX);
memcpy(msg.oper.key, tfs->key, META_KEY_MAX);
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
}
break;
case STEP_GET:
@@ -373,7 +338,7 @@ static void replay_request(ToastyFS *tfs)
};
memcpy(msg.bucket, tfs->bucket, META_BUCKET_MAX);
memcpy(msg.key, tfs->key, META_KEY_MAX);
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
}
break;
default:
@@ -381,11 +346,8 @@ static void replay_request(ToastyFS *tfs)
}
}
static int process_message(ToastyFS *tfs,
int conn_idx, uint8_t type, ByteView msg)
static int process_message(ToastyFS *tfs, uint16_t type, ByteView msg)
{
(void) conn_idx;
switch (tfs->step) {
case STEP_FETCH_CHUNK:
{
@@ -477,7 +439,7 @@ static int process_message(ToastyFS *tfs,
msg.oper.chunks[i].hash = tfs->chunks[i];
msg.oper.chunks[i].size = tfs->chunk_sizes[i];
}
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
tfs->step = STEP_COMMIT;
client_log(tfs, "SEND COMMIT_PUT", "key=%s chunks=%d req=%lu",
tfs->key, tfs->num_chunks, tfs->request_id);
@@ -513,8 +475,8 @@ static int process_message(ToastyFS *tfs,
} else if (type == MESSAGE_TYPE_REPLY) {
ReplyMessage reply;
if (msg.len != sizeof(ReplyMessage))
VsrReplyMessage reply;
if (msg.len != sizeof(VsrReplyMessage))
return -1;
memcpy(&reply, msg.ptr, sizeof(reply));
@@ -564,8 +526,8 @@ static int process_message(ToastyFS *tfs,
} else if (type == MESSAGE_TYPE_REPLY) {
ReplyMessage reply;
if (msg.len != sizeof(ReplyMessage))
VsrReplyMessage reply;
if (msg.len != sizeof(VsrReplyMessage))
return -1;
memcpy(&reply, msg.ptr, sizeof(reply));
@@ -675,44 +637,17 @@ static int process_message(ToastyFS *tfs,
void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pnum)
{
Event events[TCP_EVENT_CAPACITY];
int num_events = tcp_translate_events(&tfs->tcp, events, ctxs, pdata, pnum);
message_system_process_events(&tfs->msys, ctxs, pdata, pnum);
for (int i = 0; i < num_events; i++) {
if (events[i].type == EVENT_DISCONNECT) {
int conn_idx = events[i].conn_idx;
client_log(tfs, "DISCONNECT", "conn=%d", conn_idx);
tcp_close(&tfs->tcp, conn_idx);
continue;
}
if (events[i].type != EVENT_MESSAGE)
continue;
int conn_idx = events[i].conn_idx;
for (;;) {
ByteView msg;
uint16_t msg_type;
int ret = tcp_next_message(&tfs->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
break;
if (ret < 0) {
tcp_close(&tfs->tcp, conn_idx);
break;
}
ret = process_message(tfs, conn_idx, msg_type, msg);
if (ret < 0) {
tcp_close(&tfs->tcp, conn_idx);
break;
}
tcp_consume_message(&tfs->tcp, conn_idx);
}
void *raw;
while ((raw = get_next_message(&tfs->msys)) != NULL) {
Message *header = (Message *)raw;
ByteView msg_view = { .ptr = raw, .len = header->length };
process_message(tfs, header->type, msg_view);
consume_message(&tfs->msys, raw);
}
// Check for operation timeout — retry the current operation if the
// Check for operation timeout -- retry the current operation if the
// deadline has passed (handles initial sends that were dropped because
// the TCP connection wasn't established yet, and unresponsive servers).
if (tfs->step != STEP_IDLE
@@ -758,9 +693,9 @@ int toastyfs_register_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, i
}
*timeout = deadline_to_timeout(deadline, now);
if (pcap < TCP_POLL_CAPACITY)
if (pcap < POLL_CAPACITY)
return -1;
return tcp_register_events(&tfs->tcp, ctxs, pdata);
return message_system_register_events(&tfs->msys, ctxs, pdata, pcap);
}
static void
@@ -866,7 +801,7 @@ int toastyfs_async_get(ToastyFS *tfs, char *key, int key_len)
tfs->step_time = get_current_time(); // TODO: Handle INVALID_TIME error
tfs->step = STEP_GET;
client_log(tfs, "ASYNC GET", "key=%s leader=%d", tfs->key, leader_idx(tfs));
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
return 0;
}
@@ -900,7 +835,7 @@ int toastyfs_async_delete(ToastyFS *tfs, char *key, int key_len)
tfs->step_time = get_current_time(); // TODO: Handle INVALID_TIME error
tfs->step = STEP_DELETE;
client_log(tfs, "ASYNC DELETE", "key=%s req=%lu leader=%d", tfs->key, tfs->request_id, leader_idx(tfs));
send_message_to_server(tfs, leader_idx(tfs), &msg.base);
send_message(&tfs->msys, leader_idx(tfs), &msg.base);
return 0;
}
@@ -1073,14 +1008,18 @@ ToastyFS_Result toastyfs_get_result(ToastyFS *tfs)
static int wait_until_result(ToastyFS *tfs, ToastyFS_Result *res)
{
for (;;) {
void *ctxs[TCP_POLL_CAPACITY];
struct pollfd arr[TCP_POLL_CAPACITY];
void *ctxs[POLL_CAPACITY];
struct pollfd arr[POLL_CAPACITY];
int poll_timeout;
int num = toastyfs_register_events(tfs, ctxs, arr, TCP_POLL_CAPACITY, &poll_timeout);
int num = toastyfs_register_events(tfs, ctxs, arr, POLL_CAPACITY, &poll_timeout);
if (num < 0)
return num;
#ifdef _WIN32
WSAPoll(arr, num, poll_timeout);
#else
poll(arr, num, poll_timeout);
#endif
toastyfs_process_events(tfs, ctxs, arr, num);
*res = toastyfs_get_result(tfs);
src/client_table.c
+2 -2
View File
@@ -28,7 +28,7 @@ ClientTableEntry *client_table_find(ClientTable *client_table, uint64_t client_i
return NULL;
}
int client_table_add(ClientTable *client_table, uint64_t client_id, uint64_t request_id, int conn_tag)
int client_table_add(ClientTable *client_table, uint64_t client_id, uint64_t request_id, void *pending_message)
{
if (client_table->count == client_table->capacity) {
int n = 2 * client_table->capacity;
@@ -44,7 +44,7 @@ int client_table_add(ClientTable *client_table, uint64_t client_id, uint64_t req
.client_id = client_id,
.last_request_id = request_id,
.pending = true,
.conn_tag = conn_tag,
.pending_message = pending_message,
};
return 0;
}
src/client_table.h
+2 -2
View File
@@ -13,7 +13,7 @@ typedef struct {
// the REQUEST. After a view change, a new leader
// may find stale entries with pending=false from
// a previous view when it was leader before.
int conn_tag;
void *pending_message; // Raw message pointer for deferred reply
} ClientTableEntry;
typedef struct {
@@ -25,7 +25,7 @@ typedef struct {
void client_table_init(ClientTable *client_table);
void client_table_free(ClientTable *client_table);
ClientTableEntry *client_table_find(ClientTable *client_table, uint64_t client_id);
int client_table_add(ClientTable *client_table, uint64_t client_id, uint64_t request_id, int conn_tag);
int client_table_add(ClientTable *client_table, uint64_t client_id, uint64_t request_id, void *pending_message);
int client_table_insert(ClientTable *client_table, uint64_t client_id, uint64_t request_id);
#endif // CLIENT_TABLE_INCLUDED
src/config.h
+2
View File
@@ -1,6 +1,8 @@
#ifndef CONFIG_INCLUDED
#define CONFIG_INCLUDED
#define CEIL(a, b) (((a) + (b) - 1) / (b))
#define NODE_LIMIT 32
#define FUTURE_LIMIT 32
#define HEARTBEAT_INTERVAL_SEC 1
src/file_system.c
-459
View File
@@ -1,459 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <stdint.h>
#include <quakey.h>
#include "file_system.h"
int rename_file_or_dir(string oldpath, string newpath);
bool file_exists(string path)
{
char zt[1<<10];
if (path.len >= (int) sizeof(zt))
return false;
memcpy(zt, path.ptr, path.len);
zt[path.len] = '\0';
#ifdef __linux__
return access(zt, F_OK) == 0;
#endif
#ifdef _WIN32
DWORD attrs = GetFileAttributesA(zt);
return attrs != INVALID_FILE_ATTRIBUTES;
#endif
}
int file_open(string path, Handle *fd)
{
#ifdef __linux__
char zt[1<<10];
if (path.len >= (int) sizeof(zt))
return -1;
memcpy(zt, path.ptr, path.len);
zt[path.len] = '\0';
int ret = open(zt, O_RDWR | O_CREAT, 0644);
if (ret < 0)
return -1;
*fd = (Handle) { (uint64_t) ret };
return 0;
#endif
#ifdef _WIN32
WCHAR wpath[MAX_PATH];
MultiByteToWideChar(CP_UTF8, 0, path.ptr, path.len, wpath, MAX_PATH);
wpath[path.len] = L'\0';
HANDLE h = CreateFileW(
wpath,
GENERIC_WRITE | GENERIC_READ,
0,
NULL,
OPEN_ALWAYS,
FILE_ATTRIBUTE_NORMAL | FILE_FLAG_WRITE_THROUGH,
NULL
);
if (h == INVALID_HANDLE_VALUE)
return -1;
*fd = (Handle) { (uint64_t) h };
return 0;
#endif
}
void file_close(Handle fd)
{
#ifdef __linux__
close((int) fd.data);
#endif
#ifdef _WIN32
CloseHandle((HANDLE) fd.data);
#endif
}
int file_truncate(Handle fd, size_t new_size)
{
#ifdef __linux__
if (ftruncate((int) fd.data, new_size) < 0)
return -1;
return 0;
#endif
#ifdef _WIN32
return -1; // TODO: Not implemented
#endif
}
int file_set_offset(Handle fd, int off)
{
#ifdef __linux__
off_t ret = lseek((int) fd.data, off, SEEK_SET);
if (ret < 0)
return -1;
return 0;
#endif
#ifdef _WIN32
LARGE_INTEGER distance;
distance.QuadPart = off;
if (!SetFilePointer((HANDLE) fd.data, distance.LowPart, &distance.HighPart, FILE_BEGIN))
if (GetLastError() != 0)
return -1;
return 0;
#endif
}
int file_get_offset(Handle fd, int *off)
{
#ifdef __linux__
off_t ret = lseek((int) fd.data, 0, SEEK_CUR);
if (ret < 0)
return -1;
*off = (int) ret;
return 0;
#endif
#ifdef _WIN32
DWORD pos = SetFilePointer((HANDLE) fd.data, 0, NULL, FILE_CURRENT);
if (pos == INVALID_SET_FILE_POINTER && GetLastError() != 0)
return -1;
*off = (int) pos;
return 0;
#endif
}
int file_lock(Handle fd)
{
#ifdef __linux__
if (flock((int) fd.data, LOCK_EX) < 0)
return -1;
return 0;
#endif
#ifdef _WIN32
if (!LockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD))
return -1;
return 0;
#endif
}
int file_unlock(Handle fd)
{
#ifdef __linux__
if (flock((int) fd.data, LOCK_UN) < 0)
return -1;
return 0;
#endif
#ifdef _WIN32
if (!UnlockFile((HANDLE) fd.data, 0, 0, MAXDWORD, MAXDWORD))
return -1;
return 0;
#endif
}
int file_sync(Handle fd)
{
#ifdef __linux__
if (fsync((int) fd.data) < 0)
return -1;
return 0;
#endif
#ifdef _WIN32
if (!FlushFileBuffers((HANDLE) fd.data))
return -1;
return 0;
#endif
}
int file_read(Handle fd, char *dst, int max)
{
#ifdef __linux__
return read((int) fd.data, dst, max);
#endif
#ifdef _WIN32
DWORD num;
if (!ReadFile((HANDLE) fd.data, dst, max, &num, NULL))
return -1;
if (num > INT_MAX)
return -1;
return num;
#endif
}
int file_write(Handle fd, char *src, int len)
{
#ifdef __linux__
return write((int) fd.data, src, len);
#endif
#ifdef _WIN32
DWORD num;
if (!WriteFile((HANDLE) fd.data, src, len, &num, NULL))
return -1;
if (num > INT_MAX)
return -1;
return num;
#endif
}
int file_size(Handle fd, size_t *len)
{
#ifdef __linux__
struct stat buf;
if (fstat((int) fd.data, &buf) < 0)
return -1;
if (buf.st_size < 0 || (uint64_t) buf.st_size > SIZE_MAX)
return -1;
*len = (size_t) buf.st_size;
return 0;
#endif
#ifdef _WIN32
LARGE_INTEGER buf;
if (!GetFileSizeEx((HANDLE) fd.data, &buf))
return -1;
if (buf.QuadPart < 0 || (uint64_t) buf.QuadPart > SIZE_MAX)
return -1;
*len = buf.QuadPart;
return 0;
#endif
}
int create_dir(string path)
{
char zt[PATH_MAX];
if (path.len >= (int) sizeof(zt))
return -1;
memcpy(zt, path.ptr, path.len);
zt[path.len] = '\0';
#ifdef _WIN32
if (_mkdir(zt) < 0)
return -1;
#else
if (mkdir(zt, 0766))
return -1;
#endif
return 0;
}
int rename_file_or_dir(string oldpath, string newpath)
{
char oldpath_zt[PATH_MAX];
if (oldpath.len >= (int) sizeof(oldpath_zt))
return -1;
memcpy(oldpath_zt, oldpath.ptr, oldpath.len);
oldpath_zt[oldpath.len] = '\0';
char newpath_zt[PATH_MAX];
if (newpath.len >= (int) sizeof(newpath_zt))
return -1;
memcpy(newpath_zt, newpath.ptr, newpath.len);
newpath_zt[newpath.len] = '\0';
if (rename(oldpath_zt, newpath_zt))
return -1;
return 0;
}
int remove_file_or_dir(string path)
{
char path_zt[PATH_MAX];
if (path.len >= (int) sizeof(path_zt))
return -1;
memcpy(path_zt, path.ptr, path.len);
path_zt[path.len] = '\0';
if (remove(path_zt))
return -1;
return 0;
}
int get_full_path(string path, char *dst)
{
char path_zt[PATH_MAX];
if (path.len >= (int) sizeof(path_zt))
return -1;
memcpy(path_zt, path.ptr, path.len);
path_zt[path.len] = '\0';
#ifdef __linux__
if (realpath(path_zt, dst) == NULL)
return -1;
#endif
#ifdef _WIN32
if (_fullpath(path_zt, dst, PATH_MAX) == NULL)
return -1;
#endif
size_t path_len = strlen(dst);
if (path_len > 0 && dst[path_len-1] == '/')
dst[path_len-1] = '\0';
return 0;
}
int file_read_all(string path, string *data)
{
Handle fd;
int ret = file_open(path, &fd);
if (ret < 0)
return -1;
size_t len;
ret = file_size(fd, &len);
if (ret < 0) {
file_close(fd);
return -1;
}
char *dst = malloc(len);
if (dst == NULL) {
file_close(fd);
return -1;
}
int copied = 0;
while ((size_t) copied < len) {
ret = file_read(fd, dst + copied, len - copied);
if (ret < 0) {
free(dst);
file_close(fd);
return -1;
}
copied += ret;
}
*data = (string) { dst, len };
file_close(fd);
return 0;
}
#ifdef _WIN32
int directory_scanner_init(DirectoryScanner *scanner, string path)
{
char pattern[PATH_MAX];
int ret = snprintf(pattern, sizeof(pattern), "%.*s\\*", path.len, path.ptr);
if (ret < 0 || ret >= (int) sizeof(pattern))
return -1;
scanner->handle = FindFirstFileA(pattern, &scanner->find_data);
if (scanner->handle == INVALID_HANDLE_VALUE) {
if (GetLastError() == ERROR_FILE_NOT_FOUND) {
scanner->done = true;
return 0;
}
return -1;
}
scanner->done = false;
scanner->first = true;
return 0;
}
int directory_scanner_next(DirectoryScanner *scanner, string *name)
{
if (scanner->done)
return 1;
if (!scanner->first) {
BOOL ok = FindNextFileA(scanner->handle, &scanner->find_data);
if (!ok) {
scanner->done = true;
if (GetLastError() == ERROR_NO_MORE_FILES)
return 1;
return -1;
}
} else {
scanner->first = false;
}
char *p = scanner->find_data.cFileName;
*name = (string) { p, strlen(p) };
return 0;
}
void directory_scanner_free(DirectoryScanner *scanner)
{
FindClose(scanner->handle);
}
#else
int directory_scanner_init(DirectoryScanner *scanner, string path)
{
char path_copy[PATH_MAX];
if (path.len >= PATH_MAX)
return -1;
memcpy(path_copy, path.ptr, path.len);
path_copy[path.len] = '\0';
scanner->d = opendir(path_copy);
if (scanner->d == NULL) {
scanner->done = true;
return -1;
}
scanner->done = false;
return 0;
}
int directory_scanner_next(DirectoryScanner *scanner, string *name)
{
if (scanner->done)
return 1;
scanner->e = readdir(scanner->d);
if (scanner->e == NULL) {
scanner->done = true;
return 1;
}
*name = (string) { scanner->e->d_name, strlen(scanner->e->d_name) };
return 0;
}
void directory_scanner_free(DirectoryScanner *scanner)
{
closedir(scanner->d);
}
#endif
int file_read_exact(Handle handle, char *dst, int len)
{
int copied = 0;
while (copied < len) {
int ret = file_read(handle, dst + copied, len - copied);
if (ret < 0)
return -1;
if (ret == 0)
return 0; // EOF
copied += ret;
}
return copied;
}
int file_write_exact(Handle handle, char *src, int len)
{
int copied = 0;
while (copied < len) {
int ret = file_write(handle, src + copied, len - copied);
if (ret < 0)
return -1;
copied += ret;
}
return 0;
}
src/file_system.h
-58
View File
@@ -1,58 +0,0 @@
#ifndef FILE_SYSTEM_INCLUDED
#define FILE_SYSTEM_INCLUDED
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <stdint.h>
#include <quakey.h>
#include "basic.h"
typedef struct {
uint64_t data;
} Handle;
#ifdef _WIN32
typedef struct {
HANDLE handle;
WIN32_FIND_DATA find_data;
bool first;
bool done;
} DirectoryScanner;
#else
typedef struct {
DIR *d;
struct dirent *e;
bool done;
} DirectoryScanner;
#endif
bool file_exists(string path);
int file_open(string path, Handle *fd);
void file_close(Handle fd);
int file_truncate(Handle fd, size_t new_size);
int file_set_offset(Handle fd, int off);
int file_get_offset(Handle fd, int *off);
int file_lock(Handle fd);
int file_unlock(Handle fd);
int file_sync(Handle fd);
int file_read(Handle fd, char *dst, int max);
int file_write(Handle fd, char *src, int len);
int file_size(Handle fd, size_t *len);
int file_write_atomic(string path, string content);
int create_dir(string path);
int rename_file_or_dir(string oldpath, string newpath);
int remove_file_or_dir(string path);
int get_full_path(string path, char *dst);
int file_read_all(string path, string *data);
int directory_scanner_init(DirectoryScanner *scanner, string path);
int directory_scanner_next(DirectoryScanner *scanner, string *name);
void directory_scanner_free(DirectoryScanner *scanner);
int file_read_exact(Handle handle, char *dst, int len);
int file_write_exact(Handle handle, char *src, int len);
#endif // FILE_SYSTEM_INCLUDED
src/http_proxy.c
+41
View File
@@ -0,0 +1,41 @@
#include "http_proxy.h"
int http_proxy_init(void *state, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout)
{
HTTPProxy *proxy = state;
if (http_server_init(&proxy->http_server) < 0)
return -1;
if (http_server_listen_tcp(&proxy->http_server, xxx, yyy) < 0)
return -1;
return 0;
}
int http_proxy_tick(void *state, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout)
{
http_server_process_events(&proxy->http_server, ctxs, pdata, *pnum);
HTTP_Request *request;
HTTP_ResponseBuilder builder;
while (http_server_next_request(&proxy->http_server, &request, &builder)) {
// TODO
http_response_builder_status(builder, 200);
http_response_builder_submit(builder);
}
*timeout = -1;
*pnum = http_server_register_events(&proxy->http_server, ctxs, pdata, pcap);
return 0;
}
int http_proxy_free(void *state)
{
HTTPProxy *proxy = state;
http_server_free(&proxy->http_server);
}
src/http_proxy.h
+23
View File
@@ -0,0 +1,23 @@
#ifndef HTTP_PROXY_INCLUDED
#define HTTP_PROXY_INCLUDED
#include <toastyfs.h>
#include <lib/http_server.h>
typedef struct {
HTTP_Server http_server;
ToastyFS *toastyfs;
} HTTPProxy;
struct pollfd;
int http_proxy_init(void *state, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout);
int http_proxy_tick(void *state, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout);
int http_proxy_free(void *state);
#endif // HTTP_PROXY_INCLUDED
src/invariant_checker.c
+14 -7
View File
@@ -9,16 +9,23 @@
#include "chunk_store.h"
#include <assert.h>
#include <stdio.h>
// Restore real allocators (see checker/linearizability.c for precedent).
#undef malloc
#undef realloc
#undef free
#include <stdlib.h>
#include <string.h>
// Forward declarations for quakey host context functions
// (we can't include quakey.h because it would re-mock malloc/realloc/free)
void quakey_enter_host(unsigned long long node);
void quakey_leave_host(void);
// These helpers are static in node.c; duplicated here for the checker.
static int self_idx(ServerState *state)
static int self_idx(Server *state)
{
for (int i = 0; i < state->num_nodes; i++)
if (addr_eql(state->node_addrs[i], state->self_addr))
@@ -26,12 +33,12 @@ static int self_idx(ServerState *state)
UNREACHABLE;
}
static int leader_idx(ServerState *state)
static int leader_idx(Server *state)
{
return state->view_number % state->num_nodes;
}
static bool is_leader(ServerState *state)
static bool is_leader(Server *state)
{
if (state->status == STATUS_RECOVERY)
return false;
@@ -72,7 +79,7 @@ void invariant_checker_free(InvariantChecker *ic)
free(ic->shadow_log);
}
void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_nodes,
void invariant_checker_run(InvariantChecker *ic, Server **nodes, int num_nodes,
unsigned long long *node_handles)
{
int min_commit = -1;
@@ -90,7 +97,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
}
for (int i = 0; i < num_nodes; i++) {
ServerState *n = nodes[i];
Server *n = nodes[i];
if (n == NULL || n->status == STATUS_RECOVERY)
continue;
@@ -158,7 +165,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
}
for (int i = 0; i < num_nodes; i++) {
ServerState *s = nodes[i];
Server *s = nodes[i];
if (s == NULL)
continue;
@@ -297,7 +304,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
// Phase 2: Append newly committed entries to the shadow log.
if (source_node_idx >= 0 && observed_max_commit > ic->shadow_count) {
ServerState *source = nodes[source_node_idx];
Server *source = nodes[source_node_idx];
assert(source->log.count >= observed_max_commit);
for (int k = ic->shadow_count; k < observed_max_commit; k++) {
src/main.c
+17 -13
View File
@@ -52,7 +52,6 @@ int main(int argc, char **argv)
.addrs = (char*[]) { "127.0.0.2" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
(void) quakey_spawn(quakey, config, "cli --server 127.0.0.4:8080 --server 127.0.0.5:8080 --server 127.0.0.6:8080");
}
@@ -68,7 +67,6 @@ int main(int argc, char **argv)
.addrs = (char*[]) { "127.0.0.3" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
(void) quakey_spawn(quakey, config, "cli --server 127.0.0.4:8080 --server 127.0.0.5:8080 --server 127.0.0.6:8080");
}
@@ -84,7 +82,6 @@ int main(int argc, char **argv)
.addrs = (char*[]) { "127.0.0.7" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
(void) quakey_spawn(quakey, config, "cli --server 127.0.0.4:8080 --server 127.0.0.5:8080 --server 127.0.0.6:8080");
}
@@ -93,14 +90,13 @@ int main(int argc, char **argv)
{
QuakeySpawn config = {
.name = "server1",
.state_size = sizeof(ServerState),
.state_size = sizeof(Server),
.init_func = server_init,
.tick_func = server_tick,
.free_func = server_free,
.addrs = (char*[]) { "127.0.0.4" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
node_1 = quakey_spawn(quakey, config, "nd --addr 127.0.0.4:8080 --peer 127.0.0.5:8080 --peer 127.0.0.6:8080");
}
@@ -109,14 +105,13 @@ int main(int argc, char **argv)
{
QuakeySpawn config = {
.name = "server2",
.state_size = sizeof(ServerState),
.state_size = sizeof(Server),
.init_func = server_init,
.tick_func = server_tick,
.free_func = server_free,
.addrs = (char*[]) { "127.0.0.5" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
node_2 = quakey_spawn(quakey, config, "nd --peer 127.0.0.4:8080 --addr 127.0.0.5:8080 --peer 127.0.0.6:8080");
}
@@ -125,14 +120,13 @@ int main(int argc, char **argv)
{
QuakeySpawn config = {
.name = "server3",
.state_size = sizeof(ServerState),
.state_size = sizeof(Server),
.init_func = server_init,
.tick_func = server_tick,
.free_func = server_free,
.addrs = (char*[]) { "127.0.0.6" },
.num_addrs = 1,
.disk_size = 10<<20,
.platform = QUAKEY_LINUX,
};
node_3 = quakey_spawn(quakey, config, "nd --peer 127.0.0.4:8080 --peer 127.0.0.5:8080 --addr 127.0.0.6:8080");
}
@@ -151,7 +145,7 @@ int main(int argc, char **argv)
quakey_schedule_one(quakey);
ServerState *arr[] = {
Server *arr[] = {
quakey_node_state(node_1),
quakey_node_state(node_2),
quakey_node_state(node_3),
@@ -182,7 +176,11 @@ int main(int argc, char **argv)
////////////////////////////////////////////////////////////////////
#ifdef MAIN_CLIENT
#ifdef _WIN32
#include <winsock2.h>
#else
#include <poll.h>
#endif
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
@@ -246,8 +244,14 @@ int main(int argc, char **argv)
////////////////////////////////////////////////////////////////////
#ifdef MAIN_SERVER
#ifdef _WIN32
#include <winsock2.h>
#else
#include <poll.h>
#endif
#include <stdio.h>
#include <stdlib.h>
#include "server.h"
#define POLL_CAPACITY 1024
@@ -256,11 +260,11 @@ int main(int argc, char **argv)
{
int ret;
// ServerState is ~40 MB (MetaStore holds 4096 ObjectMeta entries),
// Server is ~40 MB (MetaStore holds 4096 ObjectMeta entries),
// which exceeds the default stack size. Heap-allocate it.
ServerState *state = malloc(sizeof(ServerState));
Server *state = malloc(sizeof(Server));
if (state == NULL) {
fprintf(stderr, "Failed to allocate ServerState\n");
fprintf(stderr, "Failed to allocate Server\n");
return -1;
}
src/message.c
-81
View File
@@ -1,81 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <stdint.h>
#include <quakey.h>
#include "message.h"
bool binary_read(BinaryReader *reader, void *dst, int len)
{
if (reader->len - reader->cur < len)
return false;
if (dst)
memcpy(dst, reader->src + reader->cur, len);
reader->cur += len;
return true;
}
void message_writer_init(MessageWriter *writer, ByteQueue *output, uint16_t type)
{
uint16_t version = MESSAGE_VERSION;
uint32_t dummy = 0; // Dummy value
writer->output = output;
writer->start = byte_queue_offset(output);
byte_queue_write(output, &version, sizeof(version));
byte_queue_write(output, &type, sizeof(type));
writer->patch = byte_queue_offset(output);
byte_queue_write(output, &dummy, sizeof(dummy));
}
bool message_writer_free(MessageWriter *writer)
{
uint32_t length = byte_queue_size_from_offset(writer->output, writer->start);
byte_queue_patch(writer->output, writer->patch, &length, sizeof(length));
if (byte_queue_error(writer->output)) // TODO: is it possible to restore the state of the queue to before the failure?
return false;
return true;
}
void message_write(MessageWriter *writer, void *mem, int len)
{
byte_queue_write(writer->output, mem, len);
}
void message_write_u8(MessageWriter *writer, uint8_t value)
{
message_write(writer, &value, (int) sizeof(value));
}
void message_write_u32(MessageWriter *writer, uint32_t value)
{
message_write(writer, &value, (int) sizeof(value));
}
void message_write_hash(MessageWriter *writer, SHA256 value)
{
message_write(writer, &value, (int) sizeof(value));
}
int message_peek(ByteView msg, uint16_t *type, uint32_t *len)
{
if (msg.len < (int) sizeof(MessageHeader))
return 0;
MessageHeader header;
memcpy(&header, msg.ptr, sizeof(header));
// (We ignore endianess for now)
if (header.version != MESSAGE_VERSION)
return -1;
if (header.length > msg.len)
return 0;
if (type) *type = header.type;
if (len) *len = header.length;
return 1;
}
src/message.h
-46
View File
@@ -1,46 +0,0 @@
#ifndef MESSAGE_INCLUDED
#define MESSAGE_INCLUDED
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <stdint.h>
#include <quakey.h>
#include "basic.h"
#include "byte_queue.h"
#define MESSAGE_VERSION 1
typedef struct {
uint8_t *src;
int len;
int cur;
} BinaryReader;
typedef struct {
uint16_t version;
uint16_t type;
uint32_t length;
} MessageHeader;
typedef struct {
ByteQueue *output;
ByteQueueOffset start;
ByteQueueOffset patch;
} MessageWriter;
bool binary_read(BinaryReader *reader, void *dst, int len);
void message_writer_init(MessageWriter *writer, ByteQueue *output, uint16_t type);
bool message_writer_free(MessageWriter *writer);
void message_write(MessageWriter *writer, void *mem, int len);
void message_write_u8(MessageWriter *writer, uint8_t value);
void message_write_u32(MessageWriter *writer, uint32_t value);
void message_write_hash(MessageWriter *writer, SHA256 value);
int message_peek(ByteView msg, uint16_t *type, uint32_t *len);
void message_dump(FILE *stream, ByteView msg);
#endif // MESSAGE_INCLUDED
src/metadata.h
+1 -1
View File
@@ -4,7 +4,7 @@
#include <stdint.h>
#include <stdbool.h>
#include "basic.h"
#include <lib/basic.h>
typedef struct {
SHA256 hash;
src/random_client.c
+1 -4
View File
@@ -82,10 +82,7 @@ int random_client_init(void *state_, int argc, char **argv,
return -1;
state->started = false;
if (pcap < TCP_POLL_CAPACITY) {
fprintf(stderr, "Random client :: Not enough poll capacity\n");
return -1;
}
(void) pcap;
*pnum = toastyfs_register_events(state->tfs, ctxs, pdata, pcap, timeout);
*timeout = 0; // Ensure first tick fires immediately to start an operation
return 0;
src/random_client.h
+2 -2
View File
@@ -3,8 +3,8 @@
#include <toastyfs.h>
#include "tcp.h"
#include "basic.h"
#include <lib/tcp.h>
#include <lib/basic.h>
#include "config.h"
#include "metadata.h"
src/server.c
+252 -595
View File
File diff suppressed because it is too large Load Diff
src/server.h
+62 -62
View File
@@ -1,15 +1,16 @@
#ifndef NODE_INCLUDED
#define NODE_INCLUDED
#ifndef SERVER_INCLUDED
#define SERVER_INCLUDED
#include <lib/message.h>
#include "tcp.h"
#include "basic.h"
#include "message.h"
#include "log.h"
#include "config.h"
#include "metadata.h"
#include "chunk_store.h"
#include "client_table.h"
#define MESSAGE_VERSION 1
enum {
// Normal Protocol
@@ -48,107 +49,107 @@ enum {
};
typedef struct {
MessageHeader base;
Message base;
MetaOper oper;
uint64_t client_id;
uint64_t request_id;
} RequestMessage;
typedef struct {
MessageHeader base;
Message base;
MetaOper oper;
int sender_idx;
int log_index;
int commit_index;
int sender_idx;
int log_index;
int commit_index;
uint64_t view_number;
uint64_t client_id;
uint64_t request_id;
} PrepareMessage;
typedef struct {
MessageHeader base;
int sender_idx;
int log_index;
Message base;
int sender_idx;
int log_index;
uint64_t view_number;
} PrepareOKMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int sender_idx;
int commit_index;
int sender_idx;
int commit_index;
} CommitMessage;
typedef struct {
MessageHeader base;
bool rejected;
Message base;
bool rejected;
MetaResult result;
uint64_t request_id;
} ReplyMessage;
uint64_t request_id;
} VsrReplyMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int sender_idx;
int sender_idx;
} BeginViewChangeMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number; // The new view number
uint64_t old_view_number; // Last view number when replica was in normal status
int op_number; // Number of entries in the log
int commit_index;
int sender_idx;
int op_number; // Number of entries in the log
int commit_index;
int sender_idx;
// Followed by: LogEntry log[op_number]
} DoViewChangeMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int commit_index;
int op_number; // Number of log entries that follow
int commit_index;
int op_number; // Number of log entries that follow
// Followed by: LogEntry log[op_number]
} BeginViewMessage;
typedef struct {
MessageHeader base;
int sender_idx;
Message base;
int sender_idx;
uint64_t nonce;
} RecoveryMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int op_number;
int op_number;
uint64_t nonce;
int commit_index;
int sender_idx;
int commit_index;
int sender_idx;
} RecoveryResponseMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int op_number; // Requester's current log count
int sender_idx;
int op_number; // Requester's current log count
int sender_idx;
} GetStateMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
int op_number; // Number of log entries that follow
int commit_index;
int start_index; // Global log index of the first entry in the suffix
int op_number; // Number of log entries that follow
int commit_index;
int start_index; // Global log index of the first entry in the suffix
// Followed by: LogEntry log[op_number]
} NewStateMessage;
typedef struct {
MessageHeader base;
Message base;
uint64_t view_number;
} RedirectMessage;
// StoreChunk: client -> any server. Carries chunk data as trailing bytes.
typedef struct {
MessageHeader base;
Message base;
SHA256 hash;
uint32_t size;
// Followed by: uint8_t data[size]
@@ -156,21 +157,21 @@ typedef struct {
// StoreChunkAck: server -> client.
typedef struct {
MessageHeader base;
SHA256 hash;
bool success;
Message base;
SHA256 hash;
bool success;
} StoreChunkAckMessage;
// FetchChunk: client/server -> server. Request a chunk by hash.
typedef struct {
MessageHeader base;
SHA256 hash;
int sender_idx; // -1 if from a client
Message base;
SHA256 hash;
int sender_idx; // -1 if from a client
} FetchChunkMessage;
// FetchChunkResponse: server -> client/server. Chunk data as trailing bytes.
typedef struct {
MessageHeader base;
Message base;
SHA256 hash;
uint32_t size; // 0 if chunk not found
// Followed by: uint8_t data[size]
@@ -179,7 +180,7 @@ typedef struct {
// CommitPut: blob client -> leader. Commits metadata after chunks uploaded.
// Processed like a REQUEST (goes through VSR log).
typedef struct {
MessageHeader base;
Message base;
MetaOper oper;
uint64_t client_id;
uint64_t request_id;
@@ -187,14 +188,14 @@ typedef struct {
// GetBlob: client -> any server. Request metadata for a blob.
typedef struct {
MessageHeader base;
char bucket[META_BUCKET_MAX];
char key[META_KEY_MAX];
Message base;
char bucket[META_BUCKET_MAX];
char key[META_KEY_MAX];
} GetBlobMessage;
// GetBlobResponse: server -> client. Returns blob metadata.
typedef struct {
MessageHeader base;
Message base;
bool found;
uint64_t size;
SHA256 content_hash;
@@ -210,7 +211,7 @@ typedef enum {
typedef struct {
TCP tcp;
MessageSystem msys;
Address self_addr;
Address node_addrs[NODE_LIMIT];
@@ -219,10 +220,9 @@ typedef struct {
Status status;
ClientTable client_table;
int next_client_tag;
uint64_t view_number;
uint64_t last_normal_view; // Latest view where status was NORMAL
uint64_t last_normal_view; // Latest view where status was NORMAL
// These fields are used in recovery mode
uint32_t recovery_votes;
@@ -265,7 +265,7 @@ typedef struct {
// Set at each wakeup
Time now;
} ServerState;
} Server;
struct pollfd;
@@ -294,7 +294,7 @@ void invariant_checker_free(InvariantChecker *ic);
// node_handles: opaque QuakeyNode handles for entering host context.
// Pass NULL when running outside the simulation (real mode).
void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_nodes,
void invariant_checker_run(InvariantChecker *ic, Server **nodes, int num_nodes,
unsigned long long *node_handles);
#endif // NODE_INCLUDED
#endif // SERVER_INCLUDED
src/tcp.c
-504
View File
@@ -1,504 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <quakey.h>
#include <assert.h>
#include "tcp.h"
#include "message.h"
static int set_socket_blocking(SOCKET sock, bool value)
{
#ifdef _WIN32
u_long mode = !value;
if (ioctlsocket(sock, FIONBIO, &mode) == SOCKET_ERROR)
return -1;
#else
int flags = fcntl(sock, F_GETFL, 0);
if (flags < 0)
return -1;
if (value) flags &= ~O_NONBLOCK;
else flags |= O_NONBLOCK;
if (fcntl(sock, F_SETFL, flags) < 0)
return -1;
#endif
return 0;
}
static SOCKET create_listen_socket(Address addr)
{
SOCKET fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == INVALID_SOCKET)
return INVALID_SOCKET;
if (set_socket_blocking(fd, false) < 0) {
CLOSE_SOCKET(fd);
return INVALID_SOCKET;
}
#ifndef MAIN_SIMULATION
int reuse = 1;
setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse));
#endif
if (!addr.is_ipv4) {
assert(0); // TODO
}
struct sockaddr_in bind_buf;
bind_buf.sin_family = AF_INET;
bind_buf.sin_port = htons(addr.port);
memcpy(&bind_buf.sin_addr, &addr.ipv4, sizeof(IPv4));
if (bind(fd, (struct sockaddr*) &bind_buf, sizeof(bind_buf))) {
CLOSE_SOCKET(fd);
return INVALID_SOCKET;
}
int backlog = 32;
if (listen(fd, backlog) < 0) {
CLOSE_SOCKET(fd);
return INVALID_SOCKET;
}
return fd;
}
static int create_socket_pair(SOCKET *a, SOCKET *b)
{
#ifdef _WIN32
SOCKET sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
if (sock == INVALID_SOCKET)
return -1;
// Bind to loopback address with port 0 (dynamic port assignment)
struct sockaddr_in addr;
int addr_len = sizeof(addr);
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); // 127.0.0.1
addr.sin_port = 0; // Let system choose port
if (bind(sock, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
closesocket(sock);
return -1;
}
if (getsockname(sock, (struct sockaddr*)&addr, &addr_len) == SOCKET_ERROR) {
closesocket(sock);
return -1;
}
if (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
closesocket(sock);
return -1;
}
*a = sock;
*b = sock;
// Optional: Set socket to non-blocking mode
// This prevents send() from blocking if the receive buffer is full
u_long mode = 1;
ioctlsocket(sock, FIONBIO, &mode); // TODO: does this fail?
return 0;
#else
int fds[2];
if (pipe(fds) < 0)
return -1;
*a = fds[0];
*b = fds[1];
return 0;
#endif
}
static void close_socket_pair(SOCKET a, SOCKET b)
{
#ifdef _WIN32
closesocket(a);
(void) b;
#else
close(a);
close(b);
#endif
}
static void conn_init(Connection *conn, SOCKET fd, bool connecting)
{
conn->fd = fd;
conn->tag = -1;
conn->connecting = connecting;
conn->closing = false;
conn->msglen = 0;
byte_queue_init(&conn->input, 1<<30);
byte_queue_init(&conn->output, 1<<30);
}
static void conn_free(Connection *conn)
{
CLOSE_SOCKET(conn->fd);
byte_queue_free(&conn->input);
byte_queue_free(&conn->output);
}
static int conn_events(Connection *conn)
{
int events = 0;
if (conn->connecting)
events |= POLLOUT;
else {
assert(!byte_queue_full(&conn->input));
if (!conn->closing)
events |= POLLIN;
if (!byte_queue_empty(&conn->output))
events |= POLLOUT;
}
return events;
}
int tcp_context_init(TCP *tcp)
{
tcp->listen_fd = INVALID_SOCKET;
tcp->num_conns = 0;
if (create_socket_pair(&tcp->wait_fd, &tcp->signal_fd) < 0)
return -1;
return 0;
}
void tcp_context_free(TCP *tcp)
{
// Free all connection byte queues without closing sockets
// (sockets are managed by the simulation and will be cleaned up separately)
for (int i = 0; i < tcp->num_conns; i++) {
byte_queue_free(&tcp->conns[i].input);
byte_queue_free(&tcp->conns[i].output);
}
tcp->num_conns = 0;
if (tcp->listen_fd != INVALID_SOCKET)
CLOSE_SOCKET(tcp->listen_fd);
close_socket_pair(tcp->wait_fd, tcp->signal_fd);
}
int tcp_wakeup(TCP *tcp)
{
send(tcp->signal_fd, "0", 1, 0); // TODO: Handle error
return 0;
}
int tcp_index_from_tag(TCP *tcp, int tag)
{
for (int i = 0; i < tcp->num_conns; i++)
if (tcp->conns[i].tag == tag)
return i;
return -1;
}
int tcp_listen(TCP *tcp, Address addr)
{
SOCKET listen_fd = create_listen_socket(addr);
if (listen_fd == INVALID_SOCKET)
return -1;
tcp->listen_fd = listen_fd;
return 0;
}
int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type)
{
*msg = byte_queue_read_buf(&tcp->conns[conn_idx].input);
uint32_t len;
int ret = message_peek(*msg, type, &len);
// Invalid message?
if (ret < 0) {
byte_queue_read_ack(&tcp->conns[conn_idx].input, 0);
return -1;
}
// Still buffering header?
if (ret == 0) {
byte_queue_read_ack(&tcp->conns[conn_idx].input, 0);
if (byte_queue_full(&tcp->conns[conn_idx].input))
return -1;
return 0;
}
// Message received
assert(ret > 0);
msg->len = len;
tcp->conns[conn_idx].msglen = len;
return 1;
}
void tcp_consume_message(TCP *tcp, int conn_idx)
{
byte_queue_read_ack(&tcp->conns[conn_idx].input, tcp->conns[conn_idx].msglen);
tcp->conns[conn_idx].msglen = 0;
}
int tcp_register_events(TCP *tcp, void **contexts, struct pollfd *polled)
{
int num_polled = 0;
polled[num_polled].fd = tcp->wait_fd;
polled[num_polled].events = POLLIN;
polled[num_polled].revents = 0;
contexts[num_polled] = NULL;
num_polled++;
if (tcp->listen_fd != INVALID_SOCKET && tcp->num_conns < TCP_CONNECTION_LIMIT) {
polled[num_polled].fd = tcp->listen_fd;
polled[num_polled].events = POLLIN;
polled[num_polled].revents = 0;
contexts[num_polled] = NULL;
num_polled++;
}
for (int i = 0; i < tcp->num_conns; i++) {
int events = conn_events(&tcp->conns[i]);
if (events) {
polled[num_polled].fd = tcp->conns[i].fd;
polled[num_polled].events = events;
polled[num_polled].revents = 0;
contexts[num_polled] = &tcp->conns[i];
num_polled++;
}
}
return num_polled;
}
// The "events" array must be an array of capacity TCP_EVENT_CAPACITY,
// while "contexts" and "polled" must have capacity TCP_POLL_CAPACITY.
int tcp_translate_events(TCP *tcp, Event *events, void **contexts, struct pollfd *polled, int num_polled)
{
bool removed[TCP_POLL_CAPACITY];
for (int i = 0; i < TCP_POLL_CAPACITY; i++)
removed[i] = false;
int num_events = 0;
for (int i = 1; i < num_polled; i++) {
if (polled[i].fd == tcp->wait_fd) {
if (polled[i].revents & POLLIN) {
char buf[100];
recv(tcp->wait_fd, buf, sizeof(buf), 0); // TODO: Make sure all bytes are consumed
events[num_events++] = (Event) { EVENT_WAKEUP, -1, -1 };
}
} else if (polled[i].fd == tcp->listen_fd) {
assert(contexts[i] == NULL);
if (polled[i].revents & POLLIN) {
SOCKET new_fd = accept(tcp->listen_fd, NULL, NULL);
if (new_fd != INVALID_SOCKET) {
if (set_socket_blocking(new_fd, false) < 0)
CLOSE_SOCKET(new_fd);
else {
conn_init(&tcp->conns[tcp->num_conns++], new_fd, false);
events[num_events++] = (Event) { EVENT_CONNECT, tcp->num_conns-1, tcp->conns[tcp->num_conns-1].tag };
}
}
}
removed[i] = false;
} else {
Connection *conn = contexts[i];
bool defer_close = false;
bool defer_ready = false;
if (conn->connecting) {
// Check for error conditions on the socket
if (polled[i].revents & (POLLERR | POLLHUP | POLLNVAL)) {
defer_close = true;
} else if (polled[i].revents & POLLOUT) {
int err = 0;
socklen_t len = sizeof(err);
if (getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, (void*) &err, &len) < 0 || err != 0)
defer_close = true;
else {
conn->connecting = false;
events[num_events++] = (Event) { EVENT_CONNECT, conn - tcp->conns, conn->tag };
}
}
} else {
if (polled[i].revents & POLLIN) {
byte_queue_write_setmincap(&conn->input, 1<<9);
ByteView buf = byte_queue_write_buf(&conn->input);
int num = recv(conn->fd, (char*) buf.ptr, buf.len, 0);
if (num == 0)
defer_close = true;
else if (num < 0) {
if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN) // TODO: does Windows return these error codes or not?
defer_close = true;
num = 0;
}
byte_queue_write_ack(&conn->input, num);
ByteView msg = byte_queue_read_buf(&conn->input);
int ret = message_peek(msg, NULL, NULL);
byte_queue_read_ack(&conn->input, 0);
if (ret < 0) {
// Invalid message
defer_close = true;
} else if (ret == 0) {
// Still buffering
if (byte_queue_full(&conn->input))
defer_close = true;
} else {
// Message received
assert(ret > 0);
defer_ready = true;
}
}
if (polled[i].revents & POLLOUT) {
ByteView buf = byte_queue_read_buf(&conn->output);
int num = send(conn->fd, (char*) buf.ptr, buf.len, 0);
if (num < 0) {
if (errno != EINTR && errno != EWOULDBLOCK && errno != EAGAIN)
defer_close = true;
num = 0;
}
byte_queue_read_ack(&conn->output, num);
if (conn->closing && byte_queue_empty(&conn->output))
defer_close = true;
}
}
// TODO: byte_queue_error here?
removed[i] = defer_close;
if (0) {}
else if (defer_close) events[num_events++] = (Event) { EVENT_DISCONNECT, conn - tcp->conns, conn->tag };
else if (defer_ready) events[num_events++] = (Event) { EVENT_MESSAGE, conn - tcp->conns, conn->tag };
}
}
for (int i = 1; i < num_polled; i++) {
if (removed[i]) {
Connection *conn = contexts[i];
assert(conn);
int removed_idx = conn - tcp->conns;
conn_free(conn);
int last_idx = --tcp->num_conns;
if (removed_idx != last_idx) {
*conn = tcp->conns[last_idx];
// Update event conn_idx values to reflect the swap
for (int j = 0; j < num_events; j++) {
if (events[j].conn_idx == last_idx)
events[j].conn_idx = removed_idx;
}
// Update contexts pointers for remaining iterations
for (int j = i + 1; j < num_polled; j++) {
if (contexts[j] == &tcp->conns[last_idx])
contexts[j] = conn;
}
}
}
}
return num_events;
}
ByteQueue *tcp_output_buffer(TCP *tcp, int conn_idx)
{
return &tcp->conns[conn_idx].output;
}
int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output)
{
if (tcp->num_conns == TCP_CONNECTION_LIMIT)
return -1;
int conn_idx = tcp->num_conns;
SOCKET fd = socket(AF_INET, SOCK_STREAM, 0);
if (fd == INVALID_SOCKET)
return -1;
if (set_socket_blocking(fd, false) < 0) {
CLOSE_SOCKET(fd);
return -1;
}
int ret;
if (addr.is_ipv4) {
struct sockaddr_in buf;
buf.sin_family = AF_INET;
buf.sin_port = htons(addr.port);
memcpy(&buf.sin_addr, &addr.ipv4, sizeof(IPv4));
ret = connect(fd, (struct sockaddr*) &buf, sizeof(buf));
} else {
struct sockaddr_in6 buf;
buf.sin6_family = AF_INET6;
buf.sin6_port = htons(addr.port);
memcpy(&buf.sin6_addr, &addr.ipv6, sizeof(IPv6));
ret = connect(fd, (struct sockaddr*) &buf, sizeof(buf));
}
bool connecting;
if (ret == 0) {
connecting = false;
} else {
if (errno != EINPROGRESS) {
CLOSE_SOCKET(fd);
return -1;
}
connecting = true;
}
// Check that this tag wasn't already used
for (int i = 0; i < tcp->num_conns; i++)
assert(tcp->conns[i].tag != tag);
conn_init(&tcp->conns[conn_idx], fd, connecting);
tcp->conns[conn_idx].tag = tag;
if (output)
*output = &tcp->conns[conn_idx].output;
tcp->num_conns++;
return 0;
}
void tcp_close(TCP *tcp, int conn_idx)
{
tcp->conns[conn_idx].closing = true;
tcp->conns[conn_idx].tag = -1; // Clear tag so new sends create a fresh connection
// TODO: if no event will be triggered, the connection will not be closed
// if the output buffer is empty, the connection should be closed here.
}
void tcp_set_tag(TCP *tcp, int conn_idx, int tag, bool unique)
{
assert(tag != -1);
if (unique) {
for (int i = 0; i < tcp->num_conns; i++)
assert(tcp->conns[i].tag != tag);
}
tcp->conns[conn_idx].tag = tag;
}
int tcp_get_tag(TCP *tcp, int conn_idx)
{
return tcp->conns[conn_idx].tag;
}
src/tcp.h
-87
View File
@@ -1,87 +0,0 @@
#ifndef TCP_INCLUDED
#define TCP_INCLUDED
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
# define QUAKEY_ENABLE_MOCKS
# include <quakey.h>
#else
# ifdef _WIN32
# include <winsock2.h>
# endif
#endif
#include "byte_queue.h"
#ifdef _WIN32
#define CLOSE_SOCKET closesocket
#else
#define SOCKET int
#define INVALID_SOCKET -1
#define CLOSE_SOCKET close
#endif
#ifndef TCP_CONNECTION_LIMIT
// Maximum number of connections that can be managed
// simultaneously.
#define TCP_CONNECTION_LIMIT 512
#endif
// This is the maximum number of descriptors that the
// TCP system will want to wait at any given time.
// One descriptor per connection plus a listener socket
// and a self-pipe handle for wakeup.
#define TCP_POLL_CAPACITY (TCP_CONNECTION_LIMIT+2)
// Number of TCP events that can be returned at a given
// time by "tcp_translate_events". There may be a single
// event per connection (MESSAGE, CONNECT, DISCONNECT)
// plus a general WAKEUP event.
#define TCP_EVENT_CAPACITY (TCP_CONNECTION_LIMIT+1)
typedef enum {
EVENT_WAKEUP,
EVENT_MESSAGE,
EVENT_CONNECT,
EVENT_DISCONNECT,
} EventType;
typedef struct {
EventType type;
int conn_idx;
int tag;
} Event;
typedef struct {
SOCKET fd;
int tag;
bool connecting;
bool closing;
uint32_t msglen;
ByteQueue input;
ByteQueue output;
} Connection;
typedef struct {
SOCKET listen_fd;
SOCKET wait_fd;
SOCKET signal_fd;
int num_conns;
Connection conns[TCP_CONNECTION_LIMIT];
} TCP;
int tcp_context_init(TCP *tcp);
void tcp_context_free(TCP *tcp);
int tcp_wakeup(TCP *tcp);
int tcp_index_from_tag(TCP *tcp, int tag);
int tcp_listen(TCP *tcp, Address addr);
int tcp_next_message(TCP *tcp, int conn_idx, ByteView *msg, uint16_t *type);
void tcp_consume_message(TCP *tcp, int conn_idx);
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);
int tcp_connect(TCP *tcp, Address addr, int tag, ByteQueue **output);
void tcp_close(TCP *tcp, int conn_idx);
void tcp_set_tag(TCP *tcp, int conn_idx, int tag, bool unique);
int tcp_get_tag(TCP *tcp, int conn_idx);
#endif // TCP_INCLUDED