diff --git a/lib/message.c b/lib/message.c index a255234..ced73af 100644 --- a/lib/message.c +++ b/lib/message.c @@ -185,6 +185,13 @@ static int find_channel_by_message(MessageSystem *msys, void *raw_message) return -1; } +int message_type(void *raw_message) +{ + Message message; + memcpy(&message, raw_message, sizeof(message)); + return message.type; +} + int message_length(void *raw_message) { Message message; diff --git a/lib/message.h b/lib/message.h index 2e2533d..821de23 100644 --- a/lib/message.h +++ b/lib/message.h @@ -30,6 +30,7 @@ int message_system_register_events(MessageSystem *msys, void *get_next_message(MessageSystem *msys); +int message_type(void *raw_message); int message_length(void *raw_message); void consume_message(MessageSystem *msys, void *ptr); diff --git a/src/client_2.c b/src/client_2.c new file mode 100644 index 0000000..f505a8a --- /dev/null +++ b/src/client_2.c @@ -0,0 +1,868 @@ +#include + +// How many chunk transfers can the client perform in parallel +#define PARALLEL_TRANSFER_LIMIT 5 + +typedef enum { + + // No operation in progress. New ones can be started. + CLIENT_IDLE, + + // The starting state of all PUT operations where chunks + // are uploaded to servers before committing any metadata. + CLIENT_UPLOADING_CHUNKS, + + // This is the state when all chunks of an object have + // been uploaded and the client is waiting for the server + // to accept the new object metadata. + CLIENT_UPLOADING_METADATA, + + // PUT operation failed. The error field of the ToastyFS + // structure is set accordingly. + CLIENT_FAILED_PUT, + + // PUT operation succeded + CLIENT_COMPLETED_PUT, + + CLIENT_DOWNLOADING_METADATA, + + CLIENT_DOWNLOADING_CHUNKS, + + CLIENT_FAILED_GET, + + CLIENT_COMPLETED_GET, + + CLIENT_DELETING_METADATA, + + CLIENT_FAILED_DELETE, + + CLIENT_COMPLETED_DELETE, + +} ClientState; + +typedef enum { + + // The transfer ready but wasn't started yet + TRANSFER_PENDING, + + // The transfer started + TRANSFER_STARTED, + + // The trasfer was stopped on our end + TRANSFER_ABORTED, + + // The transfer failed + TRANSFER_FAILED, + + // The transfer is complete + TRANSFER_COMPLETE, +} TransferState; + +// This structure represents the state of a single chunk's transfer. +typedef struct { + TransferState state; + + // Index of the first chunk with this hash + int chunk; + + // Index of the target server + int server; +} Transfer; + +// This structure holds metadata associated to a chunk that is being +// uploaded of downloaded. Note that generally speaking multiple +// trasfers may refer to a single chunk +typedef struct { + char *ptr; + int len; + SHA256 hash; +} Chunk; + +struct ToastyFS { + + // Lower level networking system for message-passing + MessageSystem *msys; + + ClientState state; + + // Error code that will be returned when the operation completion + // information is requested. + ToastyFS_Error error; + + // Metadata associated to the chunks of an object upload + // or download. + Chunk *chunks; + int num_chunks; + + Transfer *transfers; + int num_transfers; + bool is_upload; // This flag determines whether transfers are uploads or downloads +}; + +// This function initializes the ToastyFS client instance. +// +// The client_id is an arbitrary integer that identifies this client +// uniquely. Each client interacting with the ToastyFS cluster must +// have a different one. A client that crashes and restarts may and +// should reuse its old client_id. +// +// The addrs array argument contains the IPv4 addresses of the cluster +// servers, while num_addrs is the number of servers. Addresses are +// expressed in dotted-decimal notation. +ToastyFS *toastyfs_init(uint64_t client_id, char **addrs, int num_addrs) +{ + ToastyFS *tfs = malloc(sizeof(ToastyFS)); + if (tfs == NULL) + return NULL; + + tfs->msys = message_system_init(addrs, num_addrs); + if (tfs->msys == NULL) { + free(tfs); + return NULL; + } + + // TODO + + return tfs; +} + +// Release resources associated to a ToastyFS client. +void toastyfs_free(ToastyFS *tfs) +{ + message_system_free(tfs->msys); + free(tfs); +} + +static bool +is_result_for_transfer(Transfer *transfer, int server, SHA256 hash) +{ + if (transfer->state != TRANSFER_STARTED) + return false; + + if (transfer->server != server) + return false; + + if (memcmp(&transfer->hash, &hash, sizeof(SHA256))) + return false; + + return true; +} + +static void set_upload_result(ToastyFS *tfs, int server, SHA256 hash, bool success) +{ + for (int i = 0; i < tfs->num_transfers; i++) { + if (is_result_for_transfer(&tfs->transfers[i], server, hash)) { + if (success) { + tfs->transfers[i].state = TRANSFER_COMPLETE; + } else { + tfs->transfers[i].state = TRANSFER_ABORTED; + } + } + } + + // If this upload was successful and the chunk was now written + // to a majority of servers, abort any pending uploads of the + // same hash. + + int num_complete = 0; + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_COMPLETE && !memcmp(tfs->transfers[i].hash, &hash)) + num_complete++; + } + + if (num_complete > tfs->num_servers/2) { + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_PENDING && !memcmp(tfs->transfers[i].hash, &hash)) + tfs->transfers[i].state = TRANSFER_ABORTED; + } + } +} + +static bool +have_pending_or_started_transfers(ToastyFS *tfs) +{ + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_PENDING || + tfs->transfers[i].state == TRANSFER_STARTED) + return true; + } + return false; +} + +static int successful_uploads_for_chunk(ToastyFS *tfs, SHA256 hash) +{ + int count = 0; + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_COMPLETE + && !memcmp(&tfs->transfers[i].hash, &hash, sizeof(SHA256))) { + count++; + } + } + return count; +} + +static bool all_chunks_replicated(ToastyFS *tfs) +{ + for (int i = 0; i < tfs->num_chunks; i++) { + if (successful_uploads_for_chunk(tfs, tfs->chunks[i].hash) < CEIL(tfs->num_servers, 2)) + return false; + } + return true; +} + +static void process_uploading_chunks(ToastyFS *tfs, void *ptr) +{ + // We are expecting an upload success or failure. + // Ignore anything else. + if (message_type(ptr) != MESSAGE_TYPE_STORE_CHUNK_RESPONSE) + return; // Ignore + + StoreChunkResponseMessage message; + memcpy(&message, ptr, sizeof(message)); + + set_upload_result(tfs, message.base.sender_idx, message.hash, message.success); + start_uploads(tfs); + + if (!have_pending_or_started_transfers(tfs)) { + // If we managed to replicate each chunk on a majority + // of servers, we can commit the operation by sending + // the object's metadata, else we fail. + if (all_chunks_replicated(tfs)) { + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_COMMIT_PUT, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = get_next_request_id(tfs), + }; + send_message(tfs->msys, &message.base); + tfs->state = CLIENT_UPLOADING_METADATA; + } else { + tfs->state = CLIENT_FAILED_PUT; + tfs->error = xxx; + } + } +} + +static void process_uploading_metadata(ToastyFS *tfs, void *ptr) +{ + if (message_type(ptr) == MESSAGE_TYPE_REDIRECT) { + + // Replay request + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_COMMIT_PUT, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = tfs->last_request_id, + }; + send_message(tfs->msys, tfs->primary, &message.base); + + } else if (message_type(ptr) == MESSAGE_TYPE_REPLY) { + + VsrReplyMessage message; + memcpy(&message, ptr, sizeof(message)); + + if (message.request_id != tfs->last_request_id) + return; // Ignore + + if (message.rejected) { + tfs->state = CLIENT_FAILED_PUT; + tfs->error = xxx; + return; + } + + switch (message.meta.type) { + case META_RESULT_OK: + tfs->state = CLIENT_COMPLETED_PUT; + assert(tfs->state == TOASTYFS_ERROR_VOID); + break; + case META_RESULT_NOT_FOUND: + tfs->state = CLIENT_FAILED_PUT; + tfs->error = TOASTYFS_ERROR_NOT_FOUND; + break; + case META_RESULT_FULL: + tfs->state = CLIENT_ERROR; + tfs->error = TOASTYFS_ERROR_FULL; + break; + } + } +} + +// This function starts transfer operations up to the parallel +// transfer limit. +// +// Note that this function may start uploads of a single chunk +// to more servers than strictly necessary. Operations to all +// servers are scheduled per chunk in case some servers are +// not available. When a chunk is uploaded to enough servers, +// the transfers to the remaining ones are aborted. But if +// more transfers than necessary are started in parallel, it +// is possible for the chunk to become over-replicated. The +// only downside of this is unnecessary usage of network +// bandwidth. This behavior can be solved later as it does not +// impact the overall architecture of the system. +static void start_transfers(ToastyFS *tfs) +{ + // Count how many uploads are started and how many are pending + int num_started = 0; + int num_pending = 0; + for (int i = 0; i < tfs->num_transfers; i++) { + switch (tfs->transfers[i].state) { + case TRANSFER_STARTED: num_started++; break; + case TRANSFER_PENDING: num_pending++; break; + } + } + + // Start operations while some are pending and we didn't reach the limit + while (num_started < PARALLEL_TRANSFER_LIMIT && num_pending > 0) { + + // Find the next pending operation + int found = -1; + for (int i = 0; i < tfs->num_transfers; i++) { + if (tfs->transfers[i].state == TRANSFER_PENDING) { + found = i; + break; + } + } + assert(found > -1); + + int chunk = tfs->transfers[found].chunk; + + if (tfs->is_upload) { + StoreChunkMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_STORE_CHUNK, + .sender = xxx, + .length = sizeof(StoreChunkMessage) + tfs->chunks[chunk].len; + }, + .hash = xxx, + .size = tfs->chunks[chunk].len, + }; + send_message_ex(tfs->msys, tfs->transfers[found].server, + &message.base, tfs->chunks[chunk].ptr, tfs->chunks[chunk].len); + } else { + FetchChunkMessage message = { + .bse = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_FETCH_CHUNK, + .sender = xxx, + .length = sizeof(FetchChunkMessage), + } + .hash = xxx, + .sender_idx = -1, // TODO: this is unnecessary + }; + send_message(tfs->msys, tfs->transfers[found].server, &message.base); + } + + tfs->transfers[found].state = TRANSFER_STARTED; + num_started++; + num_pending++; + } +} + +static void process_downloading_metadata(ToastyFS *tfs, void *ptr) +{ + if (message_type(ptr) == MESSAGE_TYPE_REDIRECT) { + + // Replay request + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_xxx, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = tfs->last_request_id, + }; + send_message(tfs->msys, tfs->primary, &message.base); + + } else if (message_type(ptr) == MESSAGE_TYPE_REPLY) { + + VsrReplyMessage message; + memcpy(&message, ptr, sizeof(message)); + + if (message.request_id != tfs->last_request_id) + return; // Ignore + + if (message.rejected) { + tfs->state = CLIENT_FAILED_GET; + tfs->error = xxx; + return; + } + + if (message.meta.type != META_RESULT_OK) { + switch (message.meta.type) { + case META_RESULT_NOT_FOUND: + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_NOT_FOUND; + break; + case META_RESULT_FULL: + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_FULL; + break; + } + return; + } + + if (message.num_chunks == 0) { + // Early completion + assert(tfs->error == TOASTYFS_ERROR_VOID); + tfs->state = CLIENT_COMPLETED_GET; + return; + } + + tfs->chunks = malloc(message.num_chunks * sizeof(Chunk)); + if (tfs->chunks == NULL) { + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_OUT_OF_MEMORY; + return; + } + tfs->num_chunks = message.num_chunks; + + int majority = (tfs->num_servers + 1) / 2; + int max_transfers = majority * message.num_chunks; + assert(max_transers > 0); + + tfs->is_upload = false; + tfs->transfers = malloc(max_transers * sizeof(Transfer)); + if (tfs->transfers == NULL) { + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_OUT_OF_MEMORY; + return; + } + tfs->num_transfers = 0; // To be decided + + tfs->output_size = message.size; + tfs->output_data = malloc(message.size); + if (tfs->output_data == NULL) { + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_OUT_OF_MEMORY; + return; + } + + for (int i = 0; i < message.num_chunks; i++) { + + tfs->chunks[i].ptr = tfs->output_data + i * CHUNK_SIZE; + tfs->chunks[i].len = MIN(CHUNK_SIZE, tfs->output_size - i * CHUNK_SIZE); + tfs->chunks[i].hash = message.chunks[i].hash; + + // Schedule transfers if no transfers were scheduled for + // this hash yet. + bool duplicate = false; + for (int j = 0; j < i; j++) { + if (!memcmp(&tfs->chunks[j].hash, &tfs->chunks[i].hash)) { + duplicate = true; + break; + } + } + + if (!duplicate) { + for (int j = 0; j < message.chunks[i].num_servers; j++) { + tfs->transfers[tfs->num_transfers].state = TRANSFER_PENDING; + tfs->transfers[tfs->num_transfers].chunk = i; + tfs->transfers[tfs->num_transfers].server = message.chunks[i].servers[j]; + tfs->num_transfers++; + } + } + } + assert(num_transfers > 0); + + start_transfers(tfs); + tfs->state = CLIENT_DOWNLOADING_CHUNKS; + } +} + +static bool chunk_downloaded(ToastyFS *tfs, int chunk) +{ + for (int j = 0; j < tfs->num_transfers; j++) { + if (tfs->transfers[j].state == TRANSFER_COMPLETE && + !memcmp(&tfs->transfers[j].hash, &tfs->chunks[i])) + return true; + } + return false; +} + +static bool all_chunks_retrieved(ToastyFS *tfs) +{ + for (int i = 0; i < tfs->num_chunks; i++) { + if (!chunk_downloaded(tfs, i)) + return false; + } + return true; +} + +static void process_downloading_chunks(ToastyFS *tfs, void *ptr) +{ + if (message_type(ptr) != MESSAGE_TYPE_FETCH_CHUNK_RESPONSE) + return; // Ignore + + FetchChunkResponseMessage message; + memcpy(&message, ptr, sizeof(message)); + + if (message.size == 0) { + tfs->state = CLIENT_FAILED_GET; + tfs->error = TOASTYFS_ERROR_NOT_FOUND; + return; + } + + char* chunk_data = (char*) ptr + sizeof(FetchChunkResponseMessage); + uint32_t chunk_size = message.size; + + for (int i = 0; i < tfs->num_chunks; i++) { + if (!memcmp(&tfs->chunks[i].hash, &message.hash)) { + assert(chunk_size == tfs->chunks[i].len); + memcpy(tfs->chunks[i].ptr, chunk_data, chunk_size); + } + } + + start_transfers(tfs); + if (!have_pending_or_started_transfers(tfs)) { + if (all_chunks_retrieved(tfs)) { + tfs->state = CLIENT_COMPLETED_GET; + } else { + tfs->state = CLIENT_FAILED_GET; + tfs->error = xxx; + } + } +} + +static void process_deleting_metadata(ToastyFS *tfs, void *ptr) +{ + if (message_type(ptr) == MESSAGE_TYPE_REDIRECT) { + + // Replay request + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_xxx, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = tfs->last_request_id, + }; + send_message(tfs->msys, tfs->primary, &message.base); + + } else if (message_type(ptr) == MESSAGE_TYPE_REPLY) { + + VsrReplyMessage message; + memcpy(&message, ptr, sizeof(message)); + + if (message.request_id != tfs->last_request_id) + return; // Ignore + + if (message.rejected) { + tfs->state = CLIENT_FAILED_DELETE; + tfs->error = xxx; + return; + } + + switch (message.meta.type) { + case META_RESULT_OK: + tfs->state = CLIENT_COMPLETED_DELETE; + assert(tfs->state == TOASTYFS_ERROR_VOID); + break; + case META_RESULT_NOT_FOUND: + tfs->state = CLIENT_FAILED_DELETE; + tfs->error = TOASTYFS_ERROR_NOT_FOUND; + break; + case META_RESULT_FULL: + tfs->state = CLIENT_FAILED_DELETE; + tfs->error = TOASTYFS_ERROR_FULL; + break; + } + } +} + +void toastyfs_process_events(ToastyFS *tfs, + void **ptrs, struct pollfd *pfds, int num) +{ + message_system_process_events(tfs->msys, ptrs, pfds, num); + + for (void *ptr; (ptr = get_next_message(tfs->msys)); ) { + + switch (tfs->state) { + case CLIENT_UPLOADING_CHUNKS: + process_uploading_chunks(tfs, ptr); + break; + case CLIENT_UPLOADING_METADATA: + process_uploading_metadata(tfs, ptr); + break; + case CLIENT_DOWNLOADING_METADATA: + process_downloading_metadata(tfs, ptr); + break; + case CLIENT_DOWNLOADING_CHUNKS: + process_downloading_chuns(tfs, ptr); + break; + case CLIENT_DELETING_METADATA: + process_deleting_metadata(tfs, ptr); + break; + default: + break; // Wasn't expecting a message. Ignore. + } + + consume_message(tfs->msys, ptr); + } +} + +int toastyfs_register_events(ToastyFS *tfs, void **ptrs, + struct pollfd *pfds, int cap, int *timeout) +{ + return message_system_register_events(tfs->msys, ptrs, pfds, cap, timeout); +} + +// Begin an asynchronous object creation operation +// +// Note that there can only be one pending operation at a time. +int toastyfs_async_put(ToastyFS *tfs, char *key, int key_len, + char *data, int data_len) +{ + // Only one operation allowed at a time + if (tfs->state != CLIENT_IDLE) + return -1; // TODO: error code + tfs->error = TOASTYFS_ERROR_VOID; + + // We need to split the data in chunks, then schedule their + // uploads. Each chunk needs to be uploaded to a majority of + // servers. + // + // The way we do this is by creating an array of transfer + // descriptors. Each describing a possible upload we may need + // to make in order to complete the upload. + // + // For instance say we needed to upload a single chunk C0 to + // a cluster with server nodes S1, S2, S3. We would create + // the following transfer descriptors: + // + // C0 ---> S0 + // C0 ---> S1 + // C0 ---> S2 + // + // Note that C0 only needs to be uploaded to a majority of + // servers, so only 2 out of 3. This is the list of all + // possible transfers we may need to happen to complete the + // overall upload. Once the majority of uploads of a chunk + // complete, the remaining ones for that chunk are aborted. + + // Count the number of chunks we need to upload. If the data + // to upload contains a repeated chunk, we only upload that + // chunk once. This means that the number of chunks we need + // to process may be less that the object's length divided + // by the chunk size. + int max_chunks = CEIL(data_len, CHUNK_SIZE); + assert(max_chunks > 0); + + tfs->chunks = malloc(max_chunks * sizeof(Chunk)); + if (tfs->chunks == NULL) { + tfs->error = TOASTYFS_ERROR_OUT_OF_MEMORY; + return; + } + + tfs->num_chunks = 0; + for (int i = 0; i < max_chunks; i++) { + + char *chunk_ptr = data + i * CHUNK_SIZE; + int chunk_len = MIN(CHUNK_SIZE, data_len - i * CHUNK_SIZE); + + SHA256 hash = sha256(chunk_ptr, chunk_len); + + bool duplicate = false; + for (int j = 0; j < i; j++) { + if (!memcmp(&tfs->chunks[j].hash, &tfs->chunks[i].hash, sizeof(SHA256))) { + duplicate = true; + break; + } + } + + if (!duplicate) { + tfs->chunks[tfs->num_chunks].ptr = chunk_ptr; + tfs->chunks[tfs->num_chunks].len = chunk_len; + tfs->chunks[tfs->num_chunks].hash = hash; + tfs->num_chunks++; + } + } + assert(tfs->num_chunks > 0); + + int num_transfers = tfs->num_chunks * tfs->num_servers; + + tfs->transfers = malloc(num_transfers * sizeof(Chunk)); + if (tfs->transfers == NULL) { + tfs->error = TOASTYFS_ERROR_OUT_OF_MEMORY; + free(tfs->chunks); + return; + } + + tfs->is_upload = true; + tfs->num_transfers = 0; + for (int i = 0; i < num_chunks; i++) { + for (int j = 0; j < tfs->num_servers; j++) { + tfs->transfers[tfs->num_transfers].state = TRANSFER_PENDING; + tfs->transfers[tfs->num_transfers].chunk = i; + tfs->transfers[tfs->num_transfers].server = j; + tfs->num_transfers++; + } + } + + start_transfers(tfs); + tfs->state = CLIENT_UPLOADING_CHUNKS; + tfs->pending = true; + return 0; +} + +int toastyfs_async_get(ToastyFS *tfs, char *key, int key_len) +{ + // Only one operation allowed at a time + if (tfs->state != CLIENT_IDLE) + return -1; // TODO: error code + tfs->error = TOASTYFS_ERROR_VOID; + + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_XXX, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = get_next_request_id(tfs), + }; + send_message(tfs->msys, tfs->primary, &message); + + tfs->state = CLIENT_DOWNLOADING_METADATA; + return 0; +} + +int toastyfs_async_delete(ToastyFS *tfs, char *key, int key_len) +{ + // Only one operation allowed at a time + if (tfs->state != CLIENT_IDLE) + return -1; // TODO: error code + tfs->error = TOASTYFS_ERROR_VOID; + + RequestMessage message = { + .base = { + .version = MESSAGE_VERSION, + .type = MESSAGE_TYPE_XXX, + .sender = xxx, + .length = sizeof(RequestMessage), + }, + .oper = xxx, + .client_id = tfs->client_id, + .request_id = get_next_request_id(tfs), + }; + send_message(tfs->msys, tfs->primary, &message); + + tfs->state = CLIENT_DELETING_METADATA; + return 0; +} + +static ToastyFS_Result get_result(ToastyFS *tfs, bool consume) +{ + ToastyFS_Result result; + switch (tfs->state) { + case CLIENT_FAILED_PUT: + assert(tfs->error != TOASTYFS_RESULT_VOID); + result.type = TOASTYFS_RESULT_PUT; + result.error = tfs->error; + break; + case CLIENT_COMPLETED_PUT: + assert(tfs->error == TOASTYFS_RESULT_VOID); + result.type = TOASTYFS_RESULT_PUT; + result.error = TOASTYFS_ERROR_VOID; + break; + case CLIENT_FAILED_GET: + // TODO + break; + case CLIENT_COMPLETED_GET: + // TODO + break; + case CLIENT_FAILED_DELETE: + // TODO + break; + case CLIENT_COMPLETED_DELETE: + // TODO + break; + default: + result.type = TOASTYFS_RESULT_VOID; + result.error = TOASTYFS_ERROR_VOID; + break; + } + + if (consume) { + // Now restore the struct's state to allow new + // operations to start + tfs->state = CLIENT_IDLE; + tfs->error = TOASTYFS_ERROR_VOID; + } + + return result; +} + +ToastyFS_Result toastyfs_get_result(ToastyFS *tfs) +{ + return get_result(tfs, true); +} + +static bool result_available(ToastyFS *tfs) +{ + return get_result(tfs, false).type != TOASTYFS_RESULT_VOID; +} + +static void wait_completion(ToastyFS *tfs, ToastyFS_Result *res) +{ + while (!result_available(tfs)) { + + void *ptrs[xxx]; + struct pollfd pfds[xxx]; + int timeout; + + int num = toastyfs_register_events(tfs, ptrs, pfds, cap, &timeout); + // TODO: can register_events fail? + POLL(pfds, num, timeout); + toastyfs_process_events(tfs, ptrs, pfds, num); + } + + *res = toastyfs_get_result(tfs); +} + +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; + wait_completion(tfs, res); + return 0; +} + +int toastyfs_get(ToastyFS *tfs, char *key, int key_len, ToastyFS_Result *res) +{ + int ret = toastyfs_async_get(tfs, key, key_len); + if (ret < 0) + return ret; + wait_completion(tfs, res); + return 0; +} + +int toastyfs_delete(ToastyFS *tfs, char *key, int key_len, ToastyFS_Result *res) +{ + int ret = toastyfs_async_put(tfs, key, key_len); + if (ret < 0) + return ret; + wait_completion(tfs, res); + return 0; +} \ No newline at end of file diff --git a/src/server.h b/src/server.h index 5fc7a6e..d132adb 100644 --- a/src/server.h +++ b/src/server.h @@ -38,7 +38,7 @@ enum { // Chunk Storage Protocol (bypasses log) MESSAGE_TYPE_STORE_CHUNK, - MESSAGE_TYPE_STORE_CHUNK_ACK, + MESSAGE_TYPE_STORE_CHUNK_RESPONSE, MESSAGE_TYPE_FETCH_CHUNK, MESSAGE_TYPE_FETCH_CHUNK_RESPONSE, @@ -166,7 +166,7 @@ typedef struct { typedef struct { Message base; SHA256 hash; - int sender_idx; // -1 if from a client + int sender_idx; // -1 if from a client (TODO: remove this) } FetchChunkMessage; // FetchChunkResponse: server -> client/server. Chunk data as trailing bytes.