Program Listing for File DataLogger.h#
↰ Return to documentation for file (include/Karana/KUtils/DataLogger.h)
/*
* Copyright (c) 2024-2026 Karana Dynamics Pty Ltd. All rights reserved.
*
* NOTICE TO USER:
*
* This source code and/or documentation (the "Licensed Materials") is
* the confidential and proprietary information of Karana Dynamics Inc.
* Use of these Licensed Materials is governed by the terms and conditions
* of a separate software license agreement between Karana Dynamics and the
* Licensee ("License Agreement"). Unless expressly permitted under that
* agreement, any reproduction, modification, distribution, or disclosure
* of the Licensed Materials, in whole or in part, to any third party
* without the prior written consent of Karana Dynamics is strictly prohibited.
*
* THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
* KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
* BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
* FITNESS FOR A PARTICULAR PURPOSE.
*
* IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
* INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
* OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
*
* U.S. Government End Users: The Licensed Materials are a "commercial item"
* as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
* only as a commercial end item under the terms of this license.
*
* Any use of the Licensed Materials in individual or commercial software must
* include, in the user documentation and internal source code comments,
* this Notice, Disclaimer, and U.S. Government Use Provision.
*/
/**
* @file
* @brief Contains the declarations for the DataLogger class.
*/
#pragma once
// Turn formatting off to avoid changing the order of these includes.
// clang-format off
#include "hdf5.h"
// clang-format on
#include "H5PTpublic.h"
#include "Karana/KCore/LockingBase.h"
#include "Karana/KCore/ThreadPool.h"
#include "Karana/KCore/UsageTrackingMap.h"
#include "Karana/KCore/Var_T.h"
#include "Karana/KUtils/Defs.h"
#include "Karana/Math/Defs.h"
#include <Eigen/Core>
#include <boost/lockfree/spsc_queue.hpp>
#include <cstring>
#include <format>
#include <functional>
#include <latch>
#include <memory>
#include <optional>
#include <typeindex>
namespace Karana::KUtils {
namespace kc = Karana::Core;
namespace km = Karana::Math;
// Forward declare H5Writer
class H5Writer;
/**
* @brief Write mode for packet tables; ASYNC mode can reduce overhead by writing on a separate
* thread.
*
*/
enum class PacketTableWriteMode {
/// Always write directly to file, making log() calls blocking for this table
DIRECT,
/// Log to a temporary buffer that is written to a file on a different thread;
/// this will block if it gets too far behind
ASYNC,
};
/**
* @brief A slot in the packet table's ring buffer.
*
*/
struct PacketTableSlot {
/// Pointer to the data to be written
std::unique_ptr<char[]> data;
/// Number of rows to be consumed from the pointer
unsigned int rows = 0;
/**
* @brief Mark the slot as empty so we write to the beginning
*
*/
void drain() { rows = 0; }
};
/**
* @class PacketTableConfig
* @brief Define the columns and associated update functions for a PacketTable.
* The PacketTable is the table that stores data in the H5 log file.
*/
class PacketTableConfig : public kc::LockingBase {
friend H5Writer;
public:
/**
* @brief PacketTableConfig constructor. The constructor is not meant to be called directly.
* Please use the create(...) method instead to create an instance.
*
* @param name Name of the table. A / will be interpreted as a new group.
* Groups will be added as needed.
*/
PacketTableConfig(std::string_view name);
/**
* @brief Create a packet table config.
*
* @param name Name of the table. A / will be interpreted as a new group.
* Groups will be added as needed.
* @return A ks_ptr to the newly created instance of PacketTableConfig.
*/
// Adding suppression, as CodeChecker complains this method has
// the same name as a method in the base version. This is just a
// CodeChecker false positive.
// codechecker_suppress [cppcheck-duplInheritedMember]
static kc::ks_ptr<PacketTableConfig> create(std::string_view name);
/**
* @brief Destroy packet table config.
*/
~PacketTableConfig();
/**
* @brief Add data to the table via a Var_T<km::Vec>.
*
* @param var The Var_T<km::Vec> to add to the table.
* @param as_scalars Rather than using one column for the vector, use multiple. This logs
* each vector component as a scalar. The column names will be appended
* with a _X where "X" is an integer that corresponds with the index in
* the vector.
*/
void addData(const kc::ks_ptr<kc::Var_T<km::Vec>> &var, bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
// Add type, size, and offset info.
_addVector(var->name(), var->getVectorSize(), as_scalars);
// Add the function
_funcs.push_back(var->getLoggingFunction());
}
/**
* @brief Add data to the table via a Var_T<km::Mat>.
*
* @param var The Var_T<km::Mat> to add to the table.
* @param as_scalars Rather than using one column for the vector, use multiple. This logs
* each vector component as a scalar. The column names will be appended
* with a _X where "X" is an integer that corresponds with the index in
* the vector.
*/
void addData(const kc::ks_ptr<kc::Var_T<km::Mat>> &var, bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
auto dark = var->getMatrixSize();
// Add type, size, and offset info.
_addMatrix(var->name(), dark.first, dark.second, as_scalars);
// Add the function
_funcs.push_back(var->getLoggingFunction());
}
/**
* @brief Add data to the table via a vector-valued Var_T with fixed size,
* e.g., Var_T<km::Vec3>.
*
* @param var The fixed-size, vector-valued Var_T to add to the table.
* @param as_scalars Rather than using one column for the vector, use multiple. This logs
* each vector component as a scalar. The column names will be appended
* with a _X where "X" is an integer that corresponds with the index in
* the vector.
*/
template <int rows>
void addData(const kc::ks_ptr<kc::Var_T<Eigen::Matrix<double, rows, 1>>> &var,
bool as_scalars = false) {
static_assert(rows != Eigen::Dynamic,
"Vector must have a fixed size (no dynamic rows). For a dynamic vector, "
"use the addData signature that takes in an integer for the number rows "
"as an extra argument.");
// Ensure this is not healthy.
ensureNotHealthy();
// Add type, size, and offset info.
_addVector(var->name(), rows, as_scalars);
// Add the function
_funcs.push_back(var->getLoggingFunction());
}
/**
* @brief Add data to the table via a matrix-valued Var_T with fixed size,
* e.g., Var_T<km::Mat33>.
*
* @param var The fixed-size, matrix-valued Var_T to add to the table.
* @param as_scalars Rather than using one column for the matrix, use multiple. This logs
* each matrix component as a scalar. The column names will be appended
* with a _X_Y where "X" and "Y" are integers that correspond with the
* row and index of the element in the matrix.
*/
template <int rows, int cols>
void addData(
const kc::ks_ptr<kc::Var_T<Eigen::Matrix<double, rows, cols, Eigen::RowMajor>>> &var,
bool as_scalars = false) {
static_assert(rows != Eigen::Dynamic and cols != Eigen::Dynamic,
"Matrix must have a fixed size (no dynamic rows or columns). For a "
"dynamic matrix, use the addData signature that takes in integers for "
"the rows and columns of a dynamic matrix as extra arguments.");
// Ensure this is not healthy.
ensureNotHealthy();
// Add type, size, and offset info.
_addMatrix(var->name(), rows, cols, as_scalars);
// Add the function
_funcs.push_back(var->getLoggingFunction());
}
/**
* @brief Add data to the table via a Var_T with loggable data.
*
* @param var The Var_T to add to the table.
*/
template <class T>
#ifndef PYBIND11_MKDOC_SKIP
requires(isLoggingDataType<T> and not is_eigen_type_v<T>)
#endif
void addData(const kc::ks_ptr<kc::Var_T<T>> &var) {
// Ensure this is not healthy.
ensureNotHealthy();
auto it = _h5TypeMap.find(typeid(T));
if (it == _h5TypeMap.end()) {
throw std::runtime_error(
std::format("Cannot store H5 data for \"{}\".", var->name()));
}
_h5_types.push_back(it->second);
_names.emplace_back(var->name());
// Create a copy to store in the new function
_sizes.push_back(var->getSize());
_offsets.push_back(_offset);
// Add the function
_funcs.push_back(var->getLoggingFunction());
// Increment the offset for the next function.
_offset += _sizes.back();
}
/**
* @brief Add data to the table via function.
*
* @param name The name of the column in the table.
* @param f The function that produces the data (dynamic vector). This vector must always
* be the same size.
* @param vector_size The size of the vector that the function produces. This vector must
* always be the same size.
* @param as_scalars Rather than using one column for the vector, use multiple. This logs
* each vector component as a scalar. The column names will be appended
* with a _X where "X" is an integer that corresponds with the index in
* the vector.
*/
template <class T>
requires std::same_as<std::invoke_result_t<T>, km::Vec>
void addData(std::string_view name, const T &f, int vector_size, bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
// Add type, size, and offset info.
_addVector(name, vector_size, as_scalars);
_funcs.push_back([f, size = _sizes.back()](char *buf) {
km::Vec val = f();
void *dark = val.data();
std::memcpy(buf, dark, size);
});
}
/**
* @brief Add data to the table via function.
*
* @param name The name of the column in the table.
* @param f The function that produces the data (dynamic matrix). This matrix must always
* be the same size.
* @param matrix_rows The number of rows in the matrix that the function produces. This
* matrix must always be the same size.
* @param matrix_cols The number of columns in the matrix that the function produces. This
* matrix must always be the same size.
* @param as_scalars Rather than using one column for the matrix, use multiple. This logs
* each matrix component as a scalar. The column names will be appended
* with a _X_Y where "X" and "Y" are integers that correspond with the
* row and index of the element in the matrix.
*/
template <class T>
requires std::same_as<std::invoke_result_t<T>, km::Mat>
void addData(std::string_view name,
const T &f,
int matrix_rows,
int matrix_cols,
bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
// Add type, size, and offset info.
_addMatrix(name, matrix_rows, matrix_cols, as_scalars);
_funcs.push_back([f, size = _sizes.back()](char *buf) {
km::Mat val = f();
void *dark = val.data();
std::memcpy(buf, dark, size);
});
}
/**
* @brief Add data to the table via function.
*
* @param name The name of the column in the table.
* @param f The function that produces the data.
* @param as_scalars Rather than using one column for the vector, use multiple. This logs
* each vector component as a scalar. The column names will be appended
* with a _X where "X" is an integer that corresponds with the index in
* the vector.
*/
template <class T, class R = typename std::invoke_result<T>::type>
requires(is_eigen_type_v<R> and is_eigen_vector_v<R>)
void addData(std::string_view name, const T &f, bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
constexpr int rows = extract_rows_v<R>;
static_assert(rows != Eigen::Dynamic,
"Vector must have a fixed size (no dynamic rows). For a dynamic vector, "
"use the addData signature that takes in an integer for the number rows "
"as an extra argument.");
// Add type, size, and offset info.
_addVector(name, rows, as_scalars);
_funcs.push_back([f, size = _sizes.back()](char *buf) {
Eigen::Matrix<double, rows, 1> val = f();
void *dark = val.data();
std::memcpy(buf, dark, size);
});
}
/**
* @brief Add data to the table via function.
*
* @param name The name of the column in the table.
* @param f The function that produces the data.
* @param as_scalars Rather than using one column for the matrix, use multiple. This logs
* each matrix component as a scalar. The column names will be appended
* with a _X_Y where "X" and "Y" are integers that correspond with the
* row and index of the element in the matrix.
*/
template <class T, class R = typename std::invoke_result<T>::type>
requires(is_eigen_type_v<R> and not is_eigen_vector_v<R>)
void addData(std::string_view name, const T &f, bool as_scalars = false) {
// Ensure this is not healthy.
ensureNotHealthy();
constexpr int rows = extract_rows_v<R>;
constexpr int cols = extract_cols_v<R>;
static_assert(rows != Eigen::Dynamic and cols != Eigen::Dynamic,
"Matrix must have a fixed size (no dynamic rows or columns). For a "
"dynamic matrix, use the addData signature that takes in integers for "
"the rows and columns of a dynamic matrix as extra arguments.");
// Add type, size, and offset info.
_addMatrix(name, rows, cols, as_scalars);
_funcs.push_back([f, size = _sizes.back()](char *buf) {
Eigen::Matrix<double, rows, cols, Eigen::RowMajor> val = f();
void *dark = val.data();
std::memcpy(buf, dark, size);
});
}
/**
* @brief Add data to the table via function.
* @param name The name of the column in the table.
* @param f The function that produces the data.
*/
template <class T, class R = typename std::invoke_result<T>::type>
requires(not is_eigen_type_v<R>)
void addData(std::string_view name, const T &f) {
// Ensure this is not healthy.
ensureNotHealthy();
auto it = _h5TypeMap.find(typeid(R));
if (it == _h5TypeMap.end()) {
throw std::runtime_error(std::format("Cannot store H5 data for \"{}\".", name));
}
_h5_types.push_back(it->second);
_names.emplace_back(name);
// Create a copy to store in the new function
size_t offset = _offset;
_sizes.push_back(sizeof(R));
_offsets.push_back(offset);
_funcs.push_back([f, size = _sizes.back()](char *buf) {
R val = f();
void *dark = &val;
std::memcpy(buf, dark, size);
});
// Increment the offset for the next function.
_offset += _sizes.back();
}
/**
* @brief Add another record to the buffer.
*
* @returns Whether the buffer is full and ready to write to file.
*/
bool appendRecord();
/**
* @brief Set the size of internal buffer (in number of records) to store before writing to
* disk.
*
* @param num_rows Number of records to store before flushing. Pass 0 or 1 to disable
* buffering entirely (write on every log() call).
*/
void setBufferSize(unsigned int num_rows);
/**
* @brief Set the number of slots in the async queue; when these are full data writes will
* block until a slot is free.
*
* The size of each slot in rows can be set via setBufferSize().
*
* @param size Number of slots for the writer thread queue.
*/
void setAsyncQueueSize(unsigned int size);
/**
* @brief Mark the table as flushed so we know it can be filled again.
*/
void markFlushed();
/**
* @brief Mark whether this table should write asynchronously
*
* @param is_async Boolean flag.
*/
void setAsync(bool is_async = true);
/**
* @brief Check whether this table should write asynchronously
*
* @returns Whether writing is asynchronous
*/
bool isAsync() const noexcept;
protected:
/**
* @brief Add a vector type to this PacketTableConfig.
*
* @param name The name of the data to log.
* @param rows The number of rows the vector has.
* @param as_scalars Whether to add the vector as individual scalars or as one vector.
*/
void _addVector(std::string_view name, int rows, bool as_scalars);
/**
* @brief Add a matrix type to this PacketTableConfig.
*
* @param name The name of the data to log.
* @param rows The number of rows the matrix has.
* @param cols The number of cols the matrix has.
* @param as_scalars Whether to add the matrix as individual scalars or as one matrix.
*/
void _addMatrix(std::string_view name, int rows, int cols, bool as_scalars);
/**
* @brief Create the data type and buffer for the data. This can only be done once.
*/
void _makeHealthy();
/**
* @brief This should never be called. If it is, then there is an issue.
*/
void _makeNotHealthy();
/// Writing mode (DIRECT for synchronous file writes)
PacketTableWriteMode _write_mode = PacketTableWriteMode::DIRECT;
// TODO @kelly add a setter?
/// Number of slots in the ringbuffer
unsigned int _num_async_slots = 16;
/// Slots for writing; each one contains a buffer and a counter
std::vector<PacketTableSlot> _slots;
/// SPSC Queue for tracking free slots for writing. Uses a pointer because the queue is not
/// default-constructible.
std::unique_ptr<boost::lockfree::spsc_queue<unsigned int>> _free_slots = nullptr;
/// Current slot that we are writing to, or negative for no slot
int _current_slot = -1;
/// Atomically acquire a slot from the queue
void _acquireSlot();
/**
* @brief Publish the current slot so H5Writer can submit it to the writer thread.
*
* This only marks the slot as no longer owned by the producer. H5Writer is responsible
* for submitting the task that writes the slot and returns it to the free queue.
*
* @param force Publish a partially-filled slot. Empty slots are never published.
* @return The published slot index, or std::nullopt if there is nothing to publish.
*/
std::optional<unsigned int> _publishCurrentSlot(bool force = false);
private:
/// The name of the group for this PacketTableConfig
std::string _group_name;
// The data type of the packet table.
hid_t _data_type;
/// Size of one record in bytes
size_t _row_size = 0;
/// Capacity of each slot in # rows
unsigned int _buffer_size = 1;
/**
* This vector holds the functions that add memory to the void* data buffer
* used by the packet table. */
std::vector<std::function<void(char *buf)>> _funcs;
/// The sizes for each function return type.
std::vector<size_t> _sizes;
/// The names of the columns in the packet table.
std::vector<std::string> _names;
/// The h5 types of the columns in the table.
std::vector<hid_t> _h5_types;
/// The offsets for each type.
std::vector<size_t> _offsets;
/// Memory offset used by the next function.
size_t _offset = 0;
/// Maps C++ types to H5 types.
static const std::unordered_map<std::type_index, hid_t> _h5TypeMap;
};
/**
* @class H5Writer
* @brief Logs data to an HDF5 file. The data to be logged is configured
* via `PacketTableConfig`s.
*/
class H5Writer : public kc::BaseWithVars {
public:
/**
* @brief H5Writer constructor. The constructor is not meant to be called directly.
* Please use the create(...) method instead to create an instance.
*
* @param filename The name of the h5 file to create.
*/
H5Writer(std::string_view filename);
/**
* @brief Create an H5Writer class.
*
* @param filename The name of the h5 file to create.
* @return A ks_ptr to the newly created instance of H5Writer.
*/
// Adding suppression, as CodeChecker complains this method
// has the same name as a method in the base version. This is
// just a CodeChecker false positive.
// codechecker_suppress [cppcheck-duplInheritedMember]
static kc::ks_ptr<H5Writer> create(std::string_view filename);
/**
* @brief Destroy the H5Writer. This closes the H5 file.
*/
~H5Writer();
// Delete copy constructor and assignment operator
H5Writer(const H5Writer &) = delete;
H5Writer &operator=(const H5Writer &) = delete;
/**
* @brief Create a packet table to log.
*
* This creates the table and adds it to the active tables. Activate tables
* will have a data entry added whenever log is called. To deactivate a table
* use the deactivateTable method. The name of the tame will be the same as
* the name of the PacketTableConfig.
*
* @param config The config that specifies the data of the packet table.
*/
void createTable(const kc::ks_ptr<PacketTableConfig> &config);
/**
* @brief Log data for all the active tables.
*/
void log();
/**
* @brief Create a log entry for just the given table.
*
* This will log the table regardless of whether it is active or not.
*
* @param name The name of the table to log.
*/
void logTable(const std::string &name);
/**
* @brief Activate the table with the given name.
*
* @param name The name of the table to activate.
*/
void activateTable(const std::string &name);
/**
* @brief Deactivate the table with the given name.
*
* @param name The name of the table to deactivate.
*/
void deactivateTable(const std::string &name);
/**
* @brief Get a vector of the active table names.
*
* @return A vector of the active table names.
*/
std::vector<std::string> getActiveTableNames();
/**
* @brief Get a vector of all the table names.
*
* @return A vector of the table names.
*/
std::vector<std::string> getTableNames();
/**
* @brief Flush all tables to disk.
*
*/
void flush();
private:
kc::ks_ptr<kc::ThreadPool> _writer_thread_pool;
/**
* @brief Ensure that the writer thread exists, creating one if needed
*
*/
void _ensureWriterThread();
/**
* @brief Submit a task to the writer job queue to write a slot to its file
*
* @param table_id ID of table to write
* @param slot Slot within that table to write
* @param priority whether this should be added to the front or back of queue
*/
void _submitWriteTask(hid_t table_id, unsigned int slot, bool priority);
void _waitForWriter();
/**
* @brief Shared implementation for logging a single table by id
*
* @param table_id Table ID to log now.
*/
void _logTable(hid_t table_id);
/**
* @brief Shared implementation for flushing a single table by id
*
* @param table_id Table ID to flush now.
*/
void _flushTable(hid_t table_id);
/// Holds the id of the H5 file.
hid_t _file;
/// Usage tracking map for all the packet tables.
kc::UsageTrackingMap<hid_t, PacketTableConfig> _tables;
/// A map for all the active tables.
std::map<hid_t, kc::ks_ptr<PacketTableConfig>> _active_tables;
/// A map of all the table names to their H5 IDs
std::unordered_map<std::string, hid_t> _table_names;
/// Functions to run when closing.
std::vector<std::function<void()>> _fns_on_close;
std::unordered_set<hid_t> _ids_to_close;
};
} // namespace Karana::KUtils