Introduce log writer APIs (log_io.cpp, log_io.hpp)

production/db/core/CMakeLists.txt

@@ -132,7 +132,8 @@ endif()
 add_library(gaia_db_persistence STATIC
   src/log_file.cpp
   src/async_disk_writer.cpp
-  src/async_write_batch.cpp)
+  src/async_write_batch.cpp
+  src/log_io.cpp)
 configure_gaia_target(gaia_db_persistence)
 target_include_directories(gaia_db_persistence PRIVATE
   "${GAIA_DB_CORE_PUBLIC_INCLUDES}"

production/db/core/inc/async_disk_writer.hpp

@@ -37,7 +37,7 @@ namespace persistence
 class async_disk_writer_t
 {
 public:
-    async_disk_writer_t(int validate_flushed_batch_efd, int signal_checkpoint_efd);
+    async_disk_writer_t(int validate_flushed_batch_eventfd, int signal_checkpoint_eventfd);
 
     ~async_disk_writer_t();
 
@@ -98,43 +98,43 @@ class async_disk_writer_t
     /**
      * Copy any temporary writes (which don't exist in gaia shared memory) into the metadata buffer.
      */
-    unsigned char* copy_into_metadata_buffer(void* source, size_t size, int file_fd);
+    unsigned char* copy_into_metadata_buffer(const void* source, size_t size, int file_fd);
 
     /**
      * Perform maintenance actions on in_flight batch after all of its IO entries have been processed.
      */
     void perform_post_completion_maintenance();
 
-    void add_decisions_to_batch(decision_list_t& decisions);
+    void add_decisions_to_batch(const decision_list_t& decisions);
 
     /**
      * For each commit ts, keep track of the eventfd which the session thread blocks on. Once the txn
      * has been made durable, this eventfd is written to so that the session thread can make progress and
      * return commit decision to the client.
      */
-    void map_commit_ts_to_session_decision_efd(gaia_txn_id_t commit_ts, int session_decision_efd);
+    void map_commit_ts_to_session_decision_eventfd(gaia_txn_id_t commit_ts, int session_decision_eventfd);
 
 private:
     // Reserve slots in the in_progress batch to be able to append additional operations to it (before it gets submitted to the kernel)
     static constexpr size_t c_submit_batch_sqe_count = 3;
     static constexpr size_t c_single_submission_entry_count = 1;
     static constexpr size_t c_async_batch_size = 32;
     static constexpr size_t c_max_iovec_array_size_bytes = IOV_MAX * sizeof(iovec);
-    static inline eventfd_t c_default_flush_efd_value = 1;
-    static inline iovec c_default_iov = {static_cast<void*>(&c_default_flush_efd_value), sizeof(eventfd_t)};
+    static inline eventfd_t c_default_flush_eventfd_value = 1;
+    static inline iovec c_default_iov = {static_cast<void*>(&c_default_flush_eventfd_value), sizeof(eventfd_t)};
 
     // eventfd to signal that a batch flush has completed.
     // Used to block new writes to disk when a batch is already getting flushed.
-    static inline int s_flush_efd = -1;
+    static inline int s_flush_eventfd = -1;
 
     // eventfd to signal that the IO results belonging to a batch are ready to be validated.
-    int m_validate_flush_efd = -1;
+    int m_validate_flush_eventfd = -1;
 
     // eventfd to signal that a file is ready to be checkpointed.
-    int m_signal_checkpoint_efd = -1;
+    int m_signal_checkpoint_eventfd = -1;
 
     // Keep track of session threads to unblock.
-    std::unordered_map<gaia_txn_id_t, int> m_ts_to_session_decision_fd_map;
+    std::unordered_map<gaia_txn_id_t, int> m_ts_to_session_decision_eventfd_map;
 
     // Writes are batched and we maintain two buffers so that writes to a buffer
     // can still proceed when the other buffer is getting flushed to disk.

production/db/core/inc/db_internal_types.hpp

@@ -83,8 +83,13 @@ constexpr size_t c_max_locators = (1ULL << 32) - 1;
 // similarly optimized by substituting locators for gaia_ids.
 constexpr size_t c_hash_buckets = 1ULL << 20;
 
-// This is arbitrary, but we need to keep txn logs to a reasonable size.
-constexpr size_t c_max_log_records = 1ULL << 20;
+// Track maximum number of new chunks (apart from the one that the txn is already using)
+// that can be allocated per transaction.
+// This sets an upper bound on txn size: 32MB < txn_size < 36MB
+constexpr size_t c_max_chunks_per_txn = 8;
+
+// 8 chunks can hold up to 8 * (2^16 - 2^8) = 522240 64B objects,
+constexpr size_t c_max_log_records = 522240;
 
 // This is an array of offsets in the data segment corresponding to object
 // versions, where each array index is referred to as a "locator."
@@ -108,6 +113,8 @@ struct txn_log_t
     // convenient place for shared state between the client and server.
     memory_manager::chunk_offset_t current_chunk;
     size_t record_count;
+    int session_decision_eventfd;
+    size_t chunk_count;
 
     struct log_record_t
     {
@@ -135,6 +142,7 @@ struct txn_log_t
     };
 
     log_record_t log_records[c_max_log_records];
+    gaia_offset_t chunks[c_max_chunks_per_txn];
 
     friend std::ostream& operator<<(std::ostream& os, const txn_log_t& l)
     {

production/db/core/inc/log_file.hpp

@@ -34,12 +34,12 @@ class log_file_t
     /**
      * Obtain offset to write the next log record at.
      */
-    size_t get_current_offset();
+    file_offset_t get_current_offset();
 
     /**
      * Get remaining space in persistent log file.
      */
-    size_t get_remaining_bytes_count(size_t record_size);
+    size_t get_bytes_remaining_after_append(size_t record_size);
 
     /**
      * Allocate space in persistent log file.
@@ -48,6 +48,11 @@ class log_file_t
 
     int get_file_fd();
 
+    /**
+     * Obtain sequence number for the file.
+     */
+    file_sequence_t get_file_sequence();
+
 private:
     size_t m_file_size;
     file_sequence_t m_file_seq;

production/db/core/inc/log_io.hpp

@@ -0,0 +1,109 @@
+/////////////////////////////////////////////
+// Copyright (c) Gaia Platform LLC
+// All rights reserved.
+/////////////////////////////////////////////
+
+#pragma once
+
+#include <fcntl.h>
+#include <unistd.h>
+
+#include <cstddef>
+
+#include <atomic>
+#include <functional>
+#include <memory>
+#include <string>
+#include <unordered_map>
+
+#include "gaia/common.hpp"
+
+#include "gaia_internal/db/db_object.hpp"
+
+#include "async_disk_writer.hpp"
+#include "db_internal_types.hpp"
+#include "log_file.hpp"
+#include "memory_manager.hpp"
+
+namespace gaia
+{
+namespace db
+{
+namespace persistence
+{
+
+/*
+ * Fill the record_header.crc field with CRC_INITIAL_VALUE when
+ * computing the checksum: crc32c is vulnerable to 0-prefixing,
+ * so we make sure the initial bytes are non-zero.
+ */
+static constexpr crc32_t c_crc_initial_value = ((uint32_t)-1);
+
+class log_handler_t
+{
+public:
+    explicit log_handler_t(const std::string& directory_path);
+    ~log_handler_t();
+    void open_for_writes(int validate_flushed_batch_eventfd, int signal_checkpoint_eventfd);
+
+    /**
+     * Allocate space in the log file and return starting offset of allocation.
+     */
+    file_offset_t allocate_log_space(size_t payload_size);
+
+    /**
+     * Create a log record which stores txn information.
+     */
+    void create_txn_record(
+        gaia_txn_id_t commit_ts,
+        record_type_t type,
+        std::vector<gaia_offset_t>& object_offsets,
+        std::vector<gaia::common::gaia_id_t>& deleted_ids);
+
+    /**
+     * Process the in memory txn_log and submit the processed writes (generated log records) to the async_disk_writer.
+     */
+    void process_txn_log_and_write(int txn_log_fd, gaia_txn_id_t commit_ts);
+
+    /**
+     * Create a log record which stores decisions for one or more txns.
+     */
+    void create_decision_record(const decision_list_t& txn_decisions);
+
+    /**
+     * Submit async_disk_writer's internal I/O request queue to the kernel for processing.
+     */
+    void submit_writes(bool sync);
+
+    /**
+     * Validate the result of I/O calls submitted to the kernel for processing.
+     */
+    void validate_flushed_batch();
+
+    /**
+     * Track the session_decision_eventfd for each commit_ts; txn_commit() will only return once 
+     * session_decision_eventfd is written to by the log_writer thread - signifying that the txn decision
+     * has been persisted.
+     */
+    void register_commit_ts_for_session_notification(gaia_txn_id_t commit_ts, int session_decision_eventfd);
+
+private:
+    // TODO: Make log file size configurable.
+    static constexpr uint64_t c_file_size = 4 * 1024 * 1024;
+    static constexpr std::string_view c_gaia_wal_dir_name = "/wal_dir";
+    static constexpr int c_gaia_wal_dir_permissions = 0755;
+    static inline std::string s_wal_dir_path{};
+    static inline int s_dir_fd = -1;
+
+    // Log file sequence starts from 1.
+    static inline std::atomic<file_sequence_t::value_type> s_file_num = 1;
+
+    // Keep track of the current log file.
+    std::unique_ptr<log_file_t> m_current_file;
+
+    std::unique_ptr<async_disk_writer_t> m_async_disk_writer;
+};
+
+} // namespace persistence
+} // namespace db
+} // namespace gaia

production/db/core/inc/persistence_types.hpp

@@ -23,6 +23,13 @@ namespace db
 namespace persistence
 {
 
+enum class record_type_t : uint8_t
+{
+    not_set = 0x0,
+    txn = 0x1,
+    decision = 0x2,
+};
+
 enum class decision_type_t : uint8_t
 {
     undecided = 0,
@@ -60,6 +67,17 @@ static_assert(
 
 constexpr file_sequence_t c_invalid_file_sequence_number;
 
+typedef size_t file_offset_t;
+
+// The record size is constrained by the size of the log file.
+// We'd never need more than 32 bits for the record size.
+typedef uint32_t record_size_t;
+
+// https://stackoverflow.com/questions/2321676/data-length-vs-crc-length
+// "From the wikipedia article: "maximal total blocklength is equal to 2**r − 1". That's in bits.
+// So CRC-32 would have max message size 2^33-1 bits or about 2^30 bytes = 1GB
+typedef uint32_t crc32_t;
+
 // This assertion ensures that the default type initialization
 // matches the value of the invalid constant.
 static_assert(
@@ -72,6 +90,25 @@ struct log_file_info_t
     int file_fd;
 };
 
+struct record_header_t
+{
+    crc32_t crc;
+    record_size_t payload_size;
+    record_type_t record_type;
+    gaia_txn_id_t txn_commit_ts;
+
+    // Stores a count value depending on the record type.
+    // For a txn record, this represents the count of deleted objects.
+    // For a decision record, this represents the number of decisions in the record's payload.
+    union
+    {
+        uint32_t deleted_object_count;
+        uint32_t decision_count;
+    };
+
+    char padding[3];
+};
+
 // The primary motivation of this buffer is to keep a hold of any additional information we want to write to the log
 // apart from the shared memory objects.
 // Custom information includes