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

Move files from lib/ into src/ and refactor the random client

Browse Source
This commit is contained in:
cozis
2026-02-21 12:21:32 +01:00
parent b64517c20e
commit 99b6c9dfcc
22 changed files with 846 additions and 1100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic.c
+232
View File
@@ -0,0 +1,232 @@
#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;
return (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
+61
View File
@@ -0,0 +1,61 @@
#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 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/blob_client.c
-675
View File
@@ -1,675 +0,0 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <quakey.h>
#include <stdint.h>
#include <assert.h>
#include "blob_client.h"
#include "server.h"
#define TIME_FMT "%7.3fs"
#define TIME_VAL(t) ((double)(t) / 1000000000.0)
static uint64_t next_blob_client_id = 100;
static uint64_t blob_random(void)
{
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
return quakey_random();
#else
return (uint64_t)rand();
#endif
}
static void blob_log_impl(BlobClientState *state, Time now, const char *event, const char *detail)
{
printf("[" TIME_FMT "] BLOB %lu | %-20s %s\n",
TIME_VAL(now),
state->client_id,
event,
detail ? detail : "");
}
#define blob_log(state, now, event, fmt, ...) do { \
char _detail[256]; \
snprintf(_detail, sizeof(_detail), fmt, ##__VA_ARGS__); \
blob_log_impl(state, now, event, _detail); \
} while (0)
#define blob_log_simple(state, now, event) \
blob_log_impl(state, now, event, NULL)
static int leader_idx(BlobClientState *state)
{
return state->view_number % state->num_servers;
}
///////////////////////////////////////////////////////////////
// Chunk ack counting
///////////////////////////////////////////////////////////////
static int count_acks(uint32_t mask)
{
int n = 0;
for (int i = 0; i < 32; i++)
if (mask & (1u << i)) n++;
return n;
}
static bool all_chunks_acked(BlobClientState *state)
{
for (int i = 0; i < state->num_chunks; i++) {
if (count_acks(state->chunks[i].ack_mask) < state->f_plus_one)
return false;
}
return true;
}
static bool all_chunks_fetched(BlobClientState *state)
{
for (int i = 0; i < state->num_chunks; i++) {
if (!state->chunks[i].fetched)
return false;
}
return true;
}
///////////////////////////////////////////////////////////////
// Generate a test blob
///////////////////////////////////////////////////////////////
static void generate_test_blob(BlobClientState *state)
{
// Generate random bucket/key
snprintf(state->bucket, META_BUCKET_MAX, "blob-b%d", (int)(blob_random() % 4));
snprintf(state->key, META_KEY_MAX, "blob-k%d", (int)(blob_random() % 64));
// 1-3 chunks per blob
state->num_chunks = 1 + blob_random() % 3;
state->blob_size = 0;
for (int i = 0; i < state->num_chunks; i++) {
// Generate random hash for each chunk
for (int j = 0; j < 32; j++)
state->chunks[i].hash.data[j] = blob_random() & 0xFF;
state->chunks[i].size = BLOB_TEST_CHUNK_SIZE;
state->chunks[i].ack_mask = 0;
state->chunks[i].fetched = false;
state->blob_size += BLOB_TEST_CHUNK_SIZE;
}
// Content hash (random for now)
for (int j = 0; j < 32; j++)
state->content_hash.data[j] = blob_random() & 0xFF;
}
///////////////////////////////////////////////////////////////
// Send operations
///////////////////////////////////////////////////////////////
// Send StoreChunk for a given chunk to a specific server.
// The chunk data is the hash bytes (32 bytes).
static void send_store_chunk(BlobClientState *state, int chunk_idx, int server_idx)
{
int conn_idx = tcp_index_from_tag(&state->tcp, server_idx);
if (conn_idx < 0) {
tcp_connect(&state->tcp, state->server_addrs[server_idx], server_idx, NULL);
return; // Will retry next tick
}
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
if (output == NULL) return;
StoreChunkMessage msg = {
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_STORE_CHUNK,
.length = sizeof(StoreChunkMessage) + state->chunks[chunk_idx].size,
},
.hash = state->chunks[chunk_idx].hash,
.size = state->chunks[chunk_idx].size,
};
byte_queue_write(output, &msg, sizeof(msg));
// Chunk data = hash bytes (BLOB_TEST_CHUNK_SIZE = 32 = sizeof(SHA256))
byte_queue_write(output, state->chunks[chunk_idx].hash.data, state->chunks[chunk_idx].size);
}
static void send_commit_put(BlobClientState *state)
{
int conn_idx = tcp_index_from_tag(&state->tcp, leader_idx(state));
if (conn_idx < 0) {
tcp_connect(&state->tcp, state->server_addrs[leader_idx(state)], leader_idx(state), NULL);
return;
}
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
if (output == NULL) return;
CommitPutMessage msg = {
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_COMMIT_PUT,
.length = sizeof(CommitPutMessage),
},
.oper = {
.type = META_OPER_PUT,
.size = state->blob_size,
.content_hash = state->content_hash,
.num_chunks = state->num_chunks,
},
.client_id = state->client_id,
.request_id = state->request_id,
};
memcpy(msg.oper.bucket, state->bucket, META_BUCKET_MAX);
memcpy(msg.oper.key, state->key, META_KEY_MAX);
for (int i = 0; i < state->num_chunks; i++) {
msg.oper.chunks[i].hash = state->chunks[i].hash;
msg.oper.chunks[i].size = state->chunks[i].size;
}
byte_queue_write(output, &msg, sizeof(msg));
}
static void send_get_blob(BlobClientState *state, int server_idx)
{
int conn_idx = tcp_index_from_tag(&state->tcp, server_idx);
if (conn_idx < 0) {
tcp_connect(&state->tcp, state->server_addrs[server_idx], server_idx, NULL);
return;
}
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
if (output == NULL) return;
GetBlobMessage msg = {
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_GET_BLOB,
.length = sizeof(GetBlobMessage),
},
};
memcpy(msg.bucket, state->bucket, META_BUCKET_MAX);
memcpy(msg.key, state->key, META_KEY_MAX);
byte_queue_write(output, &msg, sizeof(msg));
}
static void send_fetch_chunk(BlobClientState *state, int chunk_idx, int server_idx)
{
int conn_idx = tcp_index_from_tag(&state->tcp, server_idx);
if (conn_idx < 0) {
tcp_connect(&state->tcp, state->server_addrs[server_idx], server_idx, NULL);
return;
}
ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx);
if (output == NULL) return;
FetchChunkMessage msg = {
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_FETCH_CHUNK,
.length = sizeof(FetchChunkMessage),
},
.hash = state->chunks[chunk_idx].hash,
.sender_idx = -1, // Client (not a peer server)
};
byte_queue_write(output, &msg, sizeof(msg));
}
///////////////////////////////////////////////////////////////
// Message processing
///////////////////////////////////////////////////////////////
static int
process_message(BlobClientState *state,
int conn_idx, uint8_t type, ByteView msg)
{
(void) conn_idx;
Time now = get_current_time();
switch (type) {
case MESSAGE_TYPE_REDIRECT: {
RedirectMessage redirect;
if (msg.len != sizeof(RedirectMessage))
return -1;
memcpy(&redirect, msg.ptr, sizeof(redirect));
if (redirect.view_number > state->view_number) {
blob_log(state, now, "RECV REDIRECT", "view=%lu -> %lu",
(unsigned long)state->view_number,
(unsigned long)redirect.view_number);
state->view_number = redirect.view_number;
// Re-send CommitPut to new leader
if (state->phase == BLOB_COMMITTING) {
send_commit_put(state);
state->phase_time = now;
}
}
return 0;
}
case MESSAGE_TYPE_STORE_CHUNK_ACK: {
if (state->phase != BLOB_UPLOADING)
return 0;
StoreChunkAckMessage ack;
if (msg.len != sizeof(StoreChunkAckMessage))
return -1;
memcpy(&ack, msg.ptr, sizeof(ack));
// Find which chunk this ack is for
for (int i = 0; i < state->num_chunks; i++) {
if (memcmp(&state->chunks[i].hash, &ack.hash, sizeof(SHA256)) == 0) {
int tag = tcp_get_tag(&state->tcp, conn_idx);
if (tag >= 0 && tag < 32 && ack.success)
state->chunks[i].ack_mask |= (1u << tag);
blob_log(state, now, "RECV STORE_ACK", "chunk=%d server=%d ok=%d acks=%d/%d",
i, tag, ack.success,
count_acks(state->chunks[i].ack_mask), state->f_plus_one);
break;
}
}
// Check if all chunks have f+1 acks
if (all_chunks_acked(state)) {
blob_log(state, now, "UPLOAD DONE", "%s/%s chunks=%d",
state->bucket, state->key, state->num_chunks);
// Move to commit phase
state->phase = BLOB_COMMITTING;
state->phase_time = now;
state->request_id++;
send_commit_put(state);
blob_log(state, now, "SEND COMMIT_PUT", "%s/%s req=%lu",
state->bucket, state->key, state->request_id);
}
return 0;
}
case MESSAGE_TYPE_REPLY: {
if (state->phase != BLOB_COMMITTING)
return 0;
ReplyMessage reply;
if (msg.len != sizeof(ReplyMessage))
return -1;
memcpy(&reply, msg.ptr, sizeof(reply));
if (reply.request_id != state->request_id)
return 0;
if (reply.rejected) {
blob_log(state, now, "RECV REPLY", "REJECTED, retrying");
state->phase_time = now;
send_commit_put(state);
return 0;
}
state->puts_completed++;
blob_log(state, now, "PUT DONE", "%s/%s puts=%d",
state->bucket, state->key, state->puts_completed);
// Next: GET this blob back to verify
state->do_get_next = true;
state->phase = BLOB_IDLE;
return 0;
}
case MESSAGE_TYPE_GET_BLOB_RESPONSE: {
if (state->phase != BLOB_FETCHING_META)
return 0;
GetBlobResponseMessage resp;
if (msg.len != sizeof(GetBlobResponseMessage))
return -1;
memcpy(&resp, msg.ptr, sizeof(resp));
if (!resp.found) {
// Blob not yet committed on this server, retry another
blob_log(state, now, "RECV GET_BLOB", "NOT_FOUND, retrying");
state->fetch_server_idx = (state->fetch_server_idx + 1) % state->num_servers;
state->phase_time = now;
send_get_blob(state, state->fetch_server_idx);
return 0;
}
blob_log(state, now, "RECV GET_BLOB", "%s/%s chunks=%u",
state->bucket, state->key, resp.num_chunks);
// Verify metadata matches what we uploaded
if (resp.num_chunks != (uint32_t)state->num_chunks) {
fprintf(stderr, "BLOB CLIENT: metadata mismatch! expected %d chunks, got %u\n",
state->num_chunks, resp.num_chunks);
return -1;
}
for (int i = 0; i < state->num_chunks; i++) {
if (memcmp(&resp.chunks[i].hash, &state->chunks[i].hash, sizeof(SHA256)) != 0) {
fprintf(stderr, "BLOB CLIENT: chunk %d hash mismatch!\n", i);
return -1;
}
}
// Start fetching chunk data
state->phase = BLOB_FETCHING_DATA;
state->phase_time = now;
state->fetch_chunk_idx = 0;
state->fetch_server_idx = 0;
for (int i = 0; i < state->num_chunks; i++)
state->chunks[i].fetched = false;
// Send FetchChunk for the first chunk
send_fetch_chunk(state, 0, state->fetch_server_idx);
blob_log(state, now, "SEND FETCH_CHUNK", "chunk=0 server=%d", state->fetch_server_idx);
return 0;
}
case MESSAGE_TYPE_FETCH_CHUNK_RESPONSE: {
if (state->phase != BLOB_FETCHING_DATA)
return 0;
FetchChunkResponseMessage resp;
if (msg.len < sizeof(FetchChunkResponseMessage))
return -1;
memcpy(&resp, msg.ptr, sizeof(resp));
// Find which chunk this is for
int chunk_idx = -1;
for (int i = 0; i < state->num_chunks; i++) {
if (memcmp(&state->chunks[i].hash, &resp.hash, sizeof(SHA256)) == 0) {
chunk_idx = i;
break;
}
}
if (chunk_idx < 0)
return 0; // Unknown chunk, ignore
if (resp.size == 0) {
// Server doesn't have the chunk. Try next server.
blob_log(state, now, "RECV FETCH_RESP", "chunk=%d NOT_FOUND, trying next server", chunk_idx);
state->fetch_server_idx = (state->fetch_server_idx + 1) % state->num_servers;
send_fetch_chunk(state, chunk_idx, state->fetch_server_idx);
return 0;
}
// Verify data size
uint32_t data_size = msg.len - sizeof(FetchChunkResponseMessage);
if (data_size != resp.size || resp.size != state->chunks[chunk_idx].size) {
fprintf(stderr, "BLOB CLIENT: chunk %d size mismatch! expected %u, got %u\n",
chunk_idx, state->chunks[chunk_idx].size, resp.size);
return -1;
}
// Verify data content (data should equal hash bytes)
uint8_t *data = (uint8_t *)(msg.ptr + sizeof(FetchChunkResponseMessage));
if (memcmp(data, state->chunks[chunk_idx].hash.data, BLOB_TEST_CHUNK_SIZE) != 0) {
fprintf(stderr, "BLOB CLIENT: chunk %d data verification FAILED!\n", chunk_idx);
return -1;
}
state->chunks[chunk_idx].fetched = true;
blob_log(state, now, "RECV FETCH_RESP", "chunk=%d VERIFIED", chunk_idx);
// Check if all chunks fetched
if (all_chunks_fetched(state)) {
state->gets_completed++;
state->gets_verified++;
blob_log(state, now, "GET DONE", "%s/%s gets=%d verified=%d",
state->bucket, state->key, state->gets_completed, state->gets_verified);
state->do_get_next = false;
state->phase = BLOB_IDLE;
return 0;
}
// Fetch next unfetched chunk
for (int i = 0; i < state->num_chunks; i++) {
if (!state->chunks[i].fetched) {
state->fetch_chunk_idx = i;
state->fetch_server_idx = 0;
send_fetch_chunk(state, i, 0);
blob_log(state, now, "SEND FETCH_CHUNK", "chunk=%d server=0", i);
break;
}
}
return 0;
}
default:
break;
}
return 0;
}
///////////////////////////////////////////////////////////////
// Init / Tick / Free
///////////////////////////////////////////////////////////////
int blob_client_init(void *state_, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout)
{
BlobClientState *state = state_;
state->num_servers = 0;
for (int i = 1; i < argc; i++) {
if (!strcmp(argv[i], "--server")) {
i++;
if (i == argc) {
fprintf(stderr, "Option --server missing value\n");
return -1;
}
if (state->num_servers == NODE_LIMIT) {
fprintf(stderr, "Node limit reached\n");
return -1;
}
if (parse_addr_arg(argv[i], &state->server_addrs[state->num_servers++]) < 0) {
fprintf(stderr, "Malformed address\n");
return -1;
}
} else {
// Ignore unknown options
}
}
addr_sort(state->server_addrs, state->num_servers);
if (tcp_context_init(&state->tcp) < 0) {
fprintf(stderr, "Blob client :: Couldn't setup TCP context\n");
return -1;
}
// f = (num_servers - 1) / 2, so f+1 = (num_servers + 1) / 2
state->f_plus_one = (state->num_servers + 1) / 2;
state->view_number = 0;
state->request_id = 0;
state->client_id = next_blob_client_id++;
state->phase = BLOB_IDLE;
state->phase_time = 0;
state->do_get_next = false;
state->puts_completed = 0;
state->gets_completed = 0;
state->gets_verified = 0;
state->reconnect_time = 0;
state->upload_server_cursor = 0;
// Connect to all servers
for (int i = 0; i < state->num_servers; i++) {
if (tcp_connect(&state->tcp, state->server_addrs[i], i, NULL) < 0) {
fprintf(stderr, "Blob client :: Couldn't connect to server %d\n", i);
tcp_context_free(&state->tcp);
return -1;
}
}
{
Time now = get_current_time();
blob_log(state, now, "INIT", "servers=%d f+1=%d", state->num_servers, state->f_plus_one);
}
*timeout = 0;
if (pcap < TCP_POLL_CAPACITY) {
fprintf(stderr, "Blob client :: Not enough poll capacity\n");
return -1;
}
*pnum = tcp_register_events(&state->tcp, ctxs, pdata);
return 0;
}
int blob_client_tick(void *state_, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout)
{
BlobClientState *state = state_;
Event events[TCP_EVENT_CAPACITY];
int num_events = tcp_translate_events(&state->tcp, events, ctxs, pdata, *pnum);
for (int i = 0; i < num_events; i++) {
if (events[i].type == EVENT_DISCONNECT) {
int conn_idx = events[i].conn_idx;
tcp_close(&state->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(&state->tcp, conn_idx, &msg, &msg_type);
if (ret == 0)
break;
if (ret < 0) {
tcp_close(&state->tcp, conn_idx);
break;
}
ret = process_message(state, conn_idx, msg_type, msg);
if (ret < 0) {
tcp_close(&state->tcp, conn_idx);
break;
}
tcp_consume_message(&state->tcp, conn_idx);
}
}
Time now = get_current_time();
// Timeout handling: if we've been in any phase too long, retry
if (state->phase != BLOB_IDLE) {
Time phase_deadline = state->phase_time + PRIMARY_DEATH_TIMEOUT_SEC * 1000000000ULL;
if (phase_deadline <= now) {
blob_log(state, now, "TIMEOUT", "phase=%d, retrying", state->phase);
switch (state->phase) {
case BLOB_UPLOADING:
// Re-send StoreChunk for chunks that need more acks
for (int c = 0; c < state->num_chunks; c++) {
if (count_acks(state->chunks[c].ack_mask) < state->f_plus_one) {
for (int s = 0; s < state->f_plus_one; s++) {
if (!(state->chunks[c].ack_mask & (1u << s)))
send_store_chunk(state, c, s);
}
}
}
state->phase_time = now;
break;
case BLOB_COMMITTING:
state->view_number++;
send_commit_put(state);
state->phase_time = now;
break;
case BLOB_FETCHING_META:
state->fetch_server_idx = (state->fetch_server_idx + 1) % state->num_servers;
send_get_blob(state, state->fetch_server_idx);
state->phase_time = now;
break;
case BLOB_FETCHING_DATA:
state->fetch_server_idx = (state->fetch_server_idx + 1) % state->num_servers;
send_fetch_chunk(state, state->fetch_chunk_idx, state->fetch_server_idx);
state->phase_time = now;
break;
default:
break;
}
}
}
// Ensure connections to all servers (needed during all phases for
// retransmission after network partitions or server restarts)
for (int i = 0; i < state->num_servers; i++) {
int ci = tcp_index_from_tag(&state->tcp, i);
if (ci < 0)
tcp_connect(&state->tcp, state->server_addrs[i], i, NULL);
}
// Start new operation when idle
if (state->phase == BLOB_IDLE) {
if (state->do_get_next) {
// GET the blob we just PUT
state->phase = BLOB_FETCHING_META;
state->phase_time = now;
state->fetch_server_idx = 0;
send_get_blob(state, state->fetch_server_idx);
blob_log(state, now, "SEND GET_BLOB", "%s/%s server=%d",
state->bucket, state->key, state->fetch_server_idx);
} else {
// PUT a new blob
generate_test_blob(state);
state->phase = BLOB_UPLOADING;
state->phase_time = now;
blob_log(state, now, "START PUT", "%s/%s chunks=%d",
state->bucket, state->key, state->num_chunks);
// Send StoreChunk for each chunk to the first f+1 servers
for (int c = 0; c < state->num_chunks; c++) {
for (int s = 0; s < state->f_plus_one; s++) {
send_store_chunk(state, c, s);
}
}
}
}
// Set timeout
Time deadline = INVALID_TIME;
if (state->phase != BLOB_IDLE) {
nearest_deadline(&deadline, state->phase_time + PRIMARY_DEATH_TIMEOUT_SEC * 1000000000ULL);
}
*timeout = deadline_to_timeout(deadline, now);
if (pcap < TCP_POLL_CAPACITY)
return -1;
*pnum = tcp_register_events(&state->tcp, ctxs, pdata);
return 0;
}
int blob_client_free(void *state_)
{
BlobClientState *state = state_;
{
Time now = get_current_time();
blob_log(state, now, "SHUTDOWN", "puts=%d gets=%d verified=%d",
state->puts_completed, state->gets_completed, state->gets_verified);
}
tcp_context_free(&state->tcp);
return 0;
}
src/blob_client.h
-86
View File
@@ -1,86 +0,0 @@
#ifndef BLOB_CLIENT_INCLUDED
#define BLOB_CLIENT_INCLUDED
#include <lib/tcp.h>
#include <lib/basic.h>
#include "config.h"
#include "metadata.h"
// Maximum number of chunks per blob in the test client.
// Kept small for simulation; real blobs use META_CHUNKS_MAX.
#define BLOB_MAX_CHUNKS 8
// Chunk data size for the test client (32 bytes = SHA256 hash length).
// Each chunk's data is its hash bytes, making verification trivial.
#define BLOB_TEST_CHUNK_SIZE 32
typedef enum {
BLOB_IDLE, // Ready to start a new operation
BLOB_UPLOADING, // Sending StoreChunk, waiting for acks
BLOB_COMMITTING, // Sent CommitPut, waiting for REPLY
BLOB_FETCHING_META, // Sent GetBlob, waiting for response
BLOB_FETCHING_DATA, // Sending FetchChunk, waiting for responses
} BlobPhase;
typedef struct {
SHA256 hash;
uint32_t size;
uint32_t ack_mask; // Bitmask: which servers acked this chunk
bool fetched; // GET: whether chunk data was received and verified
} BlobChunkState;
typedef struct {
TCP tcp;
Address server_addrs[NODE_LIMIT];
int num_servers;
int f_plus_one; // Number of servers to upload each chunk to
uint64_t view_number;
uint64_t client_id;
uint64_t request_id;
BlobPhase phase;
Time phase_time; // When we entered this phase
// Current blob metadata
char bucket[META_BUCKET_MAX];
char key[META_KEY_MAX];
uint64_t blob_size;
SHA256 content_hash;
int num_chunks;
BlobChunkState chunks[BLOB_MAX_CHUNKS];
// Upload tracking
int upload_server_cursor; // Next server index to try for StoreChunk
// Download tracking
int fetch_chunk_idx; // Which chunk we're currently fetching
int fetch_server_idx; // Which server to try next
// Alternation between PUT and GET
bool do_get_next; // If true, next op is GET; else PUT
// Statistics
int puts_completed;
int gets_completed;
int gets_verified;
Time reconnect_time;
} BlobClientState;
struct pollfd;
int blob_client_init(void *state, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout);
int blob_client_tick(void *state, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout);
int blob_client_free(void *state);
#endif // BLOB_CLIENT_INCLUDED
src/byte_queue.c
+306
View File
@@ -0,0 +1,306 @@
#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
@@ -0,0 +1,53 @@
#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 <lib/file_system.h>
#include "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 <lib/basic.h>
#include "basic.h"
typedef struct {
char base_path[256];
src/client.c
+663 -280
View File
File diff suppressed because it is too large Load Diff
src/client.h
-49
View File
@@ -1,49 +0,0 @@
#ifndef CLIENT_INCLUDED
#define CLIENT_INCLUDED
#include <lib/tcp.h>
#include <lib/basic.h>
#include "config.h"
#include "metadata.h"
typedef struct {
TCP tcp;
// True if we are waiting for a response
bool pending;
Time request_time; // When the current request was sent
// The operation sent in the current pending request (for logging)
MetaOper last_oper;
// Checker support
MetaResult last_result;
bool last_was_timeout;
bool last_was_rejected;
Address server_addrs[NODE_LIMIT];
int num_servers;
uint64_t view_number;
uint64_t client_id;
uint64_t request_id;
Time reconnect_time; // Earliest time to retry connecting to leader
} ClientState;
struct pollfd;
int client_init(void *state, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout);
int client_tick(void *state, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout);
int client_free(void *state);
#endif // CLIENT_INCLUDED
src/file_system.c
+459
View File
@@ -0,0 +1,459 @@
#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
@@ -0,0 +1,58 @@
#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/message.c
+81
View File
@@ -0,0 +1,81 @@
#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
@@ -0,0 +1,46 @@
#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 <lib/basic.h>
#include "basic.h"
typedef struct {
SHA256 hash;
src/random_client.c
+102
View File
@@ -0,0 +1,102 @@
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
#include <quakey.h>
#include <stdint.h>
#include <assert.h>
#include "server.h"
#include "randm_client.h"
static uint64_t next_random_client_id = 100;
int random_client_init(void *state_, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout)
{
RandomClient *state = state_;
char *addrs[NODE_LIMIT];
int num_addrs = 0;
for (int i = 1; i < argc; i++) {
if (!strcmp(argv[i], "--server")) {
i++;
if (i == argc) {
fprintf(stderr, "Option --server missing value\n");
return -1;
}
if (num_addrs == NODE_LIMIT) {
fprintf(stderr, "Node limit reached\n");
return -1;
}
addrs[i] = argv[i];
num_addrs++;
} else {
// Ignore unknown options
}
}
ToastyFS *tfs = toastyfs_init(addrs, num_addrs);
if (tfs == NULL)
return -1;
*timeout = 0;
if (pcap < TCP_POLL_CAPACITY) {
fprintf(stderr, "Blob client :: Not enough poll capacity\n");
return -1;
}
*pnum = tcp_register_events(&state->tcp, ctxs, pdata);
return 0;
}
int random_client_tick(void *state_, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout)
{
RandomClient *state = state_;
toastyfs_process_events(state->tfs, ctxs, pdata, *pnum);
ToastyFS_Result result = toastyfs_get_result(state->tfs);
switch (result.type) {
case TOASTYFS_RESULT_VOID:
break;
case TOASTYFS_RESULT_PUT:
{
// TODO
}
break;
case TOASTYFS_RESULT_GET:
{
// TODO
}
break;
case TOASTYFS_RESULT_DELETE:
{
// TODO
}
break;
}
switch (choose_random_oper()) {
case OPER_GET:
toastyfs_async_get(state->tfs, xxx);
break;
case OPER_PUT:
toastyfs_async_put(state->tfs, xxx);
break;
case OPER_DELETE:
toastyfs_async_delete(state->tfs, xxx);
break;
}
*pnum = toastyfs_register_events(tfs, ctxs, pdata, pcap, timeout);
return 0;
}
int random_client_free(void *state_)
{
RandomClient *state = state_;
toastyfs_free(state->tfs);
return 0;
}
src/random_client.h
+25
View File
@@ -0,0 +1,25 @@
#ifndef BLOB_CLIENT_INCLUDED
#define BLOB_CLIENT_INCLUDED
#include "tcp.h"
#include "basic.h"
#include "config.h"
#include "metadata.h"
typedef struct {
ToastyFS *tfs;
} RandomClient;
struct pollfd;
int random_client_init(void *state, int argc, char **argv,
void **ctxs, struct pollfd *pdata, int pcap, int *pnum,
int *timeout);
int random_client_tick(void *state, void **ctxs,
struct pollfd *pdata, int pcap, int *pnum, int *timeout);
int random_client_free(void *state);
#endif // BLOB_CLIENT_INCLUDED
src/server.h
+3 -4
View File
@@ -1,10 +1,9 @@
#ifndef NODE_INCLUDED
#define NODE_INCLUDED
#include <lib/tcp.h>
#include <lib/basic.h>
#include <lib/message.h>
#include "tcp.h"
#include "basic.h"
#include "message.h"
#include "log.h"
#include "config.h"
#include "metadata.h"
src/tcp.c
+501
View File
@@ -0,0 +1,501 @@
#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;
}
// TODO: mark address as reusable in debug builds
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
@@ -0,0 +1,87 @@
#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