You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
258 lines
7.8 KiB
258 lines
7.8 KiB
#include "registry.h"
|
|
#include "disruptor.h"
|
|
#include "sl/register.h"
|
|
|
|
#include <concepts>
|
|
#include <atomic>
|
|
#include <unordered_map>
|
|
#include <functional>
|
|
#include <any>
|
|
#include "nlohmann/json.hpp"
|
|
|
|
#if defined(__linux__) || defined(__linux)
|
|
#include <sys/mman.h>
|
|
#include <fcntl.h>
|
|
#include <memory>
|
|
#include <thread>
|
|
#include <future>
|
|
#include <fstream>
|
|
|
|
#endif
|
|
|
|
/**
|
|
* @brief A fast semaphore exclusion handler WITHOUT deadlock detection or yielding
|
|
*/
|
|
template<std::integral T, T default_increment, T maximum_increment>
|
|
class fast_semaphore {
|
|
std::atomic<T> flag; //< This is our counter
|
|
public:
|
|
fast_semaphore() = default;
|
|
fast_semaphore(fast_semaphore&) = delete;
|
|
fast_semaphore(fast_semaphore&&) = delete;
|
|
|
|
/// 3 things may happen when trying to unlock
|
|
enum class tristate {
|
|
success = 1, //< The unlocking was successful
|
|
timing = 0, //< Someone interfered with the unlocking
|
|
error = -1 //< The unlocking would over-unlock the semaphore
|
|
};
|
|
|
|
/// We try locking until we succeed
|
|
template<T increment = default_increment>
|
|
void lock() {
|
|
while(not try_lock<increment>());
|
|
}
|
|
|
|
/// For locking, we try to atomically increment the counter while maintaining it below the set limit
|
|
template<T increment = default_increment>
|
|
[[nodiscard]] bool try_lock() {
|
|
T expect = flag.load(std::memory_order::acquire);
|
|
T target = expect + increment;
|
|
if(target > maximum_increment) return false;
|
|
return flag.compare_exchange_strong(expect,target,std::memory_order::release);
|
|
}
|
|
|
|
/// Similarly to locking, we try unlocking until we succeed (or reach an invalid state)
|
|
template<T increment = default_increment>
|
|
void unlock() {
|
|
tristate v;
|
|
do{ v = try_unlock<increment>(); }
|
|
while(v == tristate::timing);
|
|
if(v != tristate::success) {
|
|
throw std::runtime_error("Over unlocking may have happened: potential double unlocking issue");
|
|
}
|
|
}
|
|
|
|
/// Unlocking is the reverse of locking, we have to ensure to return an error if we try to go below zero
|
|
template<T increment = default_increment>
|
|
[[nodiscard]] tristate try_unlock() {
|
|
T expect = flag.load(std::memory_order::relaxed);
|
|
T target = expect - increment;
|
|
if(target < 0) return tristate::error;
|
|
return flag.compare_exchange_strong(expect,target,std::memory_order::release) ? tristate::success : tristate::timing;
|
|
}
|
|
};
|
|
|
|
using rw_lock_type = fast_semaphore<int32_t, 256, 256>;
|
|
|
|
class lock_handler_read {
|
|
rw_lock_type& ref;
|
|
|
|
public:
|
|
explicit lock_handler_read(rw_lock_type& _ref) : ref(_ref) {
|
|
while(ref.try_lock<1>());
|
|
}
|
|
~lock_handler_read() {
|
|
while(ref.try_unlock<1>() != rw_lock_type::tristate::success);
|
|
}
|
|
};
|
|
|
|
class lock_handler_write {
|
|
rw_lock_type& ref;
|
|
|
|
public:
|
|
explicit lock_handler_write(rw_lock_type& _ref) : ref(_ref) {
|
|
while(ref.try_lock());
|
|
}
|
|
~lock_handler_write() {
|
|
while(ref.try_unlock() != rw_lock_type::tristate::success);
|
|
}
|
|
};
|
|
|
|
static fast_semaphore<int32_t, 256, 256> registry_rw_lock;
|
|
|
|
std::unordered_map<int, registry_slab> registry_map;
|
|
|
|
template<BufferStrategy buff_strat, SinkStrategy sink_strat, OverflowStrategy of_strat, OutputStrategy out_strat>
|
|
void register_log_impl(int log_id, std::string_view log_name, std::string_view filename, uint64_t buffer_size, uint64_t out_strategy_parameter, std::string_view output_directory) {
|
|
if(buffer_size % page_size != 0) {
|
|
// TODO: more meaningful exception
|
|
throw disruptor_exception{"Size provided for the disruptor doesn't align with page size"};
|
|
}
|
|
lock_handler_write lock_me{registry_rw_lock};
|
|
|
|
if(registry_map.contains(log_id)) {
|
|
// TODO: more meaningful exception
|
|
throw disruptor_exception{"Log ID already exists"};
|
|
}
|
|
|
|
registry_map.insert(std::pair<int, registry_slab>{log_id, {}});
|
|
auto& slab = registry_map[log_id];
|
|
slab.id = log_id;
|
|
slab.name = log_name;
|
|
auto* transient_reference = new disruptor<of_strat>(buff_strat{}, filename, buffer_size);
|
|
slab.disruptor = transient_reference;
|
|
slab.reserve_write = [ptr = transient_reference] (size_t sz) -> token_t{
|
|
return ptr->reserve_write(sz);
|
|
};
|
|
slab.reserve_write_c_align = [ptr = transient_reference] (size_t sz) -> token_t{
|
|
return ptr->reserve_write(sz, force_contiguous_mode);
|
|
};
|
|
slab.conclude_write = [ptr = transient_reference] (token_t tok) -> void {
|
|
ptr->conclude_write(tok);
|
|
};
|
|
slab.get_buffer = [self = slab, ptr = transient_reference] (token_t tok) -> write_span {
|
|
return write_span(tok, *ptr);
|
|
};
|
|
}
|
|
|
|
BufferStrategyInternal::buffer_type BufferStrategyInternal::build_buffer(std::string_view, size_t sz) {
|
|
return std::vector<char>(sz, 0);
|
|
}
|
|
|
|
BufferStrategyShared::buffer_type BufferStrategyShared::build_buffer(std::string_view, size_t sz) {
|
|
#if defined(__linux__) || defined(__linux)
|
|
auto buff = mmap(nullptr, sz, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE, -1, 0);
|
|
if(not buff) throw disruptor_exception{"Could not allocate memory for BufferStrategyShared"};
|
|
return {(char*)buff, sz};
|
|
#else
|
|
static_assert(false, "BufferStrategyShared strategy is unimplemented on your system");
|
|
#endif
|
|
}
|
|
|
|
BufferStrategyExternal::buffer_type BufferStrategyExternal::build_buffer(std::string_view filename, size_t sz) {
|
|
#if defined(__linux__) || defined(__linux)
|
|
auto fd = open(std::string{filename}.c_str(), O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR);
|
|
if(fd <= 0) throw disruptor_exception{"Could not open or create file for BufferStrategyExternal"};
|
|
if(ftruncate(fd, sz) != 0) throw disruptor_exception{"Could not ensure size for the file for BufferStrategyExternal"};
|
|
auto buff = mmap(nullptr, sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, 0);
|
|
if(not buff) {
|
|
throw disruptor_exception{"Could not allocate memory for BufferStrategyExternal, mapping failed"};
|
|
}
|
|
return {(char*)buff, sz};
|
|
#else
|
|
static_assert(false, "BufferStrategyExternal strategy is unimplemented on your system");
|
|
#endif
|
|
}
|
|
|
|
void SinkStrategyDirect::write(int fd, std::string_view data) {
|
|
do{
|
|
auto written = ::write(fd, data.data(), data.size());
|
|
if(written < 0) {
|
|
throw std::runtime_error("POSIX write failed");
|
|
} else if(written == 0) {
|
|
throw std::runtime_error("POSIX write made no progress");
|
|
}
|
|
data.substr(written);
|
|
} while(not data.empty());
|
|
}
|
|
|
|
void SinkStrategyFastest::write(int fd, std::string_view data) {
|
|
|
|
}
|
|
|
|
void SinkStrategyMmaped::write(int fd, std::string_view data) {
|
|
|
|
}
|
|
|
|
void SinkStrategyExternal::write(int fd, std::string_view data) {
|
|
|
|
}
|
|
|
|
void OverflowStrategyWait::wait() {
|
|
std::this_thread::yield();
|
|
}
|
|
|
|
void OverflowStrategyContinue::wait() {
|
|
|
|
}
|
|
|
|
std::pair<std::string_view, std::string_view> OutputStrategyTimed::chunk(std::string_view) {
|
|
return {};
|
|
}
|
|
|
|
int OutputStrategyTimed::on_write_completed_event(std::string_view, int) {
|
|
return 0;
|
|
}
|
|
|
|
std::pair<std::string_view, std::string_view> OutputStrategySized::chunk(std::string_view) {
|
|
return {};
|
|
}
|
|
|
|
int OutputStrategySized::on_write_completed_event(std::string_view, int) {
|
|
return 0;
|
|
}
|
|
|
|
std::pair<std::string_view, std::string_view> OutputStrategySimple::chunk(std::string_view) {
|
|
return {};
|
|
}
|
|
|
|
int OutputStrategySimple::on_write_completed_event(std::string_view, int) {
|
|
return 0;
|
|
}
|
|
|
|
static nlohmann::json type_check_log_config(nlohmann::json&& data) {
|
|
if(not data.is_object()) return {};
|
|
|
|
if(not data.contains("on_overflow") and not data["on_overflow"].is_string()) return {};
|
|
if(not data.contains("buffer_type")) return {};
|
|
if(auto& type = data.at("buffer_type"); not (type.is_string() and type.get<std::string>() == "internal") and not data.contains("buffer_filename")) return {};
|
|
}
|
|
|
|
static void process_one(const nlohmann::json& data) {
|
|
|
|
}
|
|
|
|
void sl::process_log_config(const std::string_view &filename) {
|
|
std::ifstream config{std::string(filename)};
|
|
nlohmann::json data;
|
|
config >> data;
|
|
if(data.is_array()) {
|
|
for(auto& elem : data) {
|
|
elem = type_check_log_config(std::move(elem));
|
|
if(elem.empty()) {
|
|
// TODO: handle error
|
|
}
|
|
process_one(elem);
|
|
}
|
|
} else if (data.is_object()) {
|
|
data = type_check_log_config(std::move(data));
|
|
if(data.empty()) {
|
|
// TODO: handle error
|
|
}
|
|
process_one(data);
|
|
} else {
|
|
// TODO: handle error
|
|
}
|
|
}
|
|
|