diff --git a/build.sh b/build.sh index 39ee41e..d45af7c 100644 --- a/build.sh +++ b/build.sh @@ -1,3 +1,3 @@ -gcc lib/basic.c lib/file_system.c lib/byte_queue.c lib/message.c lib/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/invariant_checker.c src/chunk_store.c src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c -o toastyfs_simulation -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION -gcc lib/basic.c lib/file_system.c lib/byte_queue.c lib/message.c lib/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_SERVER -gcc lib/basic.c lib/file_system.c lib/byte_queue.c lib/message.c lib/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs_client -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_CLIENT +gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/invariant_checker.c src/chunk_store.c src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c -o toastyfs_simulation -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION +gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_SERVER +gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/blob_client.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs_client -Iquakey/include -I. -Wall -Wextra -ggdb -O0 -DMAIN_CLIENT diff --git a/include/lib.h b/include/lib.h new file mode 100644 index 0000000..306fe76 --- /dev/null +++ b/include/lib.h @@ -0,0 +1,47 @@ +#ifndef TOASTYFS_INCLUDED +#define TOASTYFS_INCLUDED + +typedef enum { + TOASTYFS_RESULT_VOID, + TOASTYFS_RESULT_PUT, + TOASTYFS_RESULT_GET, + TOASTYFS_RESULT_DELETE, +} ToastyFS_ResultType; + +typedef enum { + TOASTYFS_ERROR_VOID, + TOASTYFS_ERROR_OUT_OF_MEMORY, +} ToastyFS_Error; + +typedef struct { + ToastyFS_ResultType type; + ToastyFS_Error error; + char *data; + int size; +} ToastyFS_Result; + +typedef struct ToastyFS ToastyFS; + +ToastyFS* toastyfs_init(char *addrs, int num_addrs); +void toastyfs_free(ToastyFS *tfs); + +void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pnum); +int toastyfs_register_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pcap); + +int toastyfs_async_put(ToastyFS *tfs, char *key, int key_len, + char *data, int data_len); + +int toastyfs_async_get(ToastyFS *tfs, char *key, int key_len); + +int toastyfs_async_delete(ToastyFS *tfs, char *key, int key_len); + +ToastyFS_Result toastyfs_get_result(ToastyFS *tfs); + +int toastyfs_put(ToastyFS *tfs, char *key, int key_len, + char *data, int data_len, ToastyFS_Result *res); + +int toastyfs_get(ToastyFS *tfs, char *key, int key_len); + +int toastyfs_delete(ToastyFS *tfs, char *key, int key_len); + +#endif // TOASTYFS_INCLUDED \ No newline at end of file diff --git a/lib/basic.c b/src/basic.c similarity index 100% rename from lib/basic.c rename to src/basic.c diff --git a/lib/basic.h b/src/basic.h similarity index 100% rename from lib/basic.h rename to src/basic.h diff --git a/src/blob_client.c b/src/blob_client.c deleted file mode 100644 index 6e03794..0000000 --- a/src/blob_client.c +++ /dev/null @@ -1,675 +0,0 @@ -#if defined(MAIN_SIMULATION) || defined(MAIN_TEST) -#define QUAKEY_ENABLE_MOCKS -#endif - -#include -#include -#include - -#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; -} diff --git a/src/blob_client.h b/src/blob_client.h deleted file mode 100644 index ff70d49..0000000 --- a/src/blob_client.h +++ /dev/null @@ -1,86 +0,0 @@ -#ifndef BLOB_CLIENT_INCLUDED -#define BLOB_CLIENT_INCLUDED - -#include -#include - -#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 diff --git a/lib/byte_queue.c b/src/byte_queue.c similarity index 100% rename from lib/byte_queue.c rename to src/byte_queue.c diff --git a/lib/byte_queue.h b/src/byte_queue.h similarity index 100% rename from lib/byte_queue.h rename to src/byte_queue.h diff --git a/src/chunk_store.c b/src/chunk_store.c index 8b4ccd4..30ef06e 100644 --- a/src/chunk_store.c +++ b/src/chunk_store.c @@ -6,7 +6,7 @@ #include #include "chunk_store.h" -#include +#include "file_system.h" // Build the full path for a chunk: "base_path/HEX_HASH" // SHA256 hex = 64 chars. Returns string pointing into buf. diff --git a/src/chunk_store.h b/src/chunk_store.h index 4a45d08..e04cae0 100644 --- a/src/chunk_store.h +++ b/src/chunk_store.h @@ -4,7 +4,7 @@ #include #include -#include +#include "basic.h" typedef struct { char base_path[256]; diff --git a/src/client.c b/src/client.c index f3ab5ab..b008695 100644 --- a/src/client.c +++ b/src/client.c @@ -6,235 +6,439 @@ #include #include -#include "client.h" -#include "server.h" +#include "tcp.h" +#include "client_lib.h" -//#define CLIENT_TRACE(fmt, ...) {} -#define CLIENT_TRACE(fmt, ...) fprintf(stderr, "CLIENT: " fmt "\n", ##__VA_ARGS__); +typedef enum { + STEP_IDLE, + STEP_STORE_CHUNK, + STEP_COMMIT, + STEP_DELETE, + STEP_GET, + STEP_GET_DONE, + STEP_PUT_DONE, + STEP_DELETE_DONE, +} Step; -#define KEY_POOL_SIZE 128 -#define BUCKET_POOL_SIZE 4 +typedef enum { + TRANSFER_PENDING, + TRANSFER_STARTED, + TRANSFER_COMPLETED, + TRANSFER_ABORTED, +} TransferState; -static uint64_t next_client_id = 1; +typedef struct { + TransferState state; + SHA256 hash; + char* data; + int size; + int location; +} Transfer; -static uint64_t client_random(void) +struct ToastyFS { + + TCP tcp; + + Address server_addrs[NODE_LIMIT]; + int num_servers; + + uint64_t view_number; + uint64_t client_id; + uint64_t request_id; + + Step step; + + Transfer transfers[META_CHUNKS_MAX * REPLICATION_FACTOR]; + int num_transfers; + + SHA256 chunks[META_CHUNKS_MAX]; + int num_chunks; +}; + +int toastyfs_init(ToastyFS *tfs, uint64_t client_id, char **addrs, int num_addrs) { -#if defined(MAIN_SIMULATION) || defined(MAIN_TEST) - return quakey_random(); -#else - return (uint64_t)rand(); -#endif -} + tfs->view_number = 0; + tfs->client_id = client_id; + tfs->request_id = 0; -static MetaOper random_oper(void) -{ - MetaOper oper = {0}; - snprintf(oper.bucket, META_BUCKET_MAX, "b%d", (int)(client_random() % BUCKET_POOL_SIZE)); - snprintf(oper.key, META_KEY_MAX, "k%d", (int)(client_random() % KEY_POOL_SIZE)); - - switch (client_random() % 2) { - case 0: - oper.type = META_OPER_PUT; - oper.size = client_random() % (1 << 20); - oper.num_chunks = 1 + client_random() % 3; - for (uint32_t i = 0; i < oper.num_chunks; i++) { - for (int j = 0; j < 32; j++) - oper.chunks[i].hash.data[j] = client_random() & 0xFF; - oper.chunks[i].size = client_random() % (4 << 20); - } - for (int j = 0; j < 32; j++) - oper.content_hash.data[j] = client_random() & 0xFF; - break; - case 1: - oper.type = META_OPER_DELETE; - break; + for (int i = 0; i < num_addrs; i++) { + if (parse_addr_arg(addrs[i], &tfs->server_addrs[i]) < 0) + return -1; } - return oper; + addr_sort(tfs->server_addrs, tfs->num_servers); + tfs->num_servers = num_addrs; + + if (tcp_context_init(&tfs->tcp) < 0) + return -1; + + tfs->step = STEP_IDLE; + return 0; } -// Format time as seconds with 3 decimal places for trace output -#define TIME_FMT "%7.3fs" -#define TIME_VAL(t) ((double)(t) / 1000000000.0) - -static void client_log_impl(ClientState *state, Time now, const char *event, const char *detail) +void toastyfs_free(ToastyFS *tfs) { - printf("[" TIME_FMT "] CLIENT %lu | V%-3lu | %-20s %s\n", - TIME_VAL(now), - state->client_id, - state->view_number, - event, - detail ? detail : ""); + tcp_context_free(&tfs->tcp); } -#define client_log(state, now, event, fmt, ...) do { \ - char _detail[256]; \ - snprintf(_detail, sizeof(_detail), fmt, ##__VA_ARGS__); \ - client_log_impl(state, now, event, _detail); \ -} while (0) - -#define client_log_simple(state, now, event) \ - client_log_impl(state, now, event, NULL) - -static int leader_idx(ClientState *state) +static bool +transfer_for_hash_already_started(ToastyFS *tfs, SHA256 hash) { - return state->view_number % state->num_servers; + for (int j = 0; j < tfs->num_transfers; j++) { + Transfer *transfer = &tfs->transfers[j]; + if (!memcmp(&hash, &transfer->hash, sizeof(SHA256)) && transfer->state == TRANSFER_STARTED) { + return true; + } + } + return false; } -static int -process_message(ClientState *state, +static bool transfer_should_start(ToastyFS *tfs, Transfer *transfer) +{ + return transfer->state == TRANSFER_WAITING + && !transfer_for_hash_already_started(tfs, transfer->hash); +} + +static void add_transfer(ToastyFS *tfs, SHA256 hash, int location, char *data, int size) +{ + assert(tfs->num_transfers < xxx); + Transfer *transfer = &tfs->transfers[tfs->num_transfers]; + transfer->state = TRANSFER_PENDING; + transfer->hash = hash; + transfer->data = data; + transfer->size = size; + transfer->location = location; + tfs->num_transfers++; +} + +static void send_message_to_server(ToastyFS *tfs, int server_idx, Message *msg) +{ + int conn_idx = tcp_index_from_tag(&tfs->tcp, server_idx); + if (conn_idx < 0) { + tcp_connect(&tfs->tcp, tfs->server_addrs[server_idx], server_idx, NULL); + return; + } + + ByteQueue *output = tcp_output_buffer(&tfs->tcp, conn_idx); + if (output == NULL) + return; + + byte_queue_write(output, msg, msg->length); +} + +static int begin_transfers(ToastyFS *tfs) +{ + int num = 0; + for (int i = 0; i < tfs->num_transfers; i++) { + if (transfer_should_start(tfs, tfs->transfers[i])) { + + FetchChunkMessage msg = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_FETCH_CHUNK, + .length = sizeof(FetchChunkMessage), + }, + .hash = tfs->chunks[chunk_idx].hash, + .sender_idx = -1, // Client (not a peer server) + }; + send_message_to_server(tfs, server_idx, &msg); + + transfer->state = TRANSFER_STARTED; + num++; + } + } + + return num; +} + +static int find_started_transfer_by_hash(ToastyFS *tfs, SHA256 hash) +{ + for (int i = 0; i < tfs->num_transfers; i++) + if (!memcmp(&tfs->transfers[i].hash, &hash, sizeof(SHA256)) + && tfs->transfers[i].state == TRANSFER_STARTED) + return i; + return -1; +} + +static void +mark_waiting_transfers_for_hash_as_aborted(ToastyFS *tfs, SHA256 hash) +{ + for (int i = 0; i < tfs->num_transfers; i++) { + if (!memcmp(&tfs->transfers[i].hash, &hash, sizeof(SHA256)) + && tfs->transfers[i].state == TRANSFER_WAITING) + tfs->transfers[i].state = TRANSFER_ABORTED; + } +} + +static int process_message(ToastyFS *tfs, int conn_idx, uint8_t type, ByteView msg) { (void) conn_idx; - if (type == MESSAGE_TYPE_REDIRECT) { - RedirectMessage redirect_message; - if (msg.len != sizeof(RedirectMessage)) - return -1; - memcpy(&redirect_message, msg.ptr, sizeof(redirect_message)); + switch (tfs->step) { + case STEP_FETCH_CHUNK: + { + if (type == MESSAGE_TYPE_FETCH_CHUNK_RESPONSE) { - if (redirect_message.view_number > state->view_number) { - Time now = get_current_time(); - client_log(state, now, "RECV REDIRECT", "view=%lu -> %lu leader=%d", - (unsigned long)state->view_number, - (unsigned long)redirect_message.view_number, - (int)(redirect_message.view_number % state->num_servers)); - state->view_number = redirect_message.view_number; - state->last_was_rejected = true; - state->last_was_timeout = false; - state->pending = false; + FetchChunkResponseMessage resp; + if (msg.len < sizeof(FetchChunkResponseMessage)) + return -1; + memcpy(&resp, msg.ptr, sizeof(resp)); + char *data = msg.ptr + sizeof(resp); + + int transfer_idx = find_started_transfer_by_hash(tfs, reap.hash); + assert(transfer_idx > -1); + + if (resp.size) { + tfs->transfers[transfer_idx].state = TRANSFER_ABORTED; + break; + } + + char *copy = malloc(resp.size); + if (copy == NULL) { + tfs->transfers[transfer_idx].state = TRANSFER_ABORTED; + break; + } + memcpy(copy, data, resp.size); + + tfs->transfers[transfer_idx].state = TRANSFER_COMPLETED; + tfs->transfers[transfer_idx].data = copy; + tfs->transfers[transfer_idx].size = resp.size; + mark_waiting_transfers_for_hash_as_aborted(tfs, resp.hash); + + if (begin_transfers(tfs) == 0) { + tfs->step = TOASTYFS_ERROR_VOID; + tfs->step = STEP_GET_DONE; + } else { + tfs->step = STEP_FETCH_CHUNK; + } + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_GET_DONE; + } } - return 0; + break; + case STEP_STORE_CHUNK: + { + if (type == MESSAGE_TYPE_STORE_CHUNK_ACK) { + + StoreChunkAckMessage ack; + if (msg.len != sizeof(StoreChunkAckMessage)) + return -1; + memcpy(&ack, msg.ptr, sizeof(ack)); + + if (ack.success) { + + int transfer_idx = find_started_transfer_by_hash(tfs, ack.hash); + assert(transfer_idx > -1); + + // TODO: Mark all waiting transfers for the same hash as ABORTED + tfs->transfers[transfer_idx].state = TRANSFER_COMPLETED; + + if (begin_transfers(tfs) == 0) { + + CommitPutMessage msg = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_COMMIT_PUT, + .length = sizeof(CommitPutMessage), + }, + .oper = { + .type = META_OPER_PUT, + .size = tfs->blob_size, + .content_hash = tfs->content_hash, + .num_chunks = tfs->num_chunks, + }, + .client_id = tfs->client_id, + .request_id = tfs->request_id, + }; + memcpy(msg.oper.bucket, tfs->bucket, META_BUCKET_MAX); + memcpy(msg.oper.key, tfs->key, META_KEY_MAX); + for (int i = 0; i < tfs->num_chunks; i++) { + msg.oper.chunks[i].hash = tfs->chunks[i].hash; + msg.oper.chunks[i].size = tfs->chunks[i].size; + } + send_message_to_server(xxx); + tfs->step = STEP_COMMIT; + + } else { + tfs->step = STEP_STORE_CHUNK; + } + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_PUT_DONE; + } + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_PUT_DONE; + } + } + break; + case STEP_COMMIT: + { + if (type == MESSAGE_TYPE_REDIRECT) { + + RedirectMessage redirect; + if (msg.len != sizeof(RedirectMessage)) + return -1; + memcpy(&redirect, msg.ptr, sizeof(redirect)); + + if (redirect.view_number > tfs->view_number) { + tfs->view_number = redirect.view_number; + replay_request(tfs); + } + + } else if (type == MESSAGE_TYPE_REPLY) { + + ReplyMessage reply; + if (msg.len != sizeof(ReplyMessage)) + return -1; + memcpy(&reply, msg.ptr, sizeof(reply)); + + if (reply.request_id != tfs->request_id) + return 0; + + if (reply.rejected) { + // Operation rejected at the VSR layer + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_PUT_DONE; + break; + } + + if (reply.result.type == META_RESULT_FULL) { + // Storage is full + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_PUT_DONE; + break; + } + + assert(reply.result.type == META_RESULT_OK); + tfs->error = TOASTYFS_ERROR_VOID; + tfs->step = STEP_PUT_DONE; + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_PUT_DONE; + } + } + break; + case STEP_DELETE: + { + if (type == MESSAGE_TYPE_REDIRECT) { + + RedirectMessage redirect; + if (msg.len != sizeof(RedirectMessage)) + return -1; + memcpy(&redirect, msg.ptr, sizeof(redirect)); + + if (redirect.view_number > tfs->view_number) { + tfs->view_number = redirect.view_number; + replay_request(tfs); + } + + } else if (type == MESSAGE_TYPE_REPLY) { + + ReplyMessage reply; + if (msg.len != sizeof(ReplyMessage)) + return -1; + memcpy(&reply, msg.ptr, sizeof(reply)); + + if (reply.request_id != tfs->request_id) + break; + + if (reply.rejected) { + // Operation rejected at the VSR layer + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_DELETE_DONE; + break; + } + + if (reply.result.type == META_RESULT_FULL) { + // Storage is full + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_DELETE_DONE; + break; + } + + assert(reply.result.type == META_RESULT_OK); + tfs->error = TOASTYFS_ERROR_VOID; + tfs->step = STEP_DELETE_DONE; + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_DELETE_DONE; + } + } + break; + case STEP_GET: + { + if (type == MESSAGE_TYPE_REDIRECT) { + + RedirectMessage redirect; + if (msg.len != sizeof(RedirectMessage)) + return -1; + memcpy(&redirect, msg.ptr, sizeof(redirect)); + + if (redirect.view_number > tfs->view_number) { + tfs->view_number = redirect.view_number; + replay_request(tfs); + } + + } else if (type == MESSAGE_TYPE_GET_BLOB_RESPONSE) { + + GetBlobResponseMessage resp; + if (msg.len != sizeof(GetBlobResponseMessage)) + return -1; + memcpy(&resp, msg.ptr, sizeof(resp)); + + if (resp.found) { + + for (int i = 0; i < resp.num_chunks; i++) { + for (int j = 0; j < REPLICATION_FACTOR; j++) { + add_transfer(tfs, resp.chunks[i], resp.locations[i][j], NULL, 0); + } + tfs->chunks[i] = resp.chunks[i]; + } + tfs->num_chunks = resp.num_chunks; + tfs->file_size = resp.size; + + if (begin_transfers(tfs) == 0) { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_GET_DONE; + } else { + tfs->step = STEP_FETCH_CHUNK; + } + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_GET_DONE; + } + + } else { + tfs->error = TOASTYFS_ERROR_XXX; + tfs->step = STEP_GET_DONE; + } + } + break; + default: + UNREACHABLE; } - if (!state->pending) - return -1; - - if (type != MESSAGE_TYPE_REPLY) - return -1; - - ReplyMessage message; - if (msg.len != sizeof(ReplyMessage)) - return -1; - memcpy(&message, msg.ptr, sizeof(message)); - - // Ignore stale replies from previous requests. After a timeout - // the client moves to a new view and sends a new request, but - // the old leader may still deliver a reply for the old request - // on the previous connection. Without this check the client - // would accept the stale result for the wrong operation. - if (message.request_id != state->request_id) - return 0; - - { - Time now = get_current_time(); - char oper_buf[128]; - meta_snprint_oper(oper_buf, sizeof(oper_buf), &state->last_oper); - if (message.rejected) { - client_log(state, now, "RECV REPLY", "%s -> REJECTED", oper_buf); - } else { - char result_buf[64]; - meta_snprint_result(result_buf, sizeof(result_buf), message.result); - client_log(state, now, "RECV REPLY", "%s -> %s", oper_buf, result_buf); - } - } - - state->last_result = message.result; - state->last_was_timeout = false; - state->last_was_rejected = message.rejected; - state->pending = false; return 0; } -int client_init(void *state_, int argc, char **argv, - void **ctxs, struct pollfd *pdata, int pcap, int *pnum, - int *timeout) +void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pnum) { - ClientState *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. Usage is --server :\n"); - return -1; - } - if (state->num_servers == NODE_LIMIT) { - fprintf(stderr, "Node limit of %d reached\n", NODE_LIMIT); - return -1; - } - // TODO: Check address is not duplicated - if (parse_addr_arg(argv[i], &state->server_addrs[state->num_servers++]) < 0) { - fprintf(stderr, "Malformed : pair for --server option\n"); - return -1; - } - } else { - printf("Ignoring option '%s'\n", argv[i]); - } - } - - // Now sort the addresses - addr_sort(state->server_addrs, state->num_servers); - - if (tcp_context_init(&state->tcp) < 0) { - fprintf(stderr, "Client :: Couldn't setup TCP context\n"); - return -1; - } - - state->pending = false; - - state->view_number = 0; - state->request_id = 0; - state->reconnect_time = 0; - - state->client_id = next_client_id++; - - // Connect to all known servers - for (int i = 0; i < state->num_servers; i++) { - if (tcp_connect(&state->tcp, state->server_addrs[i], i, NULL) < 0) { - fprintf(stderr, "Client :: Couldn't connect to server %d\n", i); - tcp_context_free(&state->tcp); - return -1; - } - } - - { - Time now = get_current_time(); - client_log(state, now, "INIT", "servers=%d leader=%d", state->num_servers, leader_idx(state)); - } - - *timeout = 0; - if (pcap < TCP_POLL_CAPACITY) { - fprintf(stderr, "Client :: Not enough poll() capacity (got %d, needed %d)\n", pcap, TCP_POLL_CAPACITY); - return -1; - } - *pnum = tcp_register_events(&state->tcp, ctxs, pdata); - return 0; -} - -int client_tick(void *state_, void **ctxs, - struct pollfd *pdata, int pcap, int *pnum, int *timeout) -{ - ClientState *state = state_; - Event events[TCP_EVENT_CAPACITY]; - int num_events = tcp_translate_events(&state->tcp, events, ctxs, pdata, *pnum); + int num_events = tcp_translate_events(&tfs->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; - int tag = tcp_get_tag(&state->tcp, conn_idx); - if (tag == leader_idx(state) && state->pending) { - Time now = get_current_time(); - client_log(state, now, "DISCONNECT", "%s/%s lost leader (node %d)", - state->last_oper.bucket, state->last_oper.key, leader_idx(state)); - state->last_was_timeout = true; - state->last_was_rejected = false; - state->pending = false; - } - tcp_close(&state->tcp, conn_idx); + tcp_close(&tfs->tcp, conn_idx); continue; } @@ -242,118 +446,297 @@ int client_tick(void *state_, void **ctxs, 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); + int ret = tcp_next_message(&tfs->tcp, conn_idx, &msg, &msg_type); if (ret == 0) break; if (ret < 0) { - tcp_close(&state->tcp, conn_idx); + tcp_close(&tfs->tcp, conn_idx); break; } ret = process_message(state, conn_idx, msg_type, msg); if (ret < 0) { - tcp_close(&state->tcp, conn_idx); + tcp_close(&tfs->tcp, conn_idx); break; } - tcp_consume_message(&state->tcp, conn_idx); + tcp_consume_message(&tfs->tcp, conn_idx); } } - Time now = get_current_time(); + return 0; +} - // If we've been waiting too long for a response, give up and - // try the next server (the current leader may have crashed and - // a view change may have happened) - if (state->pending) { - Time request_deadline = state->request_time + PRIMARY_DEATH_TIMEOUT_SEC * 1000000000ULL; - if (request_deadline <= now) { - - { - char oper_buf[128]; - meta_snprint_oper(oper_buf, sizeof(oper_buf), &state->last_oper); - client_log(state, now, "TIMEOUT", "%s", oper_buf); - } - - state->view_number++; - state->last_was_timeout = true; - state->last_was_rejected = false; - state->pending = false; - } - } - - if (!state->pending) { - - int conn_idx = tcp_index_from_tag(&state->tcp, leader_idx(state)); - if (conn_idx < 0) { - if (state->reconnect_time <= now) { - tcp_connect(&state->tcp, state->server_addrs[leader_idx(state)], leader_idx(state), NULL); - state->reconnect_time = now + HEARTBEAT_INTERVAL_SEC * 1000000000ULL; - } - } else { - - // Now start a new operation - state->request_id++; - state->last_oper = random_oper(); - - RequestMessage request_message = { - .base = { - .version = MESSAGE_VERSION, - .type = MESSAGE_TYPE_REQUEST, - .length = sizeof(RequestMessage), - }, - .oper = state->last_oper, - .client_id = state->client_id, - .request_id = state->request_id, - }; - - ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); - assert(output); - - byte_queue_write(output, &request_message, request_message.base.length); - - { - char oper_buf[128]; - meta_snprint_oper(oper_buf, sizeof(oper_buf), &state->last_oper); - client_log(state, now, "SEND REQUEST", "%s", oper_buf); - } - - state->pending = true; - state->request_time = now; - } - } - - // Set timeout based on pending request deadline or reconnection delay +int toastyfs_register_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pcap) +{ Time deadline = INVALID_TIME; - if (state->pending) { - nearest_deadline(&deadline, state->request_time + PRIMARY_DEATH_TIMEOUT_SEC * 1000000000ULL); - } else { - int conn_idx = tcp_index_from_tag(&state->tcp, leader_idx(state)); - if (conn_idx < 0 && state->reconnect_time > now) { - nearest_deadline(&deadline, state->reconnect_time); - } + + // TODO: Add timeout for the current operation + if (tfs->phase != BLOB_IDLE) { + nearest_deadline(&deadline, tfs->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; + return tcp_register_events(&tfs->tcp, ctxs, pdata); } -int client_free(void *state_) +static void +choose_store_locations_for_chunk(ToastyFS *tfs, int *locations) { - ClientState *state = state_; + // TODO: Pick REPLICATION_FACTOR servers and store their + // indices in "locations" +} - { - Time now = get_current_time(); - client_log_simple(state, now, "CRASHED"); +int toastyfs_async_put(ToastyFS *tfs, char *key, int key_len, + char *data, int data_len) +{ + if (tfs->step != STEP_IDLE) + return -1; + + for (int i = 0; i < num_chunks; i++) { + + SHA256 hash = xxx; + + int locations[REPLICATION_FACTOR]; + choose_store_locations_for_chunk(tfs, locations); + + for (int j = 0; j < REPLICATION_FACTOR; j++) + add_transfer(tfs, hash, locations[j]); + + tfs->chunks[i] = hash; } - tcp_context_free(&state->tcp); + tfs->step = STEP_STORE_CHUNK; + + if (begin_transfers(tfs) == 0) { + // Eatly completion + tfs->step = STEP_PUT_DONE; + } return 0; } + +int toastyfs_async_get(ToastyFS *tfs, char *key, int key_len) +{ + if (tfs->step != STEP_IDLE) + return -1; + + GetBlobMessage msg = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_GET_BLOB, + .length = sizeof(GetBlobMessage), + }, + }; + memcpy(msg.bucket, tfs->bucket, META_BUCKET_MAX); + memcpy(msg.key, tfs->key, META_KEY_MAX); + + send_message_to_server(tfs, xxx, &msg); + return 0; +} + +int toastyfs_async_delete(ToastyFS *tfs, char *key, int key_len) +{ + if (tfs->step != STEP_IDLE) + return -1; + + ReplyMessage msg = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_GET_BLOB, + .length = sizeof(ReplyMessage), + }, + .oper = { + .type = META_OPER_DELETE, + } + }; + memcpy(msg.oper.bucket, tfs->bucket, META_BUCKET_MAX); + memcpy(msg.oper.key, tfs->key, META_KEY_MAX); + + send_message_to_server(tfs, leader_idx(tfs), &msg); +} + +static int +find_completed_transfer_for_hash(ToastyFS *tfs, SHA256 hash) +{ + for (int i = 0; i < tfs->num_transfers; i++) { + if (!memcmp(&hash, &tfs->transfer[i].hash, sizeof(SHA256)) + && tfs->transfer[i].state == TRANSFER_COMPLETED) + return i; + } + return -1; +} + +static void get_result(ToastyFS *tfs, ToastyFS_Result *result) +{ + assert(tfs->step == STEP_GET_DONE); + tfs->step = STEP_IDLE; + + if (tfs->error != TOASTYFS_ERROR_VOID) { + result->type = TOASTYFS_RESULT_GET; + result->error = tfs->error; + result->data = NULL; + result->size = 0; + return; + } + + int blob_size = tfs->file_size; + char *blob_data = malloc(tfs->file_size); + if (blob_data == NULL) { + result->type = TOASTYFS_RESULT_GET; + result->error = TOASTYFS_ERROR_XXX; + result->data = NULL; + result->size = 0; + return; + } + + int chunk_size = xxx; + for (int i = 0; i < tfs->num_chunks; i++) { + + SHA256 hash = tfs->chunks[i]; + + int j = find_completed_transfer_for_hash(tfs, hash); + if (j < 0) { + result->type = TOASTYFS_RESULT_GET; + result->error = TOASTYFS_ERROR_XXX; + result->data = NULL; + result->size = 0; + return; + } + + char *data = tfs->transfer[j].data; + int size = tfs->transfer[j].size; + + if (size > blob_size - offset) + size = blob_size - offset; + + memcpy(blob + offset, data, size); + + offset += chunk_size; + } + + result->type = TOASTYFS_RESULT_GET; + result->error = TOASTYFS_ERROR_VOID; + result->data = blob_data; + result->size = blob_size; +} + +static bool +all_chunk_transfers_completed(ToastyFS *tfs) +{ + for (int i = 0; i < tfs->num_chunks; i++) { + int j = find_completed_transfer_for_hash(tfs, tfs->chunks[i]); + if (j < 0) + return false; + } + return true; +} + +static int put_result(ToastyFS *tfs, ToastyFS_Result *result) +{ + assert(tfs->step == STEP_PUT_DONE); + tfs->step = STEP_IDLE; + + if (tfs->error != TOASTYFS_ERROR_VOID) { + result->type = TOASTYFS_RESULT_PUT; + result->error = tfs->error; + result->data = NULL; + result->size = 0; + return; + } + + if (!all_chunk_transfers_completed(tfs)) { + result->type = TOASTYFS_RESULT_PUT; + result->error = TOASTYFS_ERROR_XXX; + result->data = NULL; + result->size = 0; + return; + } + + result->type = TOASTYFS_RESULT_PUT; + result->error = TOASTYFS_ERROR_VOID; + result->data = NULL; + result->size = 0; +} + +static int delete_result(ToastyFS *tfs, ToastyFS_Result *result) +{ + assert(tfs->step == STEP_DELETE_DONE); + tfs->step = STEP_IDLE; + + if (tfs->error != TOASTYFS_ERROR_VOID) { + result->type = TOASTYFS_RESULT_DELETE; + result->error = tfs->error; + result->data = NULL; + result->size = 0; + return; + } + + result->type = TOASTYFS_RESULT_DELETE; + result->data = NULL; + result->size = 0; +} + +ToastyFS_Result toastyfs_get_result(ToastyFS *tfs) +{ + ToastyFS_Result result; + if (tfs->step == STEP_GET_DONE) { + get_get_result(tfs, result); + } else if (tfs->step == STEP_PUT_DONE) { + get_put_result(tfs, result); + } else if (tfs->step == STEP_DELETE_DONE) { + get_delete_result(rfs, result); + } else { + result.type = TOASTYFS_RESULT_VOID; + result.error = TOASTYFS_ERROR_VOID; + result.data = NULL; + result.size = 0; + } + return result; +} + +static int wait_until_result(ToastyFS *tfs, ToastyFS_Result *res) +{ + do { + int timeout; + struct pollfd arr[xxx]; + int num = toastyfs_register_events(tfs, arr, xxx, &timeout); + if (num < 0) + return num; + + POLL(arr, num, timeout); + + toastyfs_process_events(tfs); + ret = toastyfs_get_result(tfs, res); + } while (ret == 0); + return ret; +} + +int toastyfs_put(ToastyFS *tfs, char *key, int key_len, + char *data, int data_len, ToastyFS_Result *res) +{ + int ret = toastyfs_async_put(tfs, key, key_len, data, data_len); + if (ret < 0) + return ret; + return wait_until_result(tfs, res); +} + +int toastyfs_get(ToastyFS *tfs, char *key, int key_len) +{ + int ret = toastyfs_async_put(tfs, key, key_len); + if (ret < 0) + return ret; + return wait_until_result(tfs, res); +} + +int toastyfs_delete(ToastyFS *tfs, char *key, int key_len) +{ + int ret = toastyfs_async_delete(tfs, key, key_len); + if (ret < 0) + return ret; + return wait_until_result(tfs, res); +} diff --git a/src/client.h b/src/client.h deleted file mode 100644 index bb79428..0000000 --- a/src/client.h +++ /dev/null @@ -1,49 +0,0 @@ -#ifndef CLIENT_INCLUDED -#define CLIENT_INCLUDED - -#include -#include - -#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 \ No newline at end of file diff --git a/lib/file_system.c b/src/file_system.c similarity index 100% rename from lib/file_system.c rename to src/file_system.c diff --git a/lib/file_system.h b/src/file_system.h similarity index 100% rename from lib/file_system.h rename to src/file_system.h diff --git a/lib/message.c b/src/message.c similarity index 100% rename from lib/message.c rename to src/message.c diff --git a/lib/message.h b/src/message.h similarity index 100% rename from lib/message.h rename to src/message.h diff --git a/src/metadata.h b/src/metadata.h index 4f37b99..5a045b4 100644 --- a/src/metadata.h +++ b/src/metadata.h @@ -4,7 +4,7 @@ #include #include -#include +#include "basic.h" typedef struct { SHA256 hash; diff --git a/src/random_client.c b/src/random_client.c new file mode 100644 index 0000000..adb2b04 --- /dev/null +++ b/src/random_client.c @@ -0,0 +1,102 @@ +#if defined(MAIN_SIMULATION) || defined(MAIN_TEST) +#define QUAKEY_ENABLE_MOCKS +#endif + +#include +#include +#include + +#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; +} diff --git a/src/random_client.h b/src/random_client.h new file mode 100644 index 0000000..94bb1f4 --- /dev/null +++ b/src/random_client.h @@ -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 diff --git a/src/server.h b/src/server.h index 741ec79..5b1c7bc 100644 --- a/src/server.h +++ b/src/server.h @@ -1,10 +1,9 @@ #ifndef NODE_INCLUDED #define NODE_INCLUDED -#include -#include -#include - +#include "tcp.h" +#include "basic.h" +#include "message.h" #include "log.h" #include "config.h" #include "metadata.h" diff --git a/lib/tcp.c b/src/tcp.c similarity index 100% rename from lib/tcp.c rename to src/tcp.c diff --git a/lib/tcp.h b/src/tcp.h similarity index 100% rename from lib/tcp.h rename to src/tcp.h