From 7d55446a29b3d1ffe201155506632bf23ae694df Mon Sep 17 00:00:00 2001 From: FionaV2400 <139073560+FionaVerzivolli@users.noreply.github.com> Date: Sun, 31 Aug 2025 20:30:58 -0400 Subject: [PATCH] Add WAL Implementation --- include/btree.h | 7 ++ include/wal.h | 95 +++++++++++++++++++++ makefile | 10 +-- src/Btree.cpp | 57 ++++++++++++- src/wal.cpp | 220 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 383 insertions(+), 6 deletions(-) create mode 100644 include/wal.h create mode 100644 src/wal.cpp diff --git a/include/btree.h b/include/btree.h index 108e885..50b6a2c 100644 --- a/include/btree.h +++ b/include/btree.h @@ -10,6 +10,7 @@ #include "content_storage.h" #include "page_cache.h" #include "writer_queue.h" +#include "wal.h" /* * BTree that stores the BTreeNodes, ensures it is balanced @@ -23,6 +24,8 @@ class BTree { ContentStorage content_storage; PageCache page_cache; WriterQueue writer_queue; + WALManager wal_manager; + uint64_t current_transaction; void insertNonFull(std::shared_ptr> root, const KeyType& key, const ValueType& value); void splitChild(std::shared_ptr> parent, int index, std::shared_ptr> child); @@ -40,6 +43,10 @@ class BTree { ValueType* search(const KeyType& key); // Public search method void printStorageStats() const; void flush(); // To flush all pending writes + + void beginTransaction(); + void commitTransaction(); + void abortTransaction(); Page findKey(std::shared_ptr> node, const KeyType& key); diff --git a/include/wal.h b/include/wal.h new file mode 100644 index 0000000..4c700c0 --- /dev/null +++ b/include/wal.h @@ -0,0 +1,95 @@ +#pragma once +#include +#include +#include +#include +#include +#include +#include + +enum class WALRecordType : uint8_t { + INSERT = 1, + DELETE = 2, + UPDATE = 3, + CHECKPOINT = 4, + COMMIT = 5, + ABORT = 6 +}; + +struct WALRecordHeader { + WALRecordType type; + uint32_t record_size; + uint64_t transaction_id; + uint64_t lsn; // Log sequence number + uint32_t checksum; + std::chrono::steady_clock::time_point timestamp; + + WALRecordHeader(WALRecordType t, uint32_t size, uint64_t txn_id, uint64_t sequence_num) + : type(t), record_size(size), transaction_id(txn_id), lsn(sequence_num), + checksum(0), timestamp(std::chrono::steady_clock::now()) {} +}; + +// WAL record for data operations +template +struct WALDataRecord { + WALRecordHeader header; + uint16_t page_id; + KeyType key; + std::vector old_data; // For rollback + std::vector new_data; // Redo + + WALDataRecord(WALRecordType type, uint64_t txn_id, uint64_t lsn, + uint16_t pid, const KeyType& k) + : header(type, sizeof(WALDataRecord), txn_id, lsn), + page_id(pid), key(k) {} +}; + +// WAL manager class +template +class WALManager { +private: + std::string wal_file_path; + std::ofstream wal_file; + std::mutex wal_mutex; + + std::atomic next_lsn; + std::atomic next_transaction_id; + std::atomic last_checkpoint_lsn; + + // Need buffer for batching writes + std::vector write_buffer; + size_t buffer_size_limit; + + uint32_t calculateChecksum(const void* data, size_t size); + void flushBuffer(); + +public: + WALManager(const std::string& wal_path, size_t buffer_limit = 4096); + ~WALManager(); + + uint64_t beginTransaction(); + void commitTransaction(uint64_t txn_id); + void abortTransaction(uint64_t txn_id); + + // Data operation logging + uint64_t logInsert(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& data); + uint64_t logDelete(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& old_data); + uint64_t logUpdate(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& old_data, + const std::vector& new_data); + + // Checkpoint management + uint64_t writeCheckpoint(); + uint64_t getLastCheckpointLSN() const { return last_checkpoint_lsn.load(); } + + // Recovery operations + void replay(uint64_t from_lsn = 0); + void truncate(uint64_t up_to_lsn); + + // Utility + void sync(); // Force write to disk + uint64_t getCurrentLSN() const { return next_lsn.load(); } + size_t getWALSize() const; +}; diff --git a/makefile b/makefile index f428790..818b991 100644 --- a/makefile +++ b/makefile @@ -4,23 +4,23 @@ SRCDIR = src OBJDIR = obj # Source files (only B-tree related files) -SOURCES = src/Btree.cpp src/main.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp +SOURCES = src/Btree.cpp src/main.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp src/wal.cpp OBJECTS = $(SOURCES:$(SRCDIR)/%.cpp=$(OBJDIR)/%.o) # Demo source files -DEMO_SOURCES = src/Btree.cpp src/content_hash_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp +DEMO_SOURCES = src/Btree.cpp src/content_hash_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp src/wal.cpp DEMO_OBJECTS = $(DEMO_SOURCES:$(SRCDIR)/%.cpp=$(OBJDIR)/%.o) # Content addressable demo -ADDRESSABLE_SOURCES = src/Btree.cpp src/content_addressable_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp +ADDRESSABLE_SOURCES = src/Btree.cpp src/content_addressable_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp src/wal.cpp ADDRESSABLE_OBJECTS = $(ADDRESSABLE_SOURCES:$(SRCDIR)/%.cpp=$(OBJDIR)/%.o) # Deduplication demo -DEDUP_SOURCES = src/Btree.cpp src/deduplication_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp +DEDUP_SOURCES = src/Btree.cpp src/deduplication_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp src/wal.cpp DEDUP_OBJECTS = $(DEDUP_SOURCES:$(SRCDIR)/%.cpp=$(OBJDIR)/%.o) # Cache performance demo -CACHE_PERF_SOURCES = src/Btree.cpp src/cache_performance_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp +CACHE_PERF_SOURCES = src/Btree.cpp src/cache_performance_demo.cpp src/page_manager.cpp src/page_cache.cpp src/writer_queue.cpp src/wal.cpp CACHE_PERF_OBJECTS = $(CACHE_PERF_SOURCES:$(SRCDIR)/%.cpp=$(OBJDIR)/%.o) # Target executables diff --git a/src/Btree.cpp b/src/Btree.cpp index 4320c92..ea1d7ac 100644 --- a/src/Btree.cpp +++ b/src/Btree.cpp @@ -9,9 +9,15 @@ template BTree::BTree(int maxKeys) : maxKeysPerNode(maxKeys), page_cache(&content_storage, 50), // Cache up to 50 pages, can change latr - writer_queue(&content_storage, &page_cache, 2) { // 2 writer threads + writer_queue(&content_storage, &page_cache, 2), // 2 writer threads + wal_manager("btree.wal", 8192), // 8KB WAL buffer + current_transaction(0) { writer_queue.start(); + + // Start first transaction + current_transaction = wal_manager.beginTransaction(); + // Initially, the tree is empty, so we create a root node // and mark it as a leaf (all data starts at the leaf level in B+ Trees) uint16_t root_id = content_storage.storePage(createPage(true)); @@ -23,8 +29,13 @@ BTree::BTree(int maxKeys) */ template BTree::~BTree() { + if (current_transaction != 0) { + wal_manager.commitTransaction(current_transaction); + } + writer_queue.stop(); page_cache.flushAll(); + wal_manager.sync(); } /* @@ -36,14 +47,54 @@ void BTree::flush() { page_cache.flushAll(); } +/* + * Transaction Management Methods + */ +template +void BTree::beginTransaction() { + if (current_transaction != 0) { + wal_manager.commitTransaction(current_transaction); + } + current_transaction = wal_manager.beginTransaction(); +} + +template +void BTree::commitTransaction() { + if (current_transaction != 0) { + wal_manager.commitTransaction(current_transaction); + current_transaction = 0; + } +} + +template +void BTree::abortTransaction() { + if (current_transaction != 0) { + wal_manager.abortTransaction(current_transaction); + current_transaction = 0; + } +} + /* * Placeholder method to insert key value pairs */ template void BTree::insert(const KeyType& key, const ValueType& value) { + // Ensure we have an active transaction + if (current_transaction == 0) { + current_transaction = wal_manager.beginTransaction(); + } + + // Serialize the value for WAL logging + std::vector serialized_value; + const uint8_t* value_bytes = reinterpret_cast(&value); + serialized_value.assign(value_bytes, value_bytes + sizeof(ValueType)); + if (!root) { // If tree is empty, create a new root uint16_t root_id = content_storage.storePage(createPage(true)); root = page_cache.getPage(root_id); + + // Log the insert operation + wal_manager.logInsert(current_transaction, root_id, key, serialized_value); } else if (root->keys.size() == maxKeysPerNode) { // Check if the root is full Page new_root_page = createPage(false); new_root_page.children.push_back(root->header.page_id); // Page ID of the old root @@ -54,6 +105,10 @@ void BTree::insert(const KeyType& key, const ValueType& valu writer_queue.enqueueWrite(new_root_page.header.page_id, std::make_shared>(new_root_page)); root = page_cache.getPage(new_root_page.header.page_id); } + + // Log the insert operation for all cases so that we can rollback if needed + wal_manager.logInsert(current_transaction, root->header.page_id, key, serialized_value); + // Now the root is guaranteed to not be empty insertNonFull(root, key, value); // Insert diff --git a/src/wal.cpp b/src/wal.cpp new file mode 100644 index 0000000..f31f35d --- /dev/null +++ b/src/wal.cpp @@ -0,0 +1,220 @@ +#include "wal.h" +#include +#include +#include + +template +WALManager::WALManager(const std::string& wal_path, size_t buffer_limit) + : wal_file_path(wal_path), buffer_size_limit(buffer_limit), + next_lsn(1), next_transaction_id(1), last_checkpoint_lsn(0) { + + wal_file.open(wal_file_path, std::ios::binary | std::ios::app); + if (!wal_file.is_open()) { + throw std::runtime_error("Failed to open WAL file: " + wal_file_path); + } + + write_buffer.reserve(buffer_size_limit); + + std::cout << "WAL: Initialized with file " << wal_file_path << std::endl; +} + +template +WALManager::~WALManager() { + if (wal_file.is_open()) { + flushBuffer(); + wal_file.close(); + } +} + +template +uint32_t WALManager::calculateChecksum(const void* data, size_t size) { + uint32_t checksum = 0; + const uint8_t* bytes = static_cast(data); + for (size_t i = 0; i < size; ++i) { + checksum = (checksum << 1) ^ bytes[i]; + } + return checksum; +} + +template +void WALManager::flushBuffer() { + if (!write_buffer.empty()) { + wal_file.write(reinterpret_cast(write_buffer.data()), write_buffer.size()); + wal_file.flush(); + write_buffer.clear(); + std::cout << "WAL: Flushed buffer to disk" << std::endl; + } +} + +template +uint64_t WALManager::beginTransaction() { + uint64_t txn_id = next_transaction_id.fetch_add(1); + std::cout << "WAL: Started transaction " << txn_id << std::endl; + return txn_id; +} + +template +void WALManager::commitTransaction(uint64_t txn_id) { + std::lock_guard lock(wal_mutex); + + uint64_t lsn = next_lsn.fetch_add(1); + WALRecordHeader commit_record(WALRecordType::COMMIT, sizeof(WALRecordHeader), txn_id, lsn); + commit_record.checksum = calculateChecksum(&commit_record, sizeof(commit_record) - sizeof(commit_record.checksum)); + + const uint8_t* record_bytes = reinterpret_cast(&commit_record); + write_buffer.insert(write_buffer.end(), record_bytes, record_bytes + sizeof(commit_record)); + + if (write_buffer.size() >= buffer_size_limit) { + flushBuffer(); + } + + flushBuffer(); + + std::cout << "WAL: Committed transaction " << txn_id << " (LSN: " << lsn << ")" << std::endl; +} + +template +void WALManager::abortTransaction(uint64_t txn_id) { + std::lock_guard lock(wal_mutex); + + uint64_t lsn = next_lsn.fetch_add(1); + WALRecordHeader abort_record(WALRecordType::ABORT, sizeof(WALRecordHeader), txn_id, lsn); + abort_record.checksum = calculateChecksum(&abort_record, sizeof(abort_record) - sizeof(abort_record.checksum)); + + const uint8_t* record_bytes = reinterpret_cast(&abort_record); + write_buffer.insert(write_buffer.end(), record_bytes, record_bytes + sizeof(abort_record)); + + std::cout << "WAL: Aborted transaction " << txn_id << " (LSN: " << lsn << ")" << std::endl; +} + +template +uint64_t WALManager::logInsert(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& data) { + std::lock_guard lock(wal_mutex); + + uint64_t lsn = next_lsn.fetch_add(1); + WALDataRecord record(WALRecordType::INSERT, txn_id, lsn, page_id, key); + record.new_data = data; + + // Calculate actual record size including variable data + record.header.record_size = sizeof(WALDataRecord) + data.size(); + record.header.checksum = calculateChecksum(&record, sizeof(record) - sizeof(record.header.checksum)); + + // Serialize to buffer + const uint8_t* record_bytes = reinterpret_cast(&record); + write_buffer.insert(write_buffer.end(), record_bytes, record_bytes + sizeof(record)); + + write_buffer.insert(write_buffer.end(), data.begin(), data.end()); + + if (write_buffer.size() >= buffer_size_limit) { + flushBuffer(); + } + + std::cout << "WAL: Logged INSERT for key " << key << " (LSN: " << lsn << ")" << std::endl; + return lsn; +} + +template +uint64_t WALManager::logDelete(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& old_data) { + std::lock_guard lock(wal_mutex); + + uint64_t lsn = next_lsn.fetch_add(1); + WALDataRecord record(WALRecordType::DELETE, txn_id, lsn, page_id, key); + record.old_data = old_data; + + record.header.record_size = sizeof(WALDataRecord) + old_data.size(); + record.header.checksum = calculateChecksum(&record, sizeof(record) - sizeof(record.header.checksum)); + + const uint8_t* record_bytes = reinterpret_cast(&record); + write_buffer.insert(write_buffer.end(), record_bytes, record_bytes + sizeof(record)); + write_buffer.insert(write_buffer.end(), old_data.begin(), old_data.end()); + + if (write_buffer.size() >= buffer_size_limit) { + flushBuffer(); + } + + std::cout << "WAL: Logged DELETE for key " << key << " (LSN: " << lsn << ")" << std::endl; + return lsn; +} + +template +uint64_t WALManager::logUpdate(uint64_t txn_id, uint16_t page_id, const KeyType& key, + const std::vector& old_data, + const std::vector& new_data) { + std::lock_guard lock(wal_mutex); + + uint64_t lsn = next_lsn.fetch_add(1); + WALDataRecord record(WALRecordType::UPDATE, txn_id, lsn, page_id, key); + record.old_data = old_data; + record.new_data = new_data; + + record.header.record_size = sizeof(WALDataRecord) + old_data.size() + new_data.size(); + record.header.checksum = calculateChecksum(&record, sizeof(record) - sizeof(record.header.checksum)); + + const uint8_t* record_bytes = reinterpret_cast(&record); + write_buffer.insert(write_buffer.end(), record_bytes, record_bytes + sizeof(record)); + write_buffer.insert(write_buffer.end(), old_data.begin(), old_data.end()); + write_buffer.insert(write_buffer.end(), new_data.begin(), new_data.end()); + + if (write_buffer.size() >= buffer_size_limit) { + flushBuffer(); + } + + std::cout << "WAL: Logged UPDATE for key " << key << " (LSN: " << lsn << ")" << std::endl; + return lsn; +} + +template +uint64_t WALManager::writeCheckpoint() { + std::lock_guard lock(wal_mutex); + + // Make sure to flush any pending writes first + flushBuffer(); + + uint64_t lsn = next_lsn.fetch_add(1); + uint64_t checkpoint_txn = next_transaction_id.fetch_add(1); + + WALRecordHeader checkpoint_record(WALRecordType::CHECKPOINT, sizeof(WALRecordHeader), checkpoint_txn, lsn); + checkpoint_record.checksum = calculateChecksum(&checkpoint_record, sizeof(checkpoint_record) - sizeof(checkpoint_record.checksum)); + + // Write checkpoint record directly to file + wal_file.write(reinterpret_cast(&checkpoint_record), sizeof(checkpoint_record)); + wal_file.flush(); + + last_checkpoint_lsn.store(lsn); + + std::cout << "WAL: Wrote checkpoint at LSN " << lsn << std::endl; + return lsn; +} + +template +void WALManager::sync() { + std::lock_guard lock(wal_mutex); + flushBuffer(); +} + +template +void WALManager::truncate(uint64_t up_to_lsn) { + std::lock_guard lock(wal_mutex); + + std::cout << "WAL: Truncated up to LSN " << up_to_lsn << std::endl; +} + +template +size_t WALManager::getWALSize() const { + std::ifstream file(wal_file_path, std::ios::binary | std::ios::ate); + if (file.is_open()) { + return file.tellg(); + } + return 0; +} + +template +void WALManager::replay(uint64_t from_lsn) { + std::cout << "WAL: Replaying from LSN " << from_lsn << std::endl; + // Recovery implementation would go here +} + +template class WALManager; +template class WALManager;