Complete WAL implementation with file rotation
This commit completes the Write-Ahead Log (WAL) implementation for the metadata server, including all missing functionality and error fixes: File System Changes (src/file_system.c): - Implemented file_set_offset() using lseek (Linux) and SetFilePointer (Windows) - Implemented file_get_offset() to get current file position WAL Header Changes (src/wal.h): - Added file_tree pointer to WAL structure for snapshot serialization - Added file_path to WAL structure for rotation operations WAL Implementation (src/wal.c): - Fixed handle assignment bug in wal_open (was using wal->handle before assignment) - Fixed missing return statement in append_begin() - Implemented serialize_callback for writing file tree snapshots to WAL - Implemented deserialize_callback for reading file tree snapshots from WAL - Implemented next_entry() to read WAL entries from file during recovery - Implemented wal_append_write() to write file modifications to WAL - Implemented swap_file() for complete WAL file rotation: * Creates temporary file with new snapshot * Writes WAL header and serialized file tree * Atomically replaces old WAL file * Resets entry counter for new rotation cycle - Added WAL file initialization for newly created files - Added helper functions: read_exact, read_u8, read_u16, read_u32, read_u64, write_u32 - Fixed typo in comment (Not -> Now) The WAL now supports: - Full crash recovery by replaying logged operations - Automatic file rotation when entry limit is reached - Atomic file replacement to ensure durability - Proper file locking throughout rotation process
This commit is contained in:
Claude
+30
-2
@@ -59,12 +59,40 @@ void file_close(Handle fd)
|
||||
|
||||
int file_set_offset(Handle fd, int off)
|
||||
{
|
||||
assert(0); // TODO
|
||||
#ifdef __linux__
|
||||
off_t ret = lseek((int) fd.data, off, SEEK_SET);
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
return 0;
|
||||
#endif
|
||||
|
||||
#ifdef _WIN32
|
||||
LARGE_INTEGER distance;
|
||||
distance.QuadPart = off;
|
||||
if (!SetFilePointer((HANDLE) fd.data, distance.LowPart, &distance.HighPart, FILE_BEGIN))
|
||||
if (GetLastError() != NO_ERROR)
|
||||
return -1;
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
int file_get_offset(Handle fd, int *off)
|
||||
{
|
||||
assert(0); // TODO
|
||||
#ifdef __linux__
|
||||
off_t ret = lseek((int) fd.data, 0, SEEK_CUR);
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
*off = (int) ret;
|
||||
return 0;
|
||||
#endif
|
||||
|
||||
#ifdef _WIN32
|
||||
DWORD pos = SetFilePointer((HANDLE) fd.data, 0, NULL, FILE_CURRENT);
|
||||
if (pos == INVALID_SET_FILE_POINTER && GetLastError() != NO_ERROR)
|
||||
return -1;
|
||||
*off = (int) pos;
|
||||
return 0;
|
||||
#endif
|
||||
}
|
||||
|
||||
int file_lock(Handle fd)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
#include <stddef.h>
|
||||
#include <limits.h>
|
||||
#include <assert.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "wal.h"
|
||||
#include "file_system.h"
|
||||
@@ -42,66 +43,255 @@ typedef struct {
|
||||
|
||||
} WALEntry;
|
||||
|
||||
static int write_exact(Handle handle, char *src, int len)
|
||||
{
|
||||
int copied = 0;
|
||||
while (copied < len) {
|
||||
int ret = file_write(handle, src + copied, len - copied);
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
copied += ret;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
Handle handle;
|
||||
} WriteSnapshotContext;
|
||||
|
||||
static int
|
||||
serialize_callback(char *src, int num, void *data)
|
||||
{
|
||||
WriteSnapshotContext *wsc = data;
|
||||
assert(0); // TODO
|
||||
return write_exact(wsc->handle, src, num);
|
||||
}
|
||||
|
||||
static int write_snapshot(FileTree *file_tree)
|
||||
static int write_snapshot(FileTree *file_tree, Handle handle)
|
||||
{
|
||||
WriteSnapshotContext wsc;
|
||||
wsc.handle = handle;
|
||||
if (file_tree_serialize(file_tree, serialize_callback, &wsc) < 0)
|
||||
return -1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
typedef struct {
|
||||
Handle handle;
|
||||
} ReadSnapshotContext;
|
||||
|
||||
static int
|
||||
deserialize_callback(char *dst, int num, void *data)
|
||||
{
|
||||
ReadSnapshotContext *rsc = data;
|
||||
assert(0); // TODO
|
||||
int copied = 0;
|
||||
while (copied < num) {
|
||||
int ret = file_read(rsc->handle, dst + copied, num - copied);
|
||||
if (ret <= 0)
|
||||
return -1;
|
||||
copied += ret;
|
||||
}
|
||||
return copied;
|
||||
}
|
||||
|
||||
static int read_snapshot(FileTree *file_tree, Handle handle)
|
||||
{
|
||||
ReadSnapshotContext rsc;
|
||||
rsc.handle = handle;
|
||||
int num = file_tree_deserialize(file_tree, deserialize_callback, &rsc);
|
||||
if (num < 0)
|
||||
return -1;
|
||||
return 0;
|
||||
return num;
|
||||
}
|
||||
|
||||
static int swap_file(WAL *wal)
|
||||
{
|
||||
// TODO:
|
||||
// - Create a new temporary file
|
||||
// - Write the WAL file header
|
||||
// - Serialize the current file tree
|
||||
// - Rename the temporary file to a name such that it's the next log file that will be used
|
||||
// - Delete the old log file
|
||||
// NOTE:
|
||||
// - The lock will need to be acquired at some point
|
||||
// Create a temporary file path
|
||||
char temp_path_buf[1<<10];
|
||||
if (wal->file_path.len + 5 > (int) sizeof(temp_path_buf))
|
||||
return -1;
|
||||
|
||||
assert(0); // TODO
|
||||
memcpy(temp_path_buf, wal->file_path.ptr, wal->file_path.len);
|
||||
memcpy(temp_path_buf + wal->file_path.len, ".tmp", 4);
|
||||
temp_path_buf[wal->file_path.len + 4] = '\0';
|
||||
|
||||
string temp_path = { temp_path_buf, wal->file_path.len + 4 };
|
||||
|
||||
// Create and open the temporary file
|
||||
Handle temp_handle;
|
||||
if (file_open(temp_path, &temp_handle) < 0)
|
||||
return -1;
|
||||
|
||||
// Write the WAL header
|
||||
WALHeader header;
|
||||
header.magic = WAL_MAGIC;
|
||||
header.version = WAL_VERSION;
|
||||
header.reserved = 0;
|
||||
|
||||
if (write_exact(temp_handle, (char*) &header, sizeof(header)) < 0) {
|
||||
file_close(temp_handle);
|
||||
remove_file_or_dir(temp_path);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Serialize the current file tree to the new file
|
||||
if (write_snapshot(wal->file_tree, temp_handle) < 0) {
|
||||
file_close(temp_handle);
|
||||
remove_file_or_dir(temp_path);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Sync the temporary file to ensure it's written to disk
|
||||
if (file_sync(temp_handle) < 0) {
|
||||
file_close(temp_handle);
|
||||
remove_file_or_dir(temp_path);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Lock the new file before switching
|
||||
if (file_lock(temp_handle) < 0) {
|
||||
file_close(temp_handle);
|
||||
remove_file_or_dir(temp_path);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Unlock and close the old file
|
||||
file_unlock(wal->handle);
|
||||
file_close(wal->handle);
|
||||
|
||||
// Rename the temporary file to replace the old file
|
||||
if (rename_file_or_dir(temp_path, wal->file_path) < 0) {
|
||||
file_unlock(temp_handle);
|
||||
file_close(temp_handle);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Update the WAL to use the new file handle
|
||||
wal->handle = temp_handle;
|
||||
wal->entry_count = 0;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int read_exact(Handle handle, char *dst, int len)
|
||||
{
|
||||
int copied = 0;
|
||||
while (copied < len) {
|
||||
int ret = file_read(handle, dst + copied, len - copied);
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
if (ret == 0)
|
||||
return 0; // EOF
|
||||
copied += ret;
|
||||
}
|
||||
return copied;
|
||||
}
|
||||
|
||||
static int read_u8(Handle handle, uint8_t *value)
|
||||
{
|
||||
return read_exact(handle, (char*) value, sizeof(*value));
|
||||
}
|
||||
|
||||
static int read_u16(Handle handle, uint16_t *value)
|
||||
{
|
||||
return read_exact(handle, (char*) value, sizeof(*value));
|
||||
}
|
||||
|
||||
static int read_u32(Handle handle, uint32_t *value)
|
||||
{
|
||||
return read_exact(handle, (char*) value, sizeof(*value));
|
||||
}
|
||||
|
||||
static int read_u64(Handle handle, uint64_t *value)
|
||||
{
|
||||
return read_exact(handle, (char*) value, sizeof(*value));
|
||||
}
|
||||
|
||||
static int next_entry(Handle handle, WALEntry *entry)
|
||||
{
|
||||
assert(0); // TODO
|
||||
static char path_buffer[1<<10];
|
||||
static SHA256 prev_hashes_buffer[1<<10];
|
||||
static SHA256 next_hashes_buffer[1<<10];
|
||||
|
||||
uint8_t type;
|
||||
int ret = read_u8(handle, &type);
|
||||
if (ret == 0)
|
||||
return 0; // EOF
|
||||
if (ret < 0)
|
||||
return -1;
|
||||
|
||||
entry->type = (WALEntryType) type;
|
||||
|
||||
uint16_t path_len;
|
||||
if (read_u16(handle, &path_len) <= 0)
|
||||
return -1;
|
||||
|
||||
if (path_len > sizeof(path_buffer))
|
||||
return -1;
|
||||
|
||||
if (read_exact(handle, path_buffer, path_len) <= 0)
|
||||
return -1;
|
||||
|
||||
entry->path.ptr = path_buffer;
|
||||
entry->path.len = path_len;
|
||||
|
||||
switch (entry->type) {
|
||||
case WAL_ENTRY_CREATE:
|
||||
{
|
||||
uint8_t is_dir;
|
||||
if (read_u8(handle, &is_dir) <= 0)
|
||||
return -1;
|
||||
entry->is_dir = is_dir;
|
||||
|
||||
if (!is_dir) {
|
||||
if (read_u64(handle, &entry->chunk_size) <= 0)
|
||||
return -1;
|
||||
} else {
|
||||
entry->chunk_size = 0;
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
case WAL_ENTRY_DELETE:
|
||||
// No additional fields
|
||||
break;
|
||||
|
||||
case WAL_ENTRY_WRITE:
|
||||
{
|
||||
if (read_u64(handle, &entry->offset) <= 0)
|
||||
return -1;
|
||||
if (read_u64(handle, &entry->length) <= 0)
|
||||
return -1;
|
||||
if (read_u32(handle, &entry->num_chunks) <= 0)
|
||||
return -1;
|
||||
if (read_u32(handle, (uint32_t*) &entry->chunk_size) <= 0)
|
||||
return -1;
|
||||
|
||||
if (entry->num_chunks > sizeof(prev_hashes_buffer) / sizeof(SHA256))
|
||||
return -1;
|
||||
|
||||
if (read_exact(handle, (char*) prev_hashes_buffer, entry->num_chunks * sizeof(SHA256)) <= 0)
|
||||
return -1;
|
||||
if (read_exact(handle, (char*) next_hashes_buffer, entry->num_chunks * sizeof(SHA256)) <= 0)
|
||||
return -1;
|
||||
|
||||
entry->prev_hashes = prev_hashes_buffer;
|
||||
entry->next_hashes = next_hashes_buffer;
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
return -1;
|
||||
}
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
|
||||
{
|
||||
wal->entry_count = 0;
|
||||
wal->entry_limit = entry_limit;
|
||||
wal->file_tree = file_tree;
|
||||
wal->file_path = file_path;
|
||||
|
||||
Handle handle;
|
||||
if (file_open(file_path, &handle) < 0)
|
||||
@@ -112,7 +302,41 @@ int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
|
||||
return -1;
|
||||
}
|
||||
|
||||
// TODO: If the file didn't exist already, initialize it
|
||||
// Check if the file is empty (newly created) and initialize it
|
||||
size_t size;
|
||||
if (file_size(handle, &size) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (size == 0) {
|
||||
// Initialize a new WAL file
|
||||
WALHeader header;
|
||||
header.magic = WAL_MAGIC;
|
||||
header.version = WAL_VERSION;
|
||||
header.reserved = 0;
|
||||
|
||||
if (write_exact(handle, (char*) &header, sizeof(header)) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (write_snapshot(file_tree, handle) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (file_sync(handle) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Reset to beginning after initialization
|
||||
if (file_set_offset(handle, 0) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
// Read file header
|
||||
// NOTE: For now we don't worry about fixing endianess
|
||||
@@ -153,12 +377,12 @@ int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
|
||||
return -1;
|
||||
}
|
||||
|
||||
// Not restore the offset to the correct position.
|
||||
// Now 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) {
|
||||
if (file_set_offset(handle, saved_offset + num) < 0) {
|
||||
file_close(handle);
|
||||
return -1;
|
||||
}
|
||||
@@ -192,6 +416,8 @@ int wal_open(WAL *wal, FileTree *file_tree, string file_path, int entry_limit)
|
||||
wal->entry_count++;
|
||||
}
|
||||
|
||||
wal->handle = handle;
|
||||
|
||||
if (wal->entry_count >= wal->entry_limit) {
|
||||
if (swap_file(wal) < 0)
|
||||
return -1;
|
||||
@@ -205,18 +431,6 @@ void wal_close(WAL *wal)
|
||||
file_close(wal->handle);
|
||||
}
|
||||
|
||||
static int write_exact(Handle handle, char *src, int len)
|
||||
{
|
||||
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));
|
||||
@@ -227,6 +441,11 @@ static int write_u16(Handle handle, uint16_t value)
|
||||
return write_exact(handle, (char*) &value, sizeof(value));
|
||||
}
|
||||
|
||||
static int write_u32(Handle handle, uint32_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));
|
||||
@@ -243,6 +462,7 @@ static int append_begin(WAL *wal)
|
||||
if (swap_file(wal) < 0)
|
||||
return -1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int append_end(WAL *wal)
|
||||
@@ -302,7 +522,24 @@ int wal_append_write(WAL *wal, string path, uint64_t off,
|
||||
if (append_begin(wal) < 0)
|
||||
return -1;
|
||||
|
||||
assert(0); // TODO
|
||||
if (write_u8(wal->handle, WAL_ENTRY_WRITE) < 0)
|
||||
return -1;
|
||||
if (write_u16(wal->handle, path.len) < 0)
|
||||
return -1;
|
||||
if (write_str(wal->handle, path) < 0)
|
||||
return -1;
|
||||
if (write_u64(wal->handle, off) < 0)
|
||||
return -1;
|
||||
if (write_u64(wal->handle, len) < 0)
|
||||
return -1;
|
||||
if (write_u32(wal->handle, num_chunks) < 0)
|
||||
return -1;
|
||||
if (write_u32(wal->handle, chunk_size) < 0)
|
||||
return -1;
|
||||
if (write_exact(wal->handle, (char*) prev_hashes, num_chunks * sizeof(SHA256)) < 0)
|
||||
return -1;
|
||||
if (write_exact(wal->handle, (char*) hashes, num_chunks * sizeof(SHA256)) < 0)
|
||||
return -1;
|
||||
|
||||
if (append_end(wal) < 0)
|
||||
return -1;
|
||||
|
||||
Reference in New Issue
Block a user