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

Progress

Browse Source
This commit is contained in:
cozis
2025-11-09 12:08:22 +01:00
parent 88735d1595
commit 739d6a7aee
10 changed files with 495 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic.c
+9
View File
@@ -94,3 +94,12 @@ int getargi(int argc, char **argv, char *name, int fallback)
}
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];
}
}
src/basic.h
+10
View File
@@ -8,6 +8,14 @@ typedef struct {
char data[32];
} SHA256;
typedef struct {
uint32_t data;
} IPv4;
typedef struct {
uint16_t data[8];
} IPv6;
typedef struct {
char *ptr;
int len;
@@ -29,4 +37,6 @@ 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);
#endif // BASIC_INCLUDED
src/chunk_server.c
+9 -13
View File
@@ -72,15 +72,6 @@ static void chunk_store_free(ChunkStore *store)
(void) store;
}
static 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];
}
}
static string hash2path(ChunkStore *store, SHA256 hash, char *out)
{
strcpy(out, store->path);
@@ -815,10 +806,10 @@ start_connecting_to_metadata_server(ChunkServer *state)
int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout)
{
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
int port = getargi(argc, argv, "--port", 8081);
string path = getargs(argc, argv, "--path", "chunk_server_data/");
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
int port = getargi(argc, argv, "--port", 8081);
string path = getargs(argc, argv, "--path", "chunk_server_data/");
bool trace = getargb(argc, argv, "--trace");
string remote_addr = getargs(argc, argv, "--remote-addr", "127.0.0.1");
int remote_port = getargi(argc, argv, "--remote-port", 8080);
@@ -828,6 +819,8 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts
if (remote_port <= 0 || remote_port >= 1<<16)
return -1;
state->trace = trace;
tcp_context_init(&state->tcp);
int ret = tcp_listen(&state->tcp, addr, port);
@@ -953,6 +946,9 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled
break;
}
if (state->trace)
message_dump(stdout, msg);
switch (tcp_get_tag(&state->tcp, conn_idx)) {
case TAG_METADATA_SERVER:
ret = process_metadata_server_message(state, conn_idx, msg_type, msg);
src/chunk_server.h
+1
View File
@@ -26,6 +26,7 @@ typedef struct {
} PendingDownloadList;
typedef struct {
bool trace;
Address local_addr;
Address remote_addr;
Time disconnect_time;
src/client.c
+2 -1
View File
@@ -344,7 +344,7 @@ static int send_download_chunk(TinyDFS *tdfs, int chunk_server_idx,
ByteQueue *output = tcp_output_buffer(&tdfs->tcp, conn_idx);
message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK);
message_write(&writer, &hash, sizeof(hash));
message_write(&writer, &hash, sizeof(hash));
message_write(&writer, &offset, sizeof(offset));
message_write(&writer, &length, sizeof(length));
@@ -465,6 +465,7 @@ int tinydfs_submit_list(TinyDFS *tdfs, char *path, int path_len)
MessageWriter writer;
metadata_server_request_start(tdfs, &writer, MESSAGE_TYPE_LIST);
message_write(&writer, &tmp, sizeof(tmp));
message_write(&writer, path, path_len);
src/message.c
+454
View File
@@ -1,4 +1,5 @@
#include <string.h>
#include <arpa/inet.h> // inet_ntop
#include "message.h"
@@ -59,3 +60,456 @@ int message_peek(ByteView msg, uint16_t *type, uint32_t *len)
return 1;
}
static char *message_type_to_str(uint16_t type)
{
switch (type) {
// Client -> Metadata server
case MESSAGE_TYPE_CREATE: return "CREATE";
case MESSAGE_TYPE_DELETE: return "DELETE";
case MESSAGE_TYPE_LIST: return "LIST";
case MESSAGE_TYPE_READ: return "READ";
case MESSAGE_TYPE_WRITE: return "WRITE";
// Client -> Chunk server
case MESSAGE_TYPE_CREATE_CHUNK: return "CREATE_CHUNK";
case MESSAGE_TYPE_UPLOAD_CHUNK: return "UPLOAD_CHUNK";
case MESSAGE_TYPE_DOWNLOAD_CHUNK: return "DOWNLOAD_CHUNK";
// Metadata server -> Client
case MESSAGE_TYPE_CREATE_ERROR: return "CREATE_ERROR";
case MESSAGE_TYPE_CREATE_SUCCESS: return "CREATE_SUCCESS";
case MESSAGE_TYPE_DELETE_ERROR: return "DELETE_ERROR";
case MESSAGE_TYPE_DELETE_SUCCESS: return "DELETE_SUCCESS";
case MESSAGE_TYPE_LIST_ERROR: return "LIST_ERROR";
case MESSAGE_TYPE_LIST_SUCCESS: return "LIST_SUCCESS";
case MESSAGE_TYPE_READ_ERROR: return "READ_ERROR";
case MESSAGE_TYPE_READ_SUCCESS: return "READ_SUCCESS";
case MESSAGE_TYPE_WRITE_ERROR: return "WRITE_ERROR";
case MESSAGE_TYPE_WRITE_SUCCESS: return "WRITE_SUCCESS";
// Metadata server -> Chunk server
case MESSAGE_TYPE_STATE_UPDATE: return "STATE_UPDATE";
case MESSAGE_TYPE_DOWNLOAD_LOCATIONS: return "DOWNLOAD_LOCATIONS";
// Chunk server -> Metadata server
case MESSAGE_TYPE_AUTH: return "AUTH";
case MESSAGE_TYPE_STATE_UPDATE_ERROR: return "UPDATE_ERROR";
case MESSAGE_TYPE_STATE_UPDATE_SUCCESS: return "UPDATE_SUCCESS";
// Chunk server -> Client
case MESSAGE_TYPE_CREATE_CHUNK_ERROR: return "CREATE_CHUNK_ERROR";
case MESSAGE_TYPE_CREATE_CHUNK_SUCCESS: return "CREATE_CHUNK_SUCCESS";
case MESSAGE_TYPE_UPLOAD_CHUNK_ERROR: return "UPLOAD_CHUNK_ERROR";
case MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS: return "UPLOAD_CHUNK_SUCCESS";
case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR: return "DOWNLOAD_CHUNK_ERROR";
case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS: return "DOWNLOAD_CHUNK_SUCCESS";
}
return "???";
}
void message_dump(FILE *stream, ByteView msg)
{
BinaryReader reader = { msg.ptr, msg.len, 0 };
fprintf(stream, "message:\n");
fprintf(stream, " header:\n");
uint16_t version;
if (!binary_read(&reader, &version, sizeof(version))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " version: %d\n", version);
uint16_t type;
if (!binary_read(&reader, &type, sizeof(type))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " type: %s\n", message_type_to_str(type));
uint32_t length;
if (!binary_read(&reader, &length, sizeof(length))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " length: %d\n", length);
fprintf(stream, " body:\n");
switch (type) {
// Client -> Metadata server
case MESSAGE_TYPE_CREATE:
{
uint16_t path_len;
if (!binary_read(&reader, &path_len, sizeof(path_len))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path_len: %d\n", path_len);
char *path = reader.src + reader.cur;
if (!binary_read(&reader, NULL, path_len)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path: %.*s\n", (int) path_len, path);
uint8_t is_dir;
if (!binary_read(&reader, &is_dir, sizeof(is_dir))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " is_dir: %s\n", is_dir ? "true" : "false");
if (!is_dir) {
uint32_t chunk_size;
if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " chunk_size: %d\n", chunk_size);
}
}
break;
case MESSAGE_TYPE_DELETE:
{
uint16_t path_len;
if (!binary_read(&reader, &path_len, sizeof(path_len))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path_len: %d\n", path_len);
char *path = reader.src + reader.cur;
if (!binary_read(&reader, NULL, path_len)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path: %.*s\n", (int) path_len, path);
}
break;
case MESSAGE_TYPE_LIST:
{
uint16_t path_len;
if (!binary_read(&reader, &path_len, sizeof(path_len))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path_len: %d\n", path_len);
char *path = reader.src + reader.cur;
if (!binary_read(&reader, NULL, path_len)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path: %.*s\n", (int) path_len, path);
}
break;
case MESSAGE_TYPE_READ:
{
uint16_t path_len;
if (!binary_read(&reader, &path_len, sizeof(path_len))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path_len: %d\n", path_len);
char *path = reader.src + reader.cur;
if (!binary_read(&reader, NULL, path_len)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path: %.*s\n", (int) path_len, path);
uint32_t offset;
if (!binary_read(&reader, &offset, sizeof(offset))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " offset: %d\n", offset);
uint32_t length;
if (!binary_read(&reader, &length, sizeof(length))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " length: %d\n", length);
}
break;
case MESSAGE_TYPE_WRITE:
{
uint16_t path_len;
if (!binary_read(&reader, &path_len, sizeof(path_len))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path_len: %d\n", path_len);
char *path = reader.src + reader.cur;
if (!binary_read(&reader, NULL, path_len)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " path: %.*s\n", (int) path_len, path);
uint32_t offset;
if (!binary_read(&reader, &offset, sizeof(offset))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " offset: %d\n", offset);
uint32_t length;
if (!binary_read(&reader, &length, sizeof(length))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " length: %d\n", length);
uint32_t num_chunks;
if (!binary_read(&reader, &num_chunks, sizeof(num_chunks))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " num_chunks: %d\n", num_chunks);
for (uint32_t i = 0; i < num_chunks; i++) {
char hash_str[64];
SHA256 old_hash;
if (!binary_read(&reader, &old_hash, sizeof(old_hash))) {
fprintf(stream, " (incomplete)\n");
return;
}
append_hex_as_str(hash_str, old_hash);
fprintf(stream, " old_hash: %.64s\n", hash_str); // TODO: stringify
SHA256 new_hash;
if (!binary_read(&reader, &new_hash, sizeof(new_hash))) {
fprintf(stream, " (incomplete)\n");
return;
}
append_hex_as_str(hash_str, new_hash);
fprintf(stream, " new_hash: %.64s\n", hash_str); // TODO: stringify
uint32_t num_locations;
if (!binary_read(&reader, &num_locations, sizeof(num_locations))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " num_locations: %d\n", num_locations);
for (uint32_t j = 0; j < num_locations; j++) {
uint8_t is_ipv4;
if (!binary_read(&reader, &is_ipv4, sizeof(is_ipv4))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " is_ipv4: %s (%d)\n", is_ipv4 ? "true" : "false", is_ipv4);
if (is_ipv4) {
IPv4 ipv4;
if (!binary_read(&reader, &ipv4, sizeof(ipv4))) {
fprintf(stream, " (incomplete)\n");
return;
}
char ip_str[INET_ADDRSTRLEN];
inet_ntop(AF_INET, &ipv4, ip_str, sizeof(ip_str));
fprintf(stream, " ipv4: %s\n", ip_str);
} else {
IPv6 ipv6;
if (!binary_read(&reader, &ipv6, sizeof(ipv6))) {
fprintf(stream, " (incomplete)\n");
return;
}
char ip_str[INET_ADDRSTRLEN];
inet_ntop(AF_INET6, &ipv6, ip_str, sizeof(ip_str));
fprintf(stream, " ipv6: %s\n", ip_str);
}
uint16_t port;
if (!binary_read(&reader, &port, sizeof(port))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " port: %d\n", port);
}
}
}
break;
// Client -> Chunk server
case MESSAGE_TYPE_CREATE_CHUNK:
{
uint32_t chunk_size;
if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " chunk_size: %d\n", chunk_size);
uint32_t offset;
if (!binary_read(&reader, &offset, sizeof(offset))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " offset: %d\n", offset);
uint32_t length;
if (!binary_read(&reader, &length, sizeof(length))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " length: %d\n", length);
if (!binary_read(&reader, NULL, length)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " data: (...)\n");
}
break;
case MESSAGE_TYPE_UPLOAD_CHUNK:
{
SHA256 hash;
if (!binary_read(&reader, &hash, sizeof(hash))) {
fprintf(stream, " (incomplete)\n");
return;
}
char hash_str[64];
append_hex_as_str(hash_str, hash);
fprintf(stream, " hash: %.64s\n", hash_str);
uint32_t offset;
if (!binary_read(&reader, &offset, sizeof(offset))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " offset: %d\n", offset);
uint32_t length;
if (!binary_read(&reader, &length, sizeof(length))) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " length: %d\n", length);
if (!binary_read(&reader, NULL, length)) {
fprintf(stream, " (incomplete)\n");
return;
}
fprintf(stream, " data: (...)\n");
}
break;
case MESSAGE_TYPE_DOWNLOAD_CHUNK:
printf(" (TODO)\n");
break;
// Metadata server -> Client
case MESSAGE_TYPE_CREATE_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_CREATE_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_DELETE_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_DELETE_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_LIST_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_LIST_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_READ_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_READ_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_WRITE_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_WRITE_SUCCESS:
printf(" (TODO)\n");
break;
// Metadata server -> Chunk server
case MESSAGE_TYPE_STATE_UPDATE:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_DOWNLOAD_LOCATIONS:
printf(" (TODO)\n");
break;
// Chunk server -> Metadata server
case MESSAGE_TYPE_AUTH:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_STATE_UPDATE_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_STATE_UPDATE_SUCCESS:
printf(" (TODO)\n");
break;
// Chunk server -> Client
case MESSAGE_TYPE_CREATE_CHUNK_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_CREATE_CHUNK_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_UPLOAD_CHUNK_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_UPLOAD_CHUNK_SUCCESS:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR:
printf(" (TODO)\n");
break;
case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS:
printf(" (TODO)\n");
break;
}
if (binary_read(&reader, NULL, 1))
fprintf(stream, " (unexpected bytes)\n");
}
src/message.h
+2
View File
@@ -1,6 +1,7 @@
#ifndef MESSAGE_INCLUDED
#define MESSAGE_INCLUDED
#include <stdio.h>
#include <stdbool.h>
#include "byte_queue.h"
@@ -76,5 +77,6 @@ bool message_writer_free(MessageWriter *writer);
void message_write(MessageWriter *writer, void *mem, int len);
int message_peek(ByteView msg, uint16_t *type, uint32_t *len);
void message_dump(FILE *stream, ByteView msg);
#endif // MESSAGE_INCLUDED
src/metadata_server.c
+7 -2
View File
@@ -818,12 +818,14 @@ static bool is_chunk_server_message_type(uint16_t type)
int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout)
{
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
int port = getargi(argc, argv, "--port", 8080);
string addr = getargs(argc, argv, "--addr", "127.0.0.1");
int port = getargi(argc, argv, "--port", 8080);
bool trace = getargb(argc, argv, "--trace");
if (port <= 0 || port >= 1<<16)
return -1;
state->trace = trace;
state->replication_factor = 3;
if (state->replication_factor > MAX_CHUNK_SERVERS)
return -1;
@@ -898,6 +900,9 @@ int metadata_server_step(MetadataServer *state, void **contexts, struct pollfd *
break;
}
if (state->trace)
message_dump(stdout, msg);
if (tcp_get_tag(&state->tcp, conn_idx) == CONNECTION_TAG_UNKNOWN) {
if (is_chunk_server_message_type(msg_type)) {
int chunk_server_idx = state->num_chunk_servers++;
src/metadata_server.h
+1
View File
@@ -41,6 +41,7 @@ typedef struct {
FileTree file_tree;
bool trace;
int replication_factor;
int num_chunk_servers;
src/tcp.h
-8
View File
@@ -25,14 +25,6 @@ typedef struct {
int conn_idx;
} Event;
typedef struct {
uint32_t data;
} IPv4;
typedef struct {
uint16_t data[8];
} IPv6;
typedef struct {
union {
IPv4 ipv4;