diff --git a/Makefile b/Makefile index 04a484d..e841e72 100644 --- a/Makefile +++ b/Makefile @@ -17,7 +17,7 @@ OFILES = $(CFILES:.c=.o) .PHONY: all clean coverage coverage-report coverage-html -all: toastyfs$(EXT) toastyfs_random_test$(EXT) example_client$(EXT) libtoastyfs.a +all: toastyfs$(EXT) toastyfs_random_test$(EXT) example_async_api$(EXT) example_blocking_api$(EXT) libtoastyfs.a coverage: toastyfs_random_test_coverage$(EXT) @@ -36,8 +36,11 @@ toastyfs_random_test$(EXT): $(CFILES) $(HFILES) toastyfs_random_test_coverage$(EXT): $(CFILES) $(HFILES) gcc -o $@ $(CFILES) $(COVERAGE_CFLAGS) $(LFLAGS) $(COVERAGE_LFLAGS) -Iinc -DBUILD_TEST -example_client$(EXT): libtoastyfs.a - gcc -o $@ examples/main.c $(CFLAGS) -ltoastyfs $(LFLAGS) -Iinc -L. +example_async_api$(EXT): libtoastyfs.a examples/async_api.c + gcc -o $@ examples/async_api.c $(CFLAGS) -ltoastyfs $(LFLAGS) -Iinc -L. + +example_blocking_api$(EXT): libtoastyfs.a examples/blocking_api.c + gcc -o $@ examples/blocking_api.c $(CFLAGS) -ltoastyfs $(LFLAGS) -Iinc -L. %.o: %.c $(HFILES) gcc -c -o $@ $< $(CFLAGS) -Iinc diff --git a/README.md b/README.md index 8c175c2..8cc878c 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # ToastyFS ToastyFS is a distributed file system for personal use. It's designed to be pragmatic, understandable, and robust. +𝅘𝅥𝅮 𝅘𝅥𝅮 Now let's get toasty 𝅘𝅥𝅮 𝅘𝅥𝅮 + ## Testing ToastyFS is tested by running an in-memory simulation of a cluster with many clients running hundreds of random operations in parallel. The test is run for long periods of times under valgrind or compiled with sanitizers. diff --git a/examples/async_api.c b/examples/async_api.c index c037ccc..3735a2d 100644 --- a/examples/async_api.c +++ b/examples/async_api.c @@ -1,12 +1,12 @@ -#include -#include +#include +#include int main(void) { ToastyString remote_addr = TOASTY_STR("127.0.0.1"); uint16_t remote_port = 8080; - Toasty *toasty = toasty_connect(remote_addr, remote_port); + ToastyFS *toasty = toasty_connect(remote_addr, remote_port); if (toasty == NULL) { printf("Couldn't connect to metadata server"); return -1; @@ -15,9 +15,6 @@ int main(void) ToastyString path_1 = TOASTY_STR("/first_file"); ToastyString path_2 = TOASTY_STR("/second_file"); - char msg_1[] = "This is file 1"; - char msg_2[] = "This is file 2"; - // Begin creation operation. This does not block. ToastyHandle create_handle_1 = toasty_begin_create_file(toasty, path_1, 1024); if (create_handle_1 == TOASTY_INVALID) { @@ -58,6 +55,6 @@ int main(void) printf("All files were created!\n"); - toastyfs_disconnect(tfs); + toasty_disconnect(toasty); return 0; } diff --git a/examples/blocking_api.c b/examples/blocking_api.c index 44668e5..75bba12 100644 --- a/examples/blocking_api.c +++ b/examples/blocking_api.c @@ -1,12 +1,12 @@ -#include -#include +#include +#include int main(void) { ToastyString remote_addr = TOASTY_STR("127.0.0.1"); uint16_t remote_port = 8080; - Toasty *toasty = toasty_connect(remote_addr, remote_port); + ToastyFS *toasty = toasty_connect(remote_addr, remote_port); if (toasty == NULL) { printf("Couldn't connect to metadata server"); return -1; @@ -29,6 +29,6 @@ int main(void) return -1; } - toastyfs_disconnect(tfs); + toasty_disconnect(toasty); return 0; } diff --git a/inc/ToastyFS.h b/inc/ToastyFS.h index 4c1dca2..6dd6ec7 100644 --- a/inc/ToastyFS.h +++ b/inc/ToastyFS.h @@ -155,6 +155,11 @@ ToastyHandle toasty_begin_write(ToastyFS *toasty, ToastyString path, // Note that if a result is returned, handles to that // operation are invalidated. // The "result" must be freed using "toasty_free_result". +// +// Note: +// If you call this function in a loop, the state of the +// operation will not progress. You can't wait for completion +// by calling this in a loop. You need to use "toasty_wait_result" int toasty_get_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *result); diff --git a/src/basic.h b/src/basic.h index f17b7c7..d095f12 100644 --- a/src/basic.h +++ b/src/basic.h @@ -16,6 +16,15 @@ 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; diff --git a/src/chunk_server.c b/src/chunk_server.c index 2aff06d..aaa1eb5 100644 --- a/src/chunk_server.c +++ b/src/chunk_server.c @@ -3,16 +3,123 @@ #include #include -#include "tcp.h" -#include "basic.h" -#include "byte_queue.h" #include "config.h" -#include "hash_set.h" +#include "metadata_server.h" #include "sha256.h" #include "message.h" #include "file_system.h" +#include "tcp.h" #include "chunk_server.h" +static string hash2path(ChunkServer *state, SHA256 hash, char *out) +{ + strcpy(out, state->path); + strcat(out, "/"); + + size_t tmp = strlen(out); + + append_hex_as_str(out + tmp, hash); + + out[tmp + 64] = '\0'; + + return (string) { out, strlen(out) }; +} + +// TODO: return an error when bitrot +static int load_chunk(ChunkServer *state, SHA256 hash, string *data) +{ + char buf[PATH_MAX]; + string path = hash2path(state, hash, buf); + + int ret = file_read_all(path, data); + if (ret < 0) + return -1; + + SHA256 fresh_hash; + sha256(data->ptr, data->len, (uint8_t*) &fresh_hash.data); + + if (memcmp(&hash, &fresh_hash, sizeof(SHA256))) { + free(data->ptr); + return -1; + } + + return 0; +} + +static int store_chunk(ChunkServer *state, string data, SHA256 *hash) +{ + sha256(data.ptr, data.len, (uint8_t*) hash->data); + + char buf[PATH_MAX]; + string path = hash2path(state, *hash, buf); + + // Note that this write is not atomic. If we crash + // while writing, we'll get an inconsistent file. + // This is okay as long as we check that the hash + // is correct while reading back the data. + Handle fd; + if (file_open(path, &fd) < 0) // TODO: open in overwrite mode + return -1; + int copied = 0; + while (copied < data.len) { + int ret = file_write(fd, + data.ptr + copied, + data.len - copied); + if (ret < 0) { + file_close(fd); + return -1; + } + copied += ret; + } + file_close(fd); + return 0; +} + +static bool chunk_exists(ChunkServer *state, SHA256 hash) +{ + char buf[PATH_MAX]; + string path = hash2path(state, hash, buf); + + // Try to open the file to check if it exists + // TODO: this isn't right. There should be something like file_exists + Handle fd; + if (file_open(path, &fd) == 0) { + file_close(fd); + return true; + } + return false; +} + +static int patch_chunk(ChunkServer *state, SHA256 target_chunk, + uint64_t patch_off, string patch, SHA256 *new_hash) +{ + string data; + int ret = load_chunk(state, target_chunk, &data); + if (ret < 0) + return -1; + + if (patch_off > SIZE_MAX - patch.len) { + sys_free(data.ptr); + return -1; + } + + if (patch_off + (size_t) patch.len > (size_t) data.len) { + sys_free(data.ptr); + return -1; + } + + memcpy(data.ptr + patch_off, patch.ptr, patch.len); + + ret = store_chunk(state, data, new_hash); + if (ret < 0) { + sys_free(data.ptr); + return -1; + } + + sys_free(data.ptr); + return 0; +} + static void download_targets_init(DownloadTargets *targets) { targets->items = NULL; @@ -97,151 +204,6 @@ static bool download_targets_pop(DownloadTargets *targets, return true; } -static int chunk_store_init(ChunkStore *store, string path) -{ - if (create_dir(path) && errno != EEXIST) - return -1; - - if (get_full_path(path, store->path) < 0) - return -1; - - return 0; -} - -static void chunk_store_free(ChunkStore *store) -{ - (void) store; -} - -static string hash2path(ChunkStore *store, SHA256 hash, char *out) -{ - strcpy(out, store->path); - strcat(out, "/"); - - size_t tmp = strlen(out); - - append_hex_as_str(out + tmp, hash); - - out[tmp + 64] = '\0'; - - return (string) { out, strlen(out) }; -} - -// TODO: return an error when bitrot -static int load_chunk(ChunkStore *store, SHA256 hash, string *data) -{ - char buf[PATH_MAX]; - string path = hash2path(store, hash, buf); - - int ret = file_read_all(path, data); - if (ret < 0) - return -1; - - SHA256 fresh_hash; - sha256(data->ptr, data->len, (uint8_t*) &fresh_hash.data); - - if (memcmp(&hash, &fresh_hash, sizeof(SHA256))) { - free(data->ptr); - return -1; - } - - return 0; -} - -static int store_chunk(ChunkStore *store, string data, SHA256 *hash) -{ - sha256(data.ptr, data.len, (uint8_t*) hash->data); - char buf[PATH_MAX]; - string path = hash2path(store, *hash, buf); - - // Note that this write is not atomic. If we crash - // while writing, we'll get an inconsistent file. - // This is okay as long as we check that the hash - // is correct while reading back the data. - Handle fd; - if (file_open(path, &fd) < 0) // TODO: open in overwrite mode - return -1; - int copied = 0; - while (copied < data.len) { - int ret = file_write(fd, - data.ptr + copied, - data.len - copied); - if (ret < 0) { - file_close(fd); - return -1; - } - copied += ret; - } - file_close(fd); - return 0; -} - -static int chunk_store_get(ChunkStore *store, SHA256 hash, string *data) -{ - return load_chunk(store, hash, data); -} - -static int chunk_store_add(ChunkStore *store, string data) -{ - SHA256 dummy; - return store_chunk(store, data, &dummy); -} - -static int chunk_store_remove(ChunkStore *store, SHA256 hash) -{ - char buf[PATH_MAX]; - string path = hash2path(store, hash, buf); - - if (remove_file_or_dir(path) < 0) - return -1; - return 0; -} - -static bool chunk_store_exists(ChunkStore *store, SHA256 hash) -{ - char buf[PATH_MAX]; - string path = hash2path(store, hash, buf); - - // Try to open the file to check if it exists - // TODO: this isn't right. There should be something like file_exists - Handle fd; - if (file_open(path, &fd) == 0) { - file_close(fd); - return true; - } - return false; -} - -static int chunk_store_patch(ChunkStore *store, SHA256 target_chunk, - uint64_t patch_off, string patch, SHA256 *new_hash) -{ - string data; - int ret = load_chunk(store, target_chunk, &data); - if (ret < 0) - return -1; - - if (patch_off > SIZE_MAX - patch.len) { - sys_free(data.ptr); - return -1; - } - - if (patch_off + (size_t) patch.len > (size_t) data.len) { - sys_free(data.ptr); - return -1; - } - - memcpy(data.ptr + patch_off, patch.ptr, patch.len); - - ret = store_chunk(store, data, new_hash); - if (ret < 0) { - sys_free(data.ptr); - return -1; - } - - sys_free(data.ptr); - return 0; -} - static int send_error(TCP *tcp, int conn_idx, bool close, uint16_t type, string msg) { @@ -275,15 +237,8 @@ static void start_download(ChunkServer *state) MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK); - - message_write(&writer, &target.hash, sizeof(target.hash)); - - uint32_t offset = 0; - message_write(&writer, &offset, sizeof(offset)); - - uint32_t length = 64 * 1024 * 1024; // TODO: there should be a special value for this - message_write(&writer, &length, sizeof(length)); - + message_write_hash(&writer, target.hash); + message_write_u8(&writer, 1); if (!message_writer_free(&writer)) return; @@ -319,7 +274,7 @@ static int process_metadata_server_sync_2(ChunkServer *state, // Elements in ms_add_list that are not held by the // chunk server are added to a temporary list tmp_list - if (!chunk_store_exists(&state->store, hash)) { + if (!chunk_exists(state, hash)) { if (hash_set_insert(&tmp_list, hash) < 0) { hash_set_free(&tmp_list); return -1; @@ -375,17 +330,13 @@ static int process_metadata_server_sync_2(ChunkServer *state, hash_set_free(&tmp_list); return -1; } - uint32_t count = tmp_list.count + state->cs_lst_list.count; - message_write(&writer, &count, sizeof(count)); + message_write_u32(&writer, tmp_list.count + state->cs_lst_list.count); - for (int i = 0; i < tmp_list.count; i++) { - SHA256 hash = tmp_list.items[i]; - message_write(&writer, &hash, sizeof(hash)); - } + for (int i = 0; i < tmp_list.count; i++) + message_write_hash(&writer, tmp_list.items[i]); for (int i = 0; i < state->cs_lst_list.count; i++) { - SHA256 hash = state->cs_lst_list.items[i]; - message_write(&writer, &hash, sizeof(hash)); + message_write_hash(&writer, state->cs_lst_list.items[i]); } hash_set_free(&tmp_list); @@ -404,29 +355,9 @@ process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) // The metadata server wants us to download chunks from other chunk servers BinaryReader reader = { msg.ptr, msg.len, 0 }; - - // Read header if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - // The message layout is this: - // - // struct ServerAddresses { - // uint32_t num_ipv4; - // uint32_t num_ipv6; - // IPv4Pair ipv4[num_ipv4]; - // IPv6Pair ipv6[num_ipv6]; - // } - // - // struct Message { - // uint32_t num_missing; - // struct { - // uint32_t num_holders; - // ServerAddresses holders[num_holders]; - // SHA256 hash; - // } entries[num_missing]; - // } - uint32_t num_missing; if (!binary_read(&reader, &num_missing, sizeof(num_missing))) return -1; @@ -438,45 +369,15 @@ process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) return -1; // Temporary storage for all addresses from all holders - IPv4 ipv4[256]; - IPv6 ipv6[256]; - uint16_t ipv4_port[256]; - uint16_t ipv6_port[256]; - uint32_t total_ipv4 = 0; - uint32_t total_ipv6 = 0; + Address addrs[REPLICATION_FACTOR]; + int num_addrs = 0; // Read addresses from each holder for (uint32_t j = 0; j < num_holders; j++) { - - uint32_t num_ipv4; - if (!binary_read(&reader, &num_ipv4, sizeof(num_ipv4))) + int num = binary_read_addr_list(&reader, addrs + num_addrs, REPLICATION_FACTOR - num_addrs); + if (num < 0) return -1; - - uint32_t num_ipv6; - if (!binary_read(&reader, &num_ipv6, sizeof(num_ipv6))) - return -1; - - // Read IPv4 addresses - for (uint32_t k = 0; k < num_ipv4; k++) { - if (total_ipv4 >= 256) - return -1; - if (!binary_read(&reader, &ipv4[total_ipv4], sizeof(ipv4[0]))) - return -1; - if (!binary_read(&reader, &ipv4_port[total_ipv4], sizeof(ipv4_port[0]))) - return -1; - total_ipv4++; - } - - // Read IPv6 addresses - for (uint32_t k = 0; k < num_ipv6; k++) { - if (total_ipv6 >= 256) - return -1; - if (!binary_read(&reader, &ipv6[total_ipv6], sizeof(ipv6[0]))) - return -1; - if (!binary_read(&reader, &ipv6_port[total_ipv6], sizeof(ipv6_port[0]))) - return -1; - total_ipv6++; - } + num_addrs += num; } // Read the hash @@ -484,20 +385,8 @@ process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - // Add to pending download list - for (uint32_t k = 0; k < total_ipv4; k++) - download_targets_push( - &state->download_targets, - (Address) { .is_ipv4=true, .ipv4=ipv4[k], .port=ipv4_port[k] }, - hash - ); - - for (uint32_t k = 0; k < total_ipv6; k++) - download_targets_push( - &state->download_targets, - (Address) { .is_ipv4=false, .ipv6=ipv6[k], .port=ipv6_port[k] }, - hash - ); + for (int j = 0; j < num_addrs; j++) + download_targets_push(&state->download_targets, addrs[i], hash); } if (binary_read(&reader, NULL, 1)) @@ -512,6 +401,8 @@ process_metadata_server_sync_4(ChunkServer *state, int conn_idx, ByteView msg) static int process_metadata_server_auth_response(ChunkServer *state, int conn_idx, ByteView msg) { + (void) conn_idx; + BinaryReader reader = { msg.ptr, msg.len, 0 }; if (!binary_read(&reader, NULL, sizeof(MessageHeader))) @@ -528,7 +419,7 @@ process_metadata_server_auth_response(ChunkServer *state, int conn_idx, ByteView if (avoid_full_scan) return 0; - string path = { state->store.path, strlen(state->store.path) }; + string path = { state->path, strlen(state->path) }; DirectoryScanner scanner; if (directory_scanner_init(&scanner, path) < 0) { @@ -585,17 +476,6 @@ process_metadata_server_auth_response(ChunkServer *state, int conn_idx, ByteView return 0; } -static int -process_metadata_server_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) -{ - switch (type) { - case MESSAGE_TYPE_AUTH_RESPONSE: return process_metadata_server_auth_response(state, conn_idx, msg); - case MESSAGE_TYPE_SYNC_2: return process_metadata_server_sync_2(state, conn_idx, msg); - case MESSAGE_TYPE_SYNC_4: return process_metadata_server_sync_4(state, conn_idx, msg); - } - return -1; -} - static int process_chunk_server_download_error(ChunkServer *state, int conn_idx, ByteView msg) { @@ -633,7 +513,8 @@ process_chunk_server_download_success(ChunkServer *state, int conn_idx, ByteView return -1; // Store the downloaded chunk - if (chunk_store_add(&state->store, data) < 0) + SHA256 dummy; + if (store_chunk(state, data, &dummy) < 0) return -1; // The download succeded! @@ -653,21 +534,6 @@ process_chunk_server_download_success(ChunkServer *state, int conn_idx, ByteView return 0; } -static int -process_chunk_server_message(ChunkServer *state, int conn_idx, uint16_t msg_type, ByteView msg) -{ - switch (msg_type) { - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR: - return process_chunk_server_download_error(state, conn_idx, msg); - - case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS: - return process_chunk_server_download_success(state, conn_idx, msg); - } - - return -1; -} - static int process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg) { @@ -710,7 +576,8 @@ process_client_create_chunk(ChunkServer *state, int conn_idx, ByteView msg) SHA256 new_hash; sha256(mem, chunk_size, (uint8_t*) new_hash.data); - int ret = chunk_store_add(&state->store, (string) { mem, chunk_size }); + SHA256 dummy; + int ret = store_chunk(state, (string) { mem, chunk_size }, &dummy); sys_free(mem); @@ -764,7 +631,7 @@ process_client_upload_chunk(ChunkServer *state, int conn_idx, ByteView msg) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_UPLOAD_CHUNK_ERROR, S("Invalid message")); SHA256 new_hash; - int ret = chunk_store_patch(&state->store, target_hash, target_off, data, &new_hash); + int ret = patch_chunk(state, target_hash, target_off, data, &new_hash); if (ret < 0) { // TODO: if the chunk is missing add it to the missing list @@ -799,62 +666,89 @@ process_client_download_chunk(ChunkServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, &target_hash, sizeof(target_hash))) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); - uint32_t target_off; - if (!binary_read(&reader, &target_off, sizeof(target_off))) + uint8_t full; + if (!binary_read(&reader, &full, sizeof(full))) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); + uint32_t target_off; uint32_t target_len; - if (!binary_read(&reader, &target_len, sizeof(target_len))) - return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); + if (full == 0) { + if (!binary_read(&reader, &target_off, sizeof(target_off))) + return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); + if (!binary_read(&reader, &target_len, sizeof(target_len))) + return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); + } // Check that there are no more bytes to read if (binary_read(&reader, NULL, 1)) return send_error(&state->tcp, conn_idx, true, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid message")); string data; - int ret = chunk_store_get(&state->store, target_hash, &data); + int ret = load_chunk(state, target_hash, &data); if (ret < 0) { // TODO: if the chunk is missing add it to the missing list return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("I/O error")); } - if (target_off >= (size_t) data.len || target_len > (size_t) data.len - target_off) { - sys_free(data.ptr); - return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid range")); + string slice; + if (full == 0) { + if (target_off >= (size_t) data.len || target_len > (size_t) data.len - target_off) { + sys_free(data.ptr); + return send_error(&state->tcp, conn_idx, false, MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR, S("Invalid range")); + } + slice = (string) { data.ptr + target_off, target_len }; + } else { + slice = data; } - string slice = { data.ptr + target_off, target_len }; - - MessageWriter writer; ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); + assert(output); + + MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS); - message_write(&writer, &target_len, sizeof(target_len)); - message_write(&writer, slice.ptr, slice.len); + if (!message_writer_free(&writer)) { + sys_free(data.ptr); + return -1; + } sys_free(data.ptr); - - if (!message_writer_free(&writer)) - return -1; return 0; } static int -process_client_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) +process_message(ChunkServer *state, int conn_idx, uint16_t type, ByteView msg) { - switch (type) { + switch (tcp_get_tag(&state->tcp, conn_idx)) { + case TAG_METADATA_SERVER: + switch (type) { + case MESSAGE_TYPE_AUTH_RESPONSE: + return process_metadata_server_auth_response(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_2: + return process_metadata_server_sync_2(state, conn_idx, msg); + case MESSAGE_TYPE_SYNC_4: + return process_metadata_server_sync_4(state, conn_idx, msg); + } + break; + case TAG_CHUNK_SERVER: + switch (type) { + case MESSAGE_TYPE_DOWNLOAD_CHUNK_ERROR: + return process_chunk_server_download_error(state, conn_idx, msg); + case MESSAGE_TYPE_DOWNLOAD_CHUNK_SUCCESS: + return process_chunk_server_download_success(state, conn_idx, msg); + } + break; + default: + switch (type) { case MESSAGE_TYPE_CREATE_CHUNK: - return process_client_create_chunk(state, conn_idx, msg); - + return process_client_create_chunk(state, conn_idx, msg); case MESSAGE_TYPE_UPLOAD_CHUNK: - return process_client_upload_chunk(state, conn_idx, msg); - + return process_client_upload_chunk(state, conn_idx, msg); case MESSAGE_TYPE_DOWNLOAD_CHUNK: - return process_client_download_chunk(state, conn_idx, msg); - - default: + return process_client_download_chunk(state, conn_idx, msg); + } break; } return -1; @@ -938,7 +832,6 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts return -1; state->trace = trace; - state->reconnect_delay = 1; // 1 second tcp_context_init(&state->tcp); @@ -949,8 +842,12 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts return -1; } - ret = chunk_store_init(&state->store, path); - if (ret < 0) { + if (create_dir(path) && errno != EEXIST) { + tcp_context_free(&state->tcp); + return -1; + } + + if (get_full_path(path, state->path) < 0) { tcp_context_free(&state->tcp); return -1; } @@ -971,7 +868,6 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts state->local_addr.is_ipv4 = true; if (inet_pton(AF_INET, tmp, &state->local_addr.ipv4) != 1) { tcp_context_free(&state->tcp); - chunk_store_free(&state->store); return -1; } state->local_addr.port = port; @@ -987,7 +883,6 @@ int chunk_server_init(ChunkServer *state, int argc, char **argv, void **contexts state->remote_addr.is_ipv4 = true; if (inet_pton(AF_INET, tmp, &state->remote_addr.ipv4) != 1) { tcp_context_free(&state->tcp); - chunk_store_free(&state->store); return -1; } state->remote_addr.port = remote_port; @@ -1017,7 +912,6 @@ int chunk_server_free(ChunkServer *state) timed_hash_set_free(&state->cs_rem_list); hash_set_free(&state->cs_lst_list); hash_set_free(&state->cs_add_list); - chunk_store_free(&state->store); tcp_context_free(&state->tcp); return 0; } @@ -1050,13 +944,7 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled break; case TAG_CHUNK_SERVER: - // Connection to chunk server disconnected during download - if (state->downloading) { - // Mark as not downloading and retry - state->downloading = false; - // The current download item will be retried on next call - // to start_download_if_necessary - } + state->downloading = false; break; } break; @@ -1078,20 +966,7 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled 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); - break; - - case TAG_CHUNK_SERVER: - ret = process_chunk_server_message(state, conn_idx, msg_type, msg); - break; - - default: - ret = process_client_message(state, conn_idx, msg_type, msg); - break; - } - + ret = process_message(state, conn_idx, msg_type, msg); if (ret < 0) { tcp_close(&state->tcp, conn_idx); break; @@ -1104,10 +979,7 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled } } - { - int conn_idx = tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER); - assert((conn_idx < 0) == (state->disconnect_time != INVALID_TIME)); - } + assert((tcp_index_from_tag(&state->tcp, TAG_METADATA_SERVER) < 0) == (state->disconnect_time != INVALID_TIME)); Time deadline = INVALID_TIME; @@ -1116,7 +988,9 @@ int chunk_server_step(ChunkServer *state, void **contexts, struct pollfd *polled TimedHash *removal = &state->cs_rem_list.items[i]; Time removal_time = removal->time + (Time) DELETION_TIMEOUT * 1000000000; if (removal_time < current_time) { - if (chunk_store_remove(&state->store, removal->hash) == 0) + char buf[PATH_MAX]; + string path = hash2path(state, removal->hash, buf); + if (remove_file_or_dir(path) == 0) *removal = state->cs_rem_list.items[--state->cs_rem_list.count]; } else { nearest_deadline(&deadline, removal_time); diff --git a/src/chunk_server.h b/src/chunk_server.h index a66ad9d..e91247e 100644 --- a/src/chunk_server.h +++ b/src/chunk_server.h @@ -10,10 +10,6 @@ #define TAG_METADATA_SERVER 1 #define TAG_CHUNK_SERVER 2 -typedef struct { - char path[PATH_MAX]; -} ChunkStore; - typedef struct { Address addr; SHA256 hash; @@ -27,19 +23,20 @@ typedef struct { typedef struct { + char path[PATH_MAX]; + bool trace; Address local_addr; - Address remote_addr; Time disconnect_time; Time last_sync_time; int reconnect_delay; // In seconds - TCP tcp; + // --- Subsystems --- - ChunkStore store; + TCP tcp; // --- Download Management --- diff --git a/src/client.c b/src/client.c index 4e8dd89..f4287fc 100644 --- a/src/client.c +++ b/src/client.c @@ -170,65 +170,30 @@ struct ToastyFS { Operation operations[MAX_OPERATIONS]; }; -static void request_queue_init(RequestQueue *reqs); - -ToastyFS *toasty_connect(ToastyString addr, uint16_t port) +static void request_queue_init(RequestQueue *reqs) { - ToastyFS *toasty = sys_malloc(sizeof(ToastyFS)); - if (toasty == NULL) - return NULL; - - Address addr2; - { - char tmp[128]; - if (addr.len >= (int) sizeof(tmp)) { - sys_free(toasty); - return NULL; - } - memcpy(tmp, addr.ptr, addr.len); - tmp[addr.len] = '\0'; - - addr2.is_ipv4 = true; - addr2.port = port; - if (inet_pton(AF_INET, tmp, &addr2.ipv4) != 1) { - sys_free(toasty); - return NULL; - } - } - - tcp_context_init(&toasty->tcp); - - if (tcp_connect(&toasty->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) { - tcp_context_free(&toasty->tcp); - sys_free(toasty); - return NULL; - } - - toasty->num_operations = 0; - - for (int i = 0; i < MAX_OPERATIONS; i++) { - toasty->operations[i].type = OPERATION_TYPE_FREE; - toasty->operations[i].generation = 1; // Important not to start from 0 !!! - } - - // Initialize metadata server (connected during init) - toasty->metadata_server.used = true; - toasty->metadata_server.addr = addr2; - request_queue_init(&toasty->metadata_server.reqs); - - // Initialize chunk servers array (connections created on demand) - toasty->num_chunk_servers = 0; - for (int i = 0; i < MAX_CHUNK_SERVERS; i++) { - toasty->chunk_servers[i].used = false; - } - - return toasty; + reqs->head = 0; + reqs->count = 0; } -void toasty_disconnect(ToastyFS *toasty) +static int request_queue_push(RequestQueue *reqs, Request req) { - tcp_context_free(&toasty->tcp); - sys_free(toasty); + if (reqs->count == MAX_REQUESTS_PER_QUEUE) + return -1; + int tail = (reqs->head + reqs->count) % MAX_REQUESTS_PER_QUEUE; + reqs->items[tail] = req; + reqs->count++; + return 0; +} + +static int request_queue_pop(RequestQueue *reqs, Request *req) +{ + if (reqs->count == 0) + return -1; + if (req) *req = reqs->items[reqs->head]; + reqs->head = (reqs->head + 1) % MAX_REQUESTS_PER_QUEUE; + reqs->count--; + return 0; } static int @@ -292,33 +257,96 @@ static int handle_to_operation(ToastyFS *toasty, ToastyHandle handle) return opidx; } +ToastyFS *toasty_connect(ToastyString addr, uint16_t port) +{ + ToastyFS *toasty = sys_malloc(sizeof(ToastyFS)); + if (toasty == NULL) + return NULL; + + Address addr2; + { + char tmp[128]; + if (addr.len >= (int) sizeof(tmp)) { + sys_free(toasty); + return NULL; + } + memcpy(tmp, addr.ptr, addr.len); + tmp[addr.len] = '\0'; + + addr2.is_ipv4 = true; + addr2.port = port; + if (inet_pton(AF_INET, tmp, &addr2.ipv4) != 1) { + sys_free(toasty); + return NULL; + } + } + + tcp_context_init(&toasty->tcp); + + if (tcp_connect(&toasty->tcp, addr2, TAG_METADATA_SERVER, NULL) < 0) { + tcp_context_free(&toasty->tcp); + sys_free(toasty); + return NULL; + } + + toasty->num_operations = 0; + + for (int i = 0; i < MAX_OPERATIONS; i++) { + toasty->operations[i].type = OPERATION_TYPE_FREE; + toasty->operations[i].generation = 1; // Important not to start from 0 !!! + } + + // Initialize metadata server (connected during init) + toasty->metadata_server.used = true; + toasty->metadata_server.addr = addr2; + request_queue_init(&toasty->metadata_server.reqs); + + // Initialize chunk servers array (connections created on demand) + toasty->num_chunk_servers = 0; + for (int i = 0; i < MAX_CHUNK_SERVERS; i++) { + toasty->chunk_servers[i].used = false; + } + + return toasty; +} + static void -request_queue_init(RequestQueue *reqs) +metadata_server_request_start(ToastyFS *toasty, MessageWriter *writer, uint16_t type) { - reqs->head = 0; - reqs->count = 0; + ByteQueue *output; + if (toasty->metadata_server.used) { + + int conn_idx = tcp_index_from_tag(&toasty->tcp, TAG_METADATA_SERVER); + assert(conn_idx > -1); + + output = tcp_output_buffer(&toasty->tcp, conn_idx); + } else { + if (tcp_connect(&toasty->tcp, toasty->metadata_server.addr, TAG_METADATA_SERVER, &output) < 0) { + assert(0); // TODO + } + toasty->metadata_server.used = true; + } + + message_writer_init(writer, output, type); } static int -request_queue_push(RequestQueue *reqs, Request req) +metadata_server_request_end(ToastyFS *toasty, MessageWriter *writer, int opidx, int tag) { - if (reqs->count == MAX_REQUESTS_PER_QUEUE) + if (!message_writer_free(writer)) return -1; - int tail = (reqs->head + reqs->count) % MAX_REQUESTS_PER_QUEUE; - reqs->items[tail] = req; - reqs->count++; + + RequestQueue *reqs = &toasty->metadata_server.reqs; + if (request_queue_push(reqs, (Request) { tag, opidx }) < 0) + return -1; + return 0; } -static int -request_queue_pop(RequestQueue *reqs, Request *req) +void toasty_disconnect(ToastyFS *toasty) { - if (reqs->count == 0) - return -1; - if (req) *req = reqs->items[reqs->head]; - reqs->head = (reqs->head + 1) % MAX_REQUESTS_PER_QUEUE; - reqs->count--; - return 0; + tcp_context_free(&toasty->tcp); + sys_free(toasty); } static bool @@ -387,21 +415,30 @@ static int get_chunk_server(ToastyFS *toasty, Address *addrs, int num_addrs, Byt return found; } +// TODO: is this used anywhere? +static void close_chunk_server(ToastyFS *toasty, int chunk_server_idx) +{ + int conn_idx = tcp_index_from_tag(&toasty->tcp, chunk_server_idx); + tcp_close(&toasty->tcp, conn_idx); +} + // Send download request for a chunk static int send_download_chunk(ToastyFS *toasty, int chunk_server_idx, SHA256 hash, uint32_t offset, uint32_t length, int opidx, int range_idx) { int conn_idx = tcp_index_from_tag(&toasty->tcp, chunk_server_idx); - if (conn_idx < 0) return -1; + if (conn_idx < 0) + return -1; + + ByteQueue *output = tcp_output_buffer(&toasty->tcp, conn_idx); + assert(output); MessageWriter writer; - ByteQueue *output = tcp_output_buffer(&toasty->tcp, conn_idx); message_writer_init(&writer, output, MESSAGE_TYPE_DOWNLOAD_CHUNK); - - message_write(&writer, &hash, sizeof(hash)); - message_write(&writer, &offset, sizeof(offset)); - message_write(&writer, &length, sizeof(length)); - + message_write_hash(&writer, hash); + message_write_u8 (&writer, 0); + message_write_u32 (&writer, offset); + message_write_u32 (&writer, length); if (!message_writer_free(&writer)) return -1; @@ -409,50 +446,11 @@ static int send_download_chunk(ToastyFS *toasty, int chunk_server_idx, return request_queue_push(reqs, (Request) { range_idx, opidx }); } -// TODO: is this used somewhere? -static void close_chunk_server(ToastyFS *toasty, int chunk_server_idx) -{ - int conn_idx = tcp_index_from_tag(&toasty->tcp, chunk_server_idx); - tcp_close(&toasty->tcp, conn_idx); -} - -static void -metadata_server_request_start(ToastyFS *toasty, MessageWriter *writer, uint16_t type) -{ - ByteQueue *output; - if (toasty->metadata_server.used) { - - int conn_idx = tcp_index_from_tag(&toasty->tcp, TAG_METADATA_SERVER); - assert(conn_idx > -1); - - output = tcp_output_buffer(&toasty->tcp, conn_idx); - } else { - if (tcp_connect(&toasty->tcp, toasty->metadata_server.addr, TAG_METADATA_SERVER, &output) < 0) { - assert(0); // TODO - } - toasty->metadata_server.used = true; - } - - message_writer_init(writer, output, type); -} - -static int -metadata_server_request_end(ToastyFS *toasty, MessageWriter *writer, int opidx, int tag) -{ - if (!message_writer_free(writer)) - return -1; - - RequestQueue *reqs = &toasty->metadata_server.reqs; - if (request_queue_push(reqs, (Request) { tag, opidx }) < 0) - return -1; - - return 0; -} - static ToastyHandle begin_create(ToastyFS *toasty, ToastyString path, bool is_dir, uint32_t chunk_size) { OperationType type = OPERATION_TYPE_CREATE; + int opidx = alloc_operation(toasty, type, 0, NULL, 0); if (opidx < 0) return TOASTY_INVALID; @@ -500,9 +498,10 @@ ToastyHandle toasty_begin_create_file(ToastyFS *toasty, ToastyString path, return begin_create(toasty, path, false, chunk_size); } -ToastyHandle toastyfs_begin_delete(ToastyFS *toasty, ToastyString path) +ToastyHandle toasty_begin_delete(ToastyFS *toasty, ToastyString path) { OperationType type = OPERATION_TYPE_DELETE; + int opidx = alloc_operation(toasty, type, 0, NULL, 0); if (opidx < 0) return TOASTY_INVALID; @@ -525,9 +524,10 @@ ToastyHandle toastyfs_begin_delete(ToastyFS *toasty, ToastyString path) return operation_to_handle(toasty, opidx); } -ToastyHandle toastyfs_begin_list(ToastyFS *toasty, ToastyString path) +ToastyHandle toasty_begin_list(ToastyFS *toasty, ToastyString path) { OperationType type = OPERATION_TYPE_LIST; + int opidx = alloc_operation(toasty, type, 0, NULL, 0); if (opidx < 0) return TOASTY_INVALID; @@ -569,9 +569,10 @@ static int send_read_message(ToastyFS *toasty, int opidx, int tag, ToastyString return 0; } -ToastyHandle toastyfs_begin_read(ToastyFS *toasty, ToastyString path, int off, void *dst, int len) +ToastyHandle toasty_begin_read(ToastyFS *toasty, ToastyString path, int off, void *dst, int len) { OperationType type = OPERATION_TYPE_READ; + int opidx = alloc_operation(toasty, type, off, dst, len); if (opidx < 0) return TOASTY_INVALID; @@ -584,7 +585,7 @@ ToastyHandle toastyfs_begin_read(ToastyFS *toasty, ToastyString path, int off, v return operation_to_handle(toasty, opidx); } -ToastyHandle toastyfs_begin_write(ToastyFS *toasty, ToastyString path, int off, void *src, int len) +ToastyHandle toasty_begin_write(ToastyFS *toasty, ToastyString path, int off, void *src, int len) { OperationType type = OPERATION_TYPE_WRITE; @@ -602,10 +603,10 @@ ToastyHandle toastyfs_begin_write(ToastyFS *toasty, ToastyString path, int off, return operation_to_handle(toasty, opidx); } -void toastyfs_result_free(ToastyResult *result) +void toasty_free_result(ToastyResult *result) { if (result->type == TOASTY_RESULT_LIST_SUCCESS) - sys_free(result->entities); + toasty_free_listing(&result->listing); } static void process_event_for_create(ToastyFS *toasty, @@ -791,7 +792,7 @@ static void process_event_for_list(ToastyFS *toasty, return; } - toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_SUCCESS, item_count, entities }; + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_SUCCESS, .listing=(ToastyListing) { item_count, entities } }; } static void process_event_for_read(ToastyFS *toasty, @@ -1738,7 +1739,7 @@ static void process_event_for_write(ToastyFS *toasty, uint32_t length = toasty->operations[opidx].len; if (path.len > UINT16_MAX) { - // TODO + assert(0); // TODO } uint16_t path_len = path.len; @@ -1888,7 +1889,7 @@ int toasty_get_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *resul return 1; } -int toastyfs_process_events(ToastyFS *toasty, void **contexts, struct pollfd *polled, int num_polled) +int toasty_process_events(ToastyFS *toasty, void **contexts, struct pollfd *polled, int num_polled) { int num_events; Event events[MAX_CONNS+1]; @@ -2018,7 +2019,7 @@ int toasty_wait_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *resu struct pollfd polled[MAX_CONNS+1]; int num_polled; - num_polled = toastyfs_process_events(toasty, contexts, polled, 0); + num_polled = toasty_process_events(toasty, contexts, polled, 0); for (;;) { @@ -2050,10 +2051,135 @@ int toasty_wait_result(ToastyFS *toasty, ToastyHandle handle, ToastyResult *resu if (ret < 0) return -1; - num_polled = toastyfs_process_events(toasty, contexts, polled, num_polled); + num_polled = toasty_process_events(toasty, contexts, polled, num_polled); if (num_polled < 0) return -1; } return 0; } + +int toasty_create_dir(ToastyFS *toasty, ToastyString path) +{ + ToastyHandle handle = toasty_begin_create_dir(toasty, path); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_CREATE_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + toasty_free_result(&result); + return 0; +} + +int toasty_create_file(ToastyFS *toasty, ToastyString path, + unsigned int chunk_size) +{ + ToastyHandle handle = toasty_begin_create_file(toasty, path, chunk_size); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_CREATE_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + toasty_free_result(&result); + return 0; +} + +int toasty_delete(ToastyFS *toasty, ToastyString path) +{ + ToastyHandle handle = toasty_begin_delete(toasty, path); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_DELETE_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + toasty_free_result(&result); + return 0; +} + +int toasty_list(ToastyFS *toasty, ToastyString path, ToastyListing *listing) +{ + ToastyHandle handle = toasty_begin_list(toasty, path); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_LIST_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + if (listing) + *listing = result.listing; + else + toasty_free_result(&result); + return 0; +} + +void toasty_free_listing(ToastyListing *listing) +{ + sys_free(listing->items); +} + +int toasty_read(ToastyFS *toasty, ToastyString path, + int off, void *dst, int len) +{ + ToastyHandle handle = toasty_begin_read(toasty, path, off, dst, len); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_READ_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + toasty_free_result(&result); + return 0; // TODO: return the number of bytes read? +} + +int toasty_write(ToastyFS *toasty, ToastyString path, + int off, void *src, int len) +{ + ToastyHandle handle = toasty_begin_write(toasty, path, off, src, len); + if (handle == TOASTY_INVALID) + return -1; + + ToastyResult result; + if (toasty_wait_result(toasty, handle, &result, -1) < 0) + return -1; + + if (result.type != TOASTY_RESULT_WRITE_SUCCESS) { + toasty_free_result(&result); + return -1; + } + + toasty_free_result(&result); + return 0; // TODO: return the number of bytes written? +} diff --git a/src/message.c b/src/message.c index cb0bb0f..cd948c4 100644 --- a/src/message.c +++ b/src/message.c @@ -18,6 +18,48 @@ bool binary_read(BinaryReader *reader, void *dst, int len) return true; } +bool binary_read_addr_ipv4(BinaryReader *reader, Address *addr) +{ + if (!binary_read(reader, &addr->ipv4, sizeof(IPv4))) return false; + if (!binary_read(reader, &addr->port, sizeof(uint16_t))) return false; + addr->is_ipv4 = true; + return true; +} + +bool binary_read_addr_ipv6(BinaryReader *reader, Address *addr) +{ + if (!binary_read(reader, &addr->ipv6, sizeof(IPv6))) return false; + if (!binary_read(reader, &addr->port, sizeof(uint16_t))) return false; + addr->is_ipv4 = false; + return true; +} + +int binary_read_addr_list(BinaryReader *reader, Address *addrs, int max_addrs) +{ + uint32_t num_ipv4; + uint32_t num_ipv6; + if (!binary_read(reader, &num_ipv4, sizeof(num_ipv4))) + return -1; + if (!binary_read(reader, &num_ipv6, sizeof(num_ipv6))) + return -1; + int num = 0; + for (uint32_t i = 0; i < num_ipv4; i++) { + Address tmp; + if (!binary_read_addr_ipv4(reader, &tmp)) + return -1; + if (num < max_addrs) + addrs[num++] = tmp; + } + for (uint32_t i = 0; i < num_ipv6; i++) { + Address tmp; + if (!binary_read_addr_ipv6(reader, &tmp)) + return -1; + if (num < max_addrs) + addrs[num++] = tmp; + } + return num; +} + void message_writer_init(MessageWriter *writer, ByteQueue *output, uint16_t type) { uint16_t version = MESSAGE_VERSION; @@ -44,6 +86,21 @@ 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)) diff --git a/src/message.h b/src/message.h index 1a06e72..3740ad2 100644 --- a/src/message.h +++ b/src/message.h @@ -4,6 +4,7 @@ #include #include +#include "basic.h" #include "byte_queue.h" enum { @@ -74,9 +75,16 @@ typedef struct { bool binary_read(BinaryReader *reader, void *dst, int len); +bool binary_read_addr_ipv4(BinaryReader *reader, Address *addr); +bool binary_read_addr_ipv6(BinaryReader *reader, Address *addr); +int binary_read_addr_list(BinaryReader *reader, Address *addrs, int max_addr); + 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); diff --git a/src/simulation_client.c b/src/simulation_client.c index 2adec38..6f5f0d4 100644 --- a/src/simulation_client.c +++ b/src/simulation_client.c @@ -34,8 +34,8 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv, uint16_t port; parse_server_addr(argc, argv, &addr, &port); - client->tfs = toastyfs_init(addr, port); - if (client->tfs == NULL) + client->toasty = toasty_connect((ToastyString) { addr, strlen(addr) }, port); + if (client->toasty == NULL) return -1; client->num_pending = 0; @@ -43,7 +43,7 @@ int simulation_client_init(SimulationClient *client, int argc, char **argv, printf("Client set up (remote=%s:%d)\n", addr, port); *timeout = 0; - return toastyfs_process_events(client->tfs, contexts, polled, 0); + return toasty_process_events(client->toasty, contexts, polled, 0); } static int random_in_range(int min, int max) @@ -56,96 +56,96 @@ int simulation_client_step(SimulationClient *client, void **contexts, struct pollfd *polled, int num_polled, int *timeout) { // Process any pending events from the network and get new poll descriptors - num_polled = toastyfs_process_events(client->tfs, contexts, polled, num_polled); + num_polled = toasty_process_events(client->toasty, contexts, polled, num_polled); for (int i = 0; i < client->num_pending; i++) { - ToastyFS_Result result; - if (!toastyfs_isdone(client->tfs, client->pending[i].opidx, &result)) + ToastyResult result; + if (!toasty_get_result(client->toasty, client->pending[i].handle, &result)) continue; PendingOperation pending = client->pending[i]; switch (result.type) { - case TOASTYFS_RESULT_EMPTY: + case TOASTY_RESULT_EMPTY: assert(0); break; - case TOASTYFS_RESULT_CREATE_ERROR: + case TOASTY_RESULT_CREATE_ERROR: assert(pending.type == PENDING_OPERATION_CREATE); //printf("[Client] create error\n"); break; - case TOASTYFS_RESULT_CREATE_SUCCESS: + case TOASTY_RESULT_CREATE_SUCCESS: assert(pending.type == PENDING_OPERATION_CREATE); //printf("[Client] create success\n"); break; - case TOASTYFS_RESULT_DELETE_ERROR: + case TOASTY_RESULT_DELETE_ERROR: assert(pending.type == PENDING_OPERATION_DELETE); //printf("[Client] delete error\n"); break; - case TOASTYFS_RESULT_DELETE_SUCCESS: + case TOASTY_RESULT_DELETE_SUCCESS: assert(pending.type == PENDING_OPERATION_DELETE); //printf("[Client] delete success\n"); break; - case TOASTYFS_RESULT_LIST_ERROR: + case TOASTY_RESULT_LIST_ERROR: assert(pending.type == PENDING_OPERATION_LIST); //printf("[Client] list error\n"); break; - case TOASTYFS_RESULT_LIST_SUCCESS: + case TOASTY_RESULT_LIST_SUCCESS: assert(pending.type == PENDING_OPERATION_LIST); //printf("[Client] list success\n"); break; - case TOASTYFS_RESULT_READ_ERROR: + case TOASTY_RESULT_READ_ERROR: assert(pending.type == PENDING_OPERATION_READ); //printf("[Client] read error\n"); break; - case TOASTYFS_RESULT_READ_SUCCESS: + case TOASTY_RESULT_READ_SUCCESS: assert(pending.type == PENDING_OPERATION_READ); //printf("[Client] read success\n"); break; - case TOASTYFS_RESULT_WRITE_ERROR: + case TOASTY_RESULT_WRITE_ERROR: assert(pending.type == PENDING_OPERATION_WRITE); //printf("[Client] write error\n"); break; - case TOASTYFS_RESULT_WRITE_SUCCESS: + case TOASTY_RESULT_WRITE_SUCCESS: assert(pending.type == PENDING_OPERATION_WRITE); //printf("[Client] write success\n"); break; } free(pending.ptr); - toastyfs_result_free(&result); + toasty_free_result(&result); client->pending[i--] = client->pending[--client->num_pending]; } while (client->num_pending < MAX_PENDING_OPERATION) { typedef struct { - char *path; + ToastyString path; bool is_dir; } TableEntry; static const TableEntry table[] = { - { "/f0", false }, - { "/f1", false }, - { "/d0", true }, - { "/d1", true }, - { "/d0/f1", false }, - { "/d0/f2", false }, - { "/d0/d0", true }, - { "/d0/d1", true }, - { "/d1/f1", false }, - { "/d1/f2", false }, - { "/d1/d0", true }, - { "/d1/d1", true }, + { TOASTY_STR("/f0"), false }, + { TOASTY_STR("/f1"), false }, + { TOASTY_STR("/d0"), true }, + { TOASTY_STR("/d1"), true }, + { TOASTY_STR("/d0/f1"), false }, + { TOASTY_STR("/d0/f2"), false }, + { TOASTY_STR("/d0/d0"), true }, + { TOASTY_STR("/d0/d1"), true }, + { TOASTY_STR("/d1/f1"), false }, + { TOASTY_STR("/d1/f2"), false }, + { TOASTY_STR("/d1/d0"), true }, + { TOASTY_STR("/d1/d1"), true }, }; static const int table_len = sizeof(table) / sizeof(table[0]); @@ -161,7 +161,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, void *ptr = NULL; int off = 0; int len = 0; - int ret; + ToastyHandle handle; PendingOperationType type = type_table[random_in_range(0, type_table_len-1)]; switch (type) { @@ -171,34 +171,24 @@ int simulation_client_step(SimulationClient *client, void **contexts, case PENDING_OPERATION_CREATE: entry = table[random_in_range(0, table_len-1)]; - chunk_size = entry.is_dir ? 0 : random_in_range(0, 5000); - ret = toastyfs_submit_create( - client->tfs, - entry.path, - -1, - entry.is_dir, - chunk_size - ); + if (entry.is_dir) { + handle = toasty_begin_create_dir(client->toasty, entry.path); + } else { + chunk_size = random_in_range(0, 5000); + handle = toasty_begin_create_file(client->toasty, entry.path, chunk_size); + } //printf("[Client] submit create (path=%s, is_dir=%s, chunk_size=%d)\n", entry.path, entry.is_dir ? "true" : "false", chunk_size); break; case PENDING_OPERATION_DELETE: entry = table[random_in_range(0, table_len-1)]; - ret = toastyfs_submit_delete( - client->tfs, - entry.path, - -1 - ); + handle = toasty_begin_delete(client->toasty, entry.path); //printf("[Client] submit delete (path=%s)\n", entry.path); break; case PENDING_OPERATION_LIST: entry = table[random_in_range(0, table_len-1)]; - ret = toastyfs_submit_list( - client->tfs, - entry.path, - -1 - ); + handle = toasty_begin_list(client->toasty, entry.path); //printf("[Client] submit list (path=%s)\n", entry.path); break; @@ -208,13 +198,7 @@ int simulation_client_step(SimulationClient *client, void **contexts, len = random_in_range(0, 5000); ptr = malloc(len); if (ptr == NULL) assert(0); - ret = toastyfs_submit_read(client->tfs, - entry.path, - -1, - off, - ptr, - len - ); + handle = toasty_begin_read(client->toasty, entry.path, off, ptr, len); //printf("[Client] submit read (path=%s, off=%d, len=%d)\n", entry.path, off, len); break; @@ -225,37 +209,25 @@ int simulation_client_step(SimulationClient *client, void **contexts, ptr = malloc(len); if (ptr == NULL) assert(0); memset(ptr, 'a', len); - ret = toastyfs_submit_write( - client->tfs, - entry.path, - -1, - off, - ptr, - len - ); + handle = toasty_begin_write(client->toasty, entry.path, off, ptr, len); //printf("[Client] submit write (path=%s, off=%d, len=%d)\n", entry.path, off, len); break; } - if (ret < 0) + if (handle == TOASTY_INVALID) break; - PendingOperation pending = { - .type = type, - .opidx = ret, - .ptr = ptr, - }; - client->pending[client->num_pending++] = pending; + client->pending[client->num_pending++] = (PendingOperation) { .type=type, .handle=handle, .ptr=ptr }; } if (client->num_pending == 0) *timeout = 10; else *timeout = -1; - return toastyfs_process_events(client->tfs, contexts, polled, 0); + return toasty_process_events(client->toasty, contexts, polled, 0); } void simulation_client_free(SimulationClient *client) { - toastyfs_free(client->tfs); + toasty_disconnect(client->toasty); } #endif // BUILD_TEST diff --git a/src/simulation_client.h b/src/simulation_client.h index b1ce9e8..d6a5c2d 100644 --- a/src/simulation_client.h +++ b/src/simulation_client.h @@ -24,12 +24,12 @@ typedef enum { typedef struct { PendingOperationType type; - int opidx; + ToastyHandle handle; void *ptr; } PendingOperation; typedef struct { - ToastyFS *tfs; + ToastyFS *toasty; int num_pending; PendingOperation pending[MAX_PENDING_OPERATION]; } SimulationClient; diff --git a/src/tcp.h b/src/tcp.h index 2a41a19..43d5765 100644 --- a/src/tcp.h +++ b/src/tcp.h @@ -26,15 +26,6 @@ typedef struct { int tag; } Event; -typedef struct { - union { - IPv4 ipv4; - IPv6 ipv6; - }; - bool is_ipv4; - uint16_t port; -} Address; - typedef struct { SOCKET fd; int tag;