Continued work on the WAL

misc/TODO.txt

@@ -13,3 +13,8 @@
 [ ] Add failt injections to the simulation
 [ ] Add authentication
 [ ] Add release build and allow switching between debug, release, coverage, sanitizer builds while calling make
+[ ] When an operation is committed to the WAL, it may then fail due to not deterministic reasons (out of memory). When the metadata server restarts and replays the log, the operation may succede and cause the system to diverge from the last instance of the process. Is this not a problem? How do we solve it?
+[ ] Add notes on who inspired the testing approach
+[ ] Add notes on benchmarks and that we don't do batching
+[ ] Fix endianess when writing to network and the WAL
+[ ] Add a change counter to FileTree. The counter is initialized to 0 and updated any time the tree is changed. When a new file is created, the current change counter is assigned to it. This makes it possible to verify that the file didn't change in between read and write operations without that clumsy previous hash list and previous chunk size.

src/file_system.c

@@ -57,6 +57,16 @@ void file_close(Handle fd)
 #endif
 }
 
+int file_set_offset(Handle fd, int off)
+{
+    assert(0); // TODO
+}
+
+int file_get_offset(Handle fd, int *off)
+{
+    assert(0); // TODO
+}
+
 int file_lock(Handle fd)
 {
 #ifdef __linux__

src/file_system.h

@@ -28,6 +28,8 @@ typedef struct {
 
 int  file_open(string path, Handle *fd);
 void file_close(Handle fd);
+int  file_set_offset(Handle fd, int off);
+int  file_get_offset(Handle fd, int *off);
 int  file_lock(Handle fd);
 int  file_unlock(Handle fd);
 int  file_sync(Handle fd);

src/file_tree.c

@@ -1,3 +1,4 @@
+#include <limits.h>
 #include <assert.h>
 #include <string.h>
 #include <stdlib.h>
@@ -374,28 +375,34 @@ int file_tree_write(FileTree *ft, string path,
     for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++)
         f->chunks[i] = hashes[i - first_chunk_index];
 
-    // Now check which old hashes are no longer used anywhere in the tree
-    *num_removed = 0;
-    for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) {
-        SHA256 old_hash = prev_hashes[i - first_chunk_index];
+    // Now check which old hashes are no longer used
+    // anywhere in the tree
+    //
+    // NOTE: If removed_hashes is NULL, the caller isn't
+    //       interested in which hashes are no longer reachable.
+    if (removed_hashes != NULL) {
+        *num_removed = 0;
+        for (uint64_t i = first_chunk_index; i <= last_chunk_index; i++) {
+            SHA256 old_hash = prev_hashes[i - first_chunk_index];
 
-        // Skip zero hashes
-        bool is_zero = true;
-        for (int j = 0; j < (int) sizeof(SHA256); j++) {
-            if (old_hash.data[j] != 0) {
-                is_zero = false;
-                break;
+            // Skip zero hashes
+            bool is_zero = true;
+            for (int j = 0; j < (int) sizeof(SHA256); j++) {
+                if (old_hash.data[j] != 0) {
+                    is_zero = false;
+                    break;
+                }
             }
-        }
-        if (is_zero)
-            continue;
+            if (is_zero)
+                continue;
 
-        // Check if this hash is still used anywhere in the tree
-        if (!entity_uses_hash(&ft->root, old_hash)) {
-            // Not used - add to removed list
-            if (removed_hashes)
-                removed_hashes[*num_removed] = old_hash;
-            (*num_removed)++;
+            // Check if this hash is still used anywhere in the tree
+            if (!entity_uses_hash(&ft->root, old_hash)) {
+                // Not used - add to removed list
+                if (removed_hashes)
+                    removed_hashes[*num_removed] = old_hash;
+                (*num_removed)++;
+            }
         }
     }
 

src/metadata_server.c

@@ -179,6 +179,10 @@ process_client_create(MetadataServer *state, int conn_idx, ByteView msg)
     if (binary_read(&reader, NULL, 1))
         return -1;
 
+    if (wal_append_create(&state->wal, path, is_dir, chunk_size) < 0) {
+        assert(0); // TODO
+    }
+
     int ret = file_tree_create_entity(&state->file_tree, path, is_dir, chunk_size);
 
     if (ret < 0) {
@@ -238,6 +242,10 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg)
     if (binary_read(&reader, NULL, 1))
         return -1;
 
+    if (wal_append_delete(&state->wal, path) < 0) {
+        assert(0); // TODO
+    }
+
     // TODO: return unused hashes and add them to the ms_rem_list of holder chunk servers
     int ret = file_tree_delete_entity(&state->file_tree, path);
 
@@ -597,6 +605,10 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg)
         new_hashes[i] = results[i].new_hash;
     }
 
+    if (wal_append_write(&state->wal, path, offset, length, num_chunks, chunk_size, old_hashes, new_hashes) < 0) {
+        assert(0); // TODO
+    }
+
     int ret = file_tree_write(&state->file_tree, path, offset, length,
         num_chunks, chunk_size, old_hashes, new_hashes, removed_hashes, &num_removed);
 
@@ -968,13 +980,18 @@ static bool is_chunk_server_message_type(uint16_t type)
 
 int metadata_server_init(MetadataServer *state, int argc, char **argv, void **contexts, struct pollfd *polled, int *timeout)
 {
-    string addr  = getargs(argc, argv, "--addr", "127.0.0.1");
-    int    port  = getargi(argc, argv, "--port", 8080);
-    bool   trace = getargb(argc, argv, "--trace");
+    string addr      = getargs(argc, argv, "--addr", "127.0.0.1");
+    int    port      = getargi(argc, argv, "--port", 8080);
+    bool   trace     = getargb(argc, argv, "--trace");
+    string wal_file  = getargs(argc, argv, "--wal-file", "metadata.wal");
+    int    wal_limit = getargi(argc, argv, "--wal-limit", 1000); // TODO: Choose a good default limit
 
     if (port <= 0 || port >= 1<<16)
         return -1;
 
+    if (wal_limit < 0)
+        return -1;
+
     state->trace = trace;
     state->replication_factor = 3; // TODO: what about the REPLICATION_FACTOR macro?
     if (state->replication_factor > MAX_CHUNK_SERVERS)
@@ -998,6 +1015,10 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
         return -1;
     }
 
+    if (wal_open(&state->wal, &state->file_tree, wal_file, wal_limit) < 0) {
+        assert(0); // TODO
+    }
+
     printf("Metadata server set up (local=%.*s:%d)\n",
         addr.len,
         addr.ptr,
@@ -1010,6 +1031,7 @@ int metadata_server_init(MetadataServer *state, int argc, char **argv, void **co
 
 int metadata_server_free(MetadataServer *state)
 {
+    wal_close(&state->wal);
     file_tree_free(&state->file_tree);
     tcp_context_free(&state->tcp);
     return 0;

src/metadata_server.h

@@ -2,6 +2,7 @@
 #define METADATA_SERVER_INCLUDED
 
 #include "tcp.h"
+#include "wal.h"
 #include "file_tree.h"
 #include "config.h"
 #include "basic.h"
@@ -39,6 +40,7 @@ typedef struct {
 typedef struct {
 
     TCP tcp;
+    WAL wal;
 
     FileTree file_tree;
 

src/wal.c

@@ -1,4 +1,9 @@
+#include <stddef.h>
+#include <limits.h>
+#include <assert.h>
+
 #include "wal.h"
+#include "file_system.h"
 #include "file_tree.h"
 
 #define WAL_MAGIC   0xcafebebe
@@ -8,9 +13,33 @@ typedef struct {
     uint32_t magic;
     uint32_t version;
     uint64_t reserved;
-} WALHeader
+} WALHeader;
+
+typedef enum {
+    WAL_ENTRY_CREATE,
+    WAL_ENTRY_DELETE,
+    WAL_ENTRY_WRITE,
+} WALEntryType;
 
 typedef struct {
+    WALEntryType type;
+
+    // create, delete, write
+    string path;
+
+    // create, write
+    uint64_t chunk_size;
+
+    // create
+    bool is_dir;
+
+    // write
+    uint64_t offset;
+    uint64_t length;
+    uint32_t num_chunks;
+    SHA256 *prev_hashes;
+    SHA256 *next_hashes;
+
 } WALEntry;
 
 typedef struct {
@@ -20,9 +49,7 @@ static int
 serialize_callback(char *src, int num, void *data)
 {
     WriteSnapshotContext *wsc = data;
-    if (write(src, num) < 0)
-        return -1;
-    return 0;
+    assert(0); // TODO
 }
 
 static int write_snapshot(FileTree *file_tree)
@@ -37,18 +64,18 @@ typedef struct {
 } ReadSnapshotContext;
 
 static int
-serialize_callback(char *src, int num, void *data)
+deserialize_callback(char *dst, int num, void *data)
 {
     ReadSnapshotContext *rsc = data;
-    // TODO
+    assert(0); // TODO
 }
 
-static int read_snapshot(FileTree *file_tree)
+static int read_snapshot(FileTree *file_tree, Handle handle)
 {
     ReadSnapshotContext rsc;
     int num = file_tree_deserialize(file_tree, deserialize_callback, &rsc);
     if (num < 0)
-        return 0;
+        return -1;
     return 0;
 }
 
@@ -66,34 +93,219 @@ static int swap_file(WAL *wal)
     assert(0); // TODO
 }
 
-int wal_open(WAL *wal, FileTree *file_tree, string file_path)
+static int next_entry(Handle handle, WALEntry *entry)
 {
-    // TODO:
-    //   - Open the log file at path "file_path"
-    //   - Lock the file
-    //   - Check that the header is correct
-    //   - Load the snapshot at the head of the file in the "file_tree" object
-    //   - Loop over the following log entries and replay them
-    //   - Then, if there are too many log entries, create a new log file and start using it
-
     assert(0); // TODO
 }
 
+int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
+{
+    wal->entry_count = 0;
+    wal->entry_limit = entry_limit;
+
+    Handle handle;
+    if (file_open(file_path, &handle) < 0)
+        return -1;
+
+    if (file_lock(handle) < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    // TODO: If the file didn't exist already, initialize it
+
+    // Read file header
+    // NOTE: For now we don't worry about fixing endianess
+    WALHeader header;
+    for (int copied = 0; copied < (int) sizeof(header); ) {
+        int ret = file_read(handle, (char*) &header + copied, (int) sizeof(header) - copied);
+        if (ret <= 0) {
+            file_close(handle); // TODO: what happens if I close a file without unlocking it?
+            return -1;
+        }
+        copied += ret;
+    }
+
+    // Validate header fields
+    if (header.magic != WAL_MAGIC) {
+        file_close(handle);
+        return -1;
+    }
+    if (header.version != WAL_VERSION) {
+        file_close(handle);
+        return -1;
+    }
+
+    // The read_snapshot function may read more
+    // bytes than necessary from the buffer, so
+    // we need to save our current position to
+    // later restore it to this offset plus what
+    // read_snapshot really consumed.
+    int saved_offset;
+    if (file_get_offset(handle, &saved_offset) < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    int num = read_snapshot(file_tree, handle);
+    if (num < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    // Not restore the offset to the correct position.
+    if (num > INT_MAX - saved_offset) {
+        file_close(handle);
+        return -1;
+    }
+    if (file_set_offset(wal->handle, saved_offset + num) < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    WALEntry entry;
+    for (;;) {
+
+        int ret = next_entry(handle, &entry);
+        if (ret == 0)
+            break;
+        if (ret < 0) {
+            file_close(handle);
+            return -1;
+        }
+        assert(ret == 1);
+
+        switch (entry.type) {
+        case WAL_ENTRY_CREATE:
+            file_tree_create_entity(file_tree, entry.path, entry.is_dir, entry.chunk_size);
+            break;
+        case WAL_ENTRY_DELETE:
+            file_tree_delete_entity(file_tree, entry.path);
+            break;
+        case WAL_ENTRY_WRITE:
+            file_tree_write(file_tree, entry.path, entry.offset, entry.length, entry.num_chunks, entry.chunk_size, entry.prev_hashes, entry.next_hashes, NULL, NULL);
+            break;
+        default:
+            UNREACHABLE;
+        }
+
+        wal->entry_count++;
+    }
+
+    if (wal->entry_count >= wal->entry_limit) {
+        if (swap_file(wal) < 0)
+            return -1;
+    }
+    return 0;
+}
+
 void wal_close(WAL *wal)
 {
-    // TODO:
-    //   - Unlock the file
-    //   - Close the file handle
-
-    assert(0); // TODO
+    file_unlock(wal->handle);
+    file_close(wal->handle);
 }
 
-int wal_append(WAL *wal, WALEntry entry)
+static int write_exact(Handle handle, char *src, int len)
 {
-    // TODO:
-    //   - If too many entries were created, start a new log file
-    //   - Write the entry
-    //   - Sync to disk
+    int copied = 0;
+    while (copied < len) {
+        int ret = file_write(handle, src + copied, len - copied);
+        if (ret < 0)
+            return -1;
+        copied += ret;
+    }
+    return 0;
+}
+
+static int write_u8(Handle handle, uint8_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_u16(Handle handle, uint16_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_u64(Handle handle, uint64_t value)
+{
+    return write_exact(handle, (char*) &value, sizeof(value));
+}
+
+static int write_str(Handle handle, string value)
+{
+    return write_exact(handle, value.ptr, value.len);
+}
+
+static int append_begin(WAL *wal)
+{
+    if (wal->entry_count >= wal->entry_limit) {
+        if (swap_file(wal) < 0)
+            return -1;
+    }
+}
+
+static int append_end(WAL *wal)
+{
+    if (file_sync(wal->handle) < 0)
+        return -1;
+    wal->entry_count++;
+    return 0;
+}
+
+int wal_append_create(WAL *wal, string path, bool is_dir, uint64_t chunk_size)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
+
+    write_u8(wal->handle, WAL_ENTRY_CREATE);
+    write_u16(wal->handle, path.len);
+    write_str(wal->handle, path);
+    write_u8(wal->handle, is_dir);
+    if (!is_dir)
+        write_u64(wal->handle, chunk_size);
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
+}
+
+int wal_append_delete(WAL *wal, string path)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
+
+    write_u8(wal->handle, WAL_ENTRY_DELETE);
+    write_u16(wal->handle, path.len);
+    write_str(wal->handle, path);
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
+}
+
+int wal_append_write(WAL *wal, string path, uint64_t off,
+    uint64_t len, uint32_t num_chunks, uint32_t chunk_size,
+    SHA256 *prev_hashes, SHA256 *hashes)
+{
+    if (path.len > UINT16_MAX)
+        return -1;
+
+    if (append_begin(wal) < 0)
+        return -1;
 
     assert(0); // TODO
+
+    if (append_end(wal) < 0)
+        return -1;
+
+    return 0;
 }

src/wal.h

@@ -1,11 +1,19 @@
 #ifndef WAL_INCLUDED
 #define WAL_INCLUDED
 
+#include "file_tree.h"
+#include "file_system.h"
+
 typedef struct {
+    Handle handle;
+    int entry_count;
+    int entry_limit;
 } WAL;
 
-int  wal_open   (WAL *wal, FileTree *file_tree, string file_path)
-void wal_close  (WAL *wal);
-int  wal_append (WAL *wal, WALEntry entry);
+int  wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit);
+void wal_close(WAL *wal);
+int  wal_append_create(WAL *wal, string path, bool is_dir, uint64_t chunk_size);
+int  wal_append_delete(WAL *wal, string path);
+int  wal_append_write(WAL *wal, string path, uint64_t off, uint64_t len, uint32_t num_chunks, uint32_t chunk_size, SHA256 *prev_hashes, SHA256 *hashes);
 
 #endif // WAL_INCLUDED