Rewrite of the ToastyFS client

README.md

@@ -34,13 +34,15 @@ I initially started this project to learn about distributed systems. I asked mys
 This project should be considered a robust proof-of-concept at this time. Features that would be required for long-running instances are missing, such as:
 
 * Log compaction
-* Ability to shut down all nodes without resetting the system
+* Log persistence on disk
+
+If a majority of nodes is turned off, the system's state will be lost. This is in accordance with disk-free version of the Viewstamped Replication protocol.
 
 ## Getting Started
 
 ### Building
 
-ToastyFS supports Windows and Linux. Each platform has its own build script:
+ToastyFS supports Windows and Linux. It can be compiled by calling the `build.bat` script on Windows and `build.sh` on Linux
 
 ```
 # Windows
@@ -52,14 +54,12 @@ ToastyFS supports Windows and Linux. Each platform has its own build script:
 
 The build script will produce the following executables:
 
-| Name | Description |
+toasty_simulation: Runs a ToastyFS cluster in-memory, with sped-up time and serving a number of random operations. See [Testing](#testing) section for details.
-| :--- | :---------- |
+toasty: The actual ToastyFS program. This is what you need to run to use ToastyFS.
-| toasty_simulation | Runs a ToastyFS cluster in-memory, with sped-up time and serving a number of random operations. See [Testing](#testing) section for details. |
+toasty_random_client: An utility client which spams random requests towards a ToastyFS cluster. Useful for testing.
-| toasty | The actual ToastyFS program. This is what you need to run to use ToastyFS. |
+toasty_proxy: An HTTP proxy that translates HTTP request to the ToastyFS-specific request protocol.
-| toasty_random_client | An utility client which spams random requests towards a ToastyFS cluster. Useful for testing. |
-| toasty_proxy | An HTTP proxy that translates HTTP request to the ToastyFS-specific request protocol. |
 
-On Windows the executables will have the `.exe` extension.
+If you are on Windows, the executable names will have the .exe extension (so you will get `toasty_simulation.exe` instead of `toasty_simulation` for instance).
 
 ### Running a Cluster
 
@@ -92,13 +92,15 @@ I'm the first object
 
 ### The Need for Replication
 
-To build a fault-tolerant service, it's necessary for it to run on multiple machines in such a way that if a machine dies, the others can continue its work. This is referred to as replication. Designing and implementing a replication system that properly manages all edge cases is **incredibly hard**. Such an algorithm needs to account for any number of node crashes at any phase of the protocol, it needs to account for arbitrary network partitions (network failures that cause the system to be split in multiple groups of nodes) making it impossible for the state of each group to diverge. This is such a hard problem that not only few general algorithms have been designed, but also the number of their implementations is vanishingly low. The common replication algorithms are Raft and Paxos (*). This project uses a less known algorithm called Viewstamped Replication (VSR).
+In order to build fault tolerant systems, it is necessary to build them as distributed systems: multiple nodes that coordinate to offer a single service. This allows for the system to stay alive in case some nodes fail. The system must be able to detect when some nodes become unavailable and choose other nodes to take their places to ensure the service is not interrupted. In practice, this is achieved via replication algorithms, such as Raft and Viewstamped Replication.
 
-(*) Strictly speaking, Paxos is a consensus algorithm. Replication can be built on top of it.
+Replication algorithms, which are related but not exactly the same as consensus algorithms, allow the creation of a group of nodes (called replicas) that are synchronized maintaining the same state. The replication algorithm ensures that if a node of the group dies, it's impossible for a node that takes its place to have a stale state, causing an inconsistency in the overall system from the perspective of the user.
+
+This is generally considered as a very hard problem, since the value of such algorithms is in the mathematical certainty that once a request is accepted by the system, that information will never be lost. The algorithm needs both be designed and implemented correctly. ToastyFS solves this problem using the Viewstamped Replication algorithm.
 
 ### Raft VS Viewstamped Replication
 
-For historical reasons, Paxos (1989) is the established algorithm to achieve replication. Paxos is notorious for being hard to understand and implement, so in 2014 a more understandable algorithm called Raft was introduced. Since then, Raft has been the go-to algorithm from new open source projects.
+For historical reasons, the established algorithm to achieve replication is Paxos (strictly speaking, it's a consensus algorithm on top of which replication is built). Paxos (1989) is known for being hard to understand, and therefore to implement, so in 2014 a new simpler algorithm called Raft was introduced. Since then it has been the go-to algorithm for new open source projects.
 
 There is also a lesser known replication algorithm called Viewstamped Replication (VSR), which never got much attention from the industry even though it was invented around the same time as Paxos (and arguably, before it). In 2012 the "Viewstamped Replication Revisited" paper was published which offered a modernized design of the protocol. The later paper is the reference for VSR used in this project.
 

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;

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);

src/client_2.c

@@ -0,0 +1,868 @@
+#include <ToastyFS.h>
+
+// 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;
+}

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.