record_logger.hpp

#pragma once
 
#include <any>
#include <atomic>
#include <cassert>
#include <chrono>
#include <memory>
#include <optional>
#include <spdlog/spdlog.h>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>
 
#ifndef NDEBUG
    #include <iostream>
    #include <sstream>
#endif
 
#include "phonebook.hpp"
 
namespace ILLIXR {
 
class record_header {
public:
    record_header(const std::string& name_, std::vector<std::pair<std::string, const std::type_info&>> columns_)
        : id{std::hash<std::string>{}(name_)}
        , name{name_}
        , columns{std::move(columns_)} { }
 
    bool operator==(const record_header& other) const {
        // Check pointer first
        if (this == &other) {
            return true;
        }
 
        if (name != other.name || columns.size() != other.columns.size() || id != other.id) {
            return false;
        }
        for (std::size_t i = 0; i < columns.size(); ++i) {
            if (columns[i] != other.columns[i]) {
                return false;
            }
        }
        return true;
    }
 
    bool operator!=(const record_header& other) const {
        return !(*this == other);
    }
 
    [[nodiscard]] std::size_t get_id() const {
        return id;
    }
 
    [[nodiscard]] const std::string& get_name() const {
        return name;
    }
 
    [[nodiscard]] const std::string& get_column_name(unsigned column) const {
        return columns[column].first;
    }
 
    [[nodiscard]] const std::type_info& get_column_type(unsigned column) const {
        return columns[column].second;
    }
 
    [[nodiscard]] unsigned get_columns() const {
        return columns.size();
    }
 
    [[nodiscard]] std::string to_string() const {
        std::string ret = std::string{"record_header "} + name + std::string{" { "};
        for (const auto& pair : columns) {
            ret += std::string{pair.second.name()} + std::string{" "} + pair.first + std::string{"; "};
        }
        ret.erase(ret.size() - 2);
        ret += std::string{" }"};
        return ret;
    }
 
private:
    std::size_t                                                      id;
    std::string                                                      name;
    const std::vector<std::pair<std::string, const std::type_info&>> columns;
};
 
class data_use_indicator {
public:
    data_use_indicator()
        : used{false} { }
 
    data_use_indicator(const data_use_indicator& other)
        : used{false} {
        other.used = true;
    }
 
    data_use_indicator& operator=(const data_use_indicator& other) {
        if (&other != this) {
            other.used = true;
            used       = false;
        }
        return *this;
    }
 
    /*
      copy constructors are just as efficient as move constructors would be,
      so I won't define move constructors. C++ will invoke copy instead (for no loss).
    */
    bool is_used() const {
        return used;
    }
 
    void mark_used() const {
        used = true;
    }
 
    void mark_unused() const {
        used = false;
    }
 
private:
    mutable bool used;
};
 
class record {
public:
    record(const record_header& rh_, std::vector<std::any> values_)
        : rh{rh_}
        , values(std::move(values_)) {
#ifndef NDEBUG
        assert(rh);
        if (values.size() != rh->get().get_columns()) {
            spdlog::get("illixr")->error("[record_logger] {} elements passed, but rh for {} only specifies {}.", values.size(),
                                         rh->get().get_name(), rh->get().get_columns());
            abort();
        }
        for (std::size_t column = 0; column < values.size(); ++column) {
            if (values[column].type() != rh->get().get_column_type(column)) {
                spdlog::get("illixr")->error("[record_logger] Caller got wrong type for column {} of {}.", column,
                                             rh->get().get_name());
                spdlog::get("illixr")->error("[record_logger] Caller passed: {}; record_header specifies: {}",
                                             values[column].type().name(), rh->get().get_column_type(column).name());
                abort();
            }
        }
#endif
    }
 
    record() = default;
 
    ~record() {
#ifndef NDEBUG
        if (rh && !data_use_indicator_.is_used()) {
            spdlog::get("illixr")->error("[record_logger] Record was deleted without being logged.");
            abort();
        }
#endif
    }
 
    template<typename T>
    T get_value(unsigned column) const {
#ifndef NDEBUG
        assert(rh);
        data_use_indicator_.mark_used();
        if (rh->get().get_column_type(column) != typeid(T)) {
            std::ostringstream ss;
            ss << "Caller column type for " << column << " of " << rh->get().get_name() << ". "
               << "Caller passed: " << typeid(T).name() << "; "
               << "record_header specifies: " << rh->get().get_column_type(column).name() << ". ";
            throw std::runtime_error{ss.str()};
        }
#endif
        return std::any_cast<T>(values[column]);
    }
 
    const record_header& get_record_header() const {
        assert(rh);
        return rh->get();
    }
 
    void mark_used() const {
#ifndef NDEBUG
        assert(rh);
        data_use_indicator_.mark_used();
#endif
    }
 
private:
    // Holding a pointer to a record_header is more efficient than
    // requiring each record to hold a list of its column names
    // and table name. This is just one pointer.
    std::optional<std::reference_wrapper<const record_header>> rh;
    std::vector<std::any>                                      values;
#ifndef NDEBUG
    data_use_indicator data_use_indicator_;
#endif
};
 
class record_logger : public phonebook::service {
public:
    ~record_logger() override = default;
 
    virtual void log(const record& r) = 0;
 
    virtual void log(const std::vector<record>& rs) {
        for (const record& r : rs) {
            log(r);
        }
    }
};
 
class gen_guid : public phonebook::service {
public:
    std::size_t get(std::size_t namespace_ = 0, std::size_t subnamespace = 0, std::size_t subsubnamespace = 0) {
        if (guid_starts[namespace_][subnamespace].count(subsubnamespace) == 0) {
            guid_starts[namespace_][subnamespace][subsubnamespace].store(1);
        }
        return guid_starts[namespace_][subnamespace][subsubnamespace]++;
    }
 
private:
    std::unordered_map<std::size_t, std::unordered_map<std::size_t, std::unordered_map<std::size_t, std::atomic<std::size_t>>>>
        guid_starts;
};
 
static std::chrono::milliseconds LOG_BUFFER_DELAY{1000};
 
class record_coalescer {
private:
    std::shared_ptr<record_logger>                              logger;
    std::chrono::time_point<std::chrono::high_resolution_clock> last_log;
    std::vector<record>                                         buffer;
 
public:
    explicit record_coalescer(std::shared_ptr<record_logger> logger_)
        : logger{std::move(logger_)}
        , last_log{std::chrono::high_resolution_clock::now()} { }
 
    ~record_coalescer() {
        flush();
    }
 
    void log(const record& r) {
        if (logger) {
            buffer.push_back(r);
            // Log coalescer should only be used with
            // In the common case, they will be the same pointer, quickly check the pointers.
            // In the less common case, we check for object-structural equality.
#ifndef NDEBUG
            if (&r.get_record_header() != &buffer[0].get_record_header() &&
                r.get_record_header() == buffer[0].get_record_header()) {
                spdlog::get("illixr")->error("[record_logger] Tried to push a record of type {} to a record logger for type {}",
                                             r.get_record_header().to_string(), buffer[0].get_record_header().to_string());
                abort();
            }
#endif
            maybe_flush();
        }
    }
 
    void maybe_flush() {
        if (std::chrono::high_resolution_clock::now() > last_log + LOG_BUFFER_DELAY) {
            flush();
        }
    }
 
    void flush() {
        if (logger) {
            std::vector<record> buffer2;
            buffer.swap(buffer2);
            logger->log(buffer2);
            last_log = std::chrono::high_resolution_clock::now();
        }
    }
 
    explicit operator bool() const {
        return bool(logger);
    }
};
} // namespace ILLIXR