diff --git a/CMakeLists.txt b/CMakeLists.txt index 50ef474e..b917e15b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -91,7 +91,8 @@ endif() if(REELAY_BUILD_APPS) message(STATUS "Building Reelay apps...") - add_subdirectory(apps) + add_subdirectory(apps/rybinx) + add_subdirectory(apps/ryjson1) endif() if(REELAY_BUILD_PYTHON_BINDINGS) diff --git a/Makefile b/Makefile index 1d0d4636..2481ec92 100644 --- a/Makefile +++ b/Makefile @@ -18,7 +18,7 @@ test: apps: cmake -S $(WORKSPACE) -B $(BUILD_DIRECTORY) -DREELAY_BUILD_APPS=ON - cmake --build $(BUILD_DIRECTORY) --target ryjson + cmake --build $(BUILD_DIRECTORY) cmake --install $(BUILD_DIRECTORY) install: build @@ -27,5 +27,13 @@ install: build clean: rm -rf $(BUILD_DIRECTORY) +benchmark: + pip install git+https://github.com/doganulus/timescales.git --break-system-packages + timescales-generate-large --failing-end --output-dir /tmp/benchmarks/data + +benchmark-binary: benchmark + git clone https://github.com/doganulus/timescales.git /tmp/timescales + cd /tmp/benchmarks/data && python /tmp/timescales/scripts/to_binary_row.py + docs: mkdocs serve diff --git a/apps/CMakeLists.txt b/apps/CMakeLists.txt deleted file mode 100644 index a917b472..00000000 --- a/apps/CMakeLists.txt +++ /dev/null @@ -1 +0,0 @@ -add_subdirectory(ryjson1) diff --git a/apps/rybinx/CMakeLists.txt b/apps/rybinx/CMakeLists.txt new file mode 100644 index 00000000..8dbaf643 --- /dev/null +++ b/apps/rybinx/CMakeLists.txt @@ -0,0 +1,6 @@ +add_executable(rybinx) + +target_sources(rybinx PRIVATE "main.cpp") +target_link_libraries(rybinx PRIVATE reelay::reelay) + +install(TARGETS rybinx) diff --git a/apps/rybinx/main.cpp b/apps/rybinx/main.cpp new file mode 100644 index 00000000..b48f54ce --- /dev/null +++ b/apps/rybinx/main.cpp @@ -0,0 +1,266 @@ +#include +#include +#include +#include + +#include +#include +#include + +// argp option keys +enum RYBINX_OPTS : uint8_t { OPT_DENSE = 'v', OPT_DISCRETE = 'x' }; + +const char* argp_program_version = "rybinx 0.1.0"; +const char* argp_program_bug_address = "Dogan Ulus "; +static const char* doc = "Reelay on Timescales Raw Binary Format"; +static const char* args_doc = "SPEC FILE"; + +struct arguments { + char* spec; + char* file; + bool dense = false; + bool discrete = false; +}; + +static std::array options = { + {{"dense", OPT_DENSE, nullptr, 0, "Use dense time model (default)", 0}, + {"discrete", OPT_DISCRETE, nullptr, 0, "Use discrete time model", 0}, + {nullptr}}}; + +static error_t parse_opt(int key, char* arg, struct argp_state* state) +{ + auto* arguments = (struct arguments*)state->input; + switch(key) { + case OPT_DENSE: + arguments->dense = true; + break; + case OPT_DISCRETE: + arguments->discrete = true; + break; + case ARGP_KEY_ARG: + if(state->arg_num == 0) { + arguments->spec = arg; + } + else { + arguments->file = arg; + } + break; + case ARGP_KEY_END: + if(state->arg_num < 2) { + argp_usage(state); + } + break; + default: + return ARGP_ERR_UNKNOWN; + } + return 0; +} +static struct argp argp = {options.data(), parse_opt, args_doc, doc}; + +#pragma pack(push, 1) +struct tx_binary_input_t { + int32_t time; + bool p; + bool q; + bool r; + bool s; +}; +#pragma pack(pop) +static_assert(std::is_trivially_copyable_v); + +template<> +struct reelay::timefield { + using input_t = tx_binary_input_t; + inline static int32_t get_time(const input_t& msg) + { + return msg.time; + } +}; + +template<> +struct reelay::datafield { + using input_t = tx_binary_input_t; + + inline static input_t at( + const input_t& /*container*/, const std::string& /*key*/) + { + return tx_binary_input_t{}; + } + + inline static input_t at(const input_t& /*container*/, std::size_t /*index*/) + { + return tx_binary_input_t{}; + } + + inline static bool contains( + const input_t& /*container*/, const std::string& /*key*/) + { + return true; + } + + inline static bool as_bool(const input_t& msg, const std::string& key) + { + bool result = false; + if(key == "p") { + result = msg.p; + } + else if(key == "q") { + result = msg.q; + } + else if(key == "r") { + result = msg.r; + } + else if(key == "s") { + result = msg.s; + } + else { + throw std::out_of_range("Key not found"); + } + return result; + } + + inline static int64_t as_integer( + const input_t& /*container*/, const std::string& /*key*/) + { + return 0; + } + + inline static double as_floating( + const input_t& /*container*/, const std::string& /*key*/) + { + return 0.0; + } + + inline static std::string as_string( + const input_t& /*container*/, const std::string& /*key*/) + { + return std::string{}; + } + + inline static bool contains( + const input_t& /*container*/, std::size_t /*index*/) + { + return true; + } + + inline static bool as_bool( + const input_t& /*container*/, std::size_t /*index*/) + { + return false; + } + + inline static int as_integer( + const input_t& /*container*/, std::size_t /*index*/) + { + return 0; + } + + inline static double as_floating( + const input_t& /*container*/, std::size_t /*index*/) + { + return 0.0; + } + + inline static std::string as_string( + const input_t& /*container*/, std::size_t /*index*/) + { + return std::string{}; + } +}; + +static constexpr size_t buffer_size = + static_cast(64 * 1024); // 64 KiB + +int main(int argc, char** argv) +{ + struct arguments arguments; + argp_parse(&argp, argc, argv, 0, nullptr, &arguments); + + using input_t = tx_binary_input_t; + using output_t = reelay::json; + + // Choices + bool use_discrete = arguments.discrete; + bool use_dense = arguments.dense; + + // Apply defaults + if(!use_discrete && !use_dense) { + use_discrete = true; + } + + auto monitor = reelay::monitor(); + auto discrete_opts = + reelay::discrete_timed::monitor::options() + .with_condensing(true); + auto dense_opts = + reelay::dense_timed::monitor::options(); + + if(use_dense) { + monitor = reelay::make_monitor(arguments.spec, dense_opts); + } + else { + monitor = reelay::make_monitor(arguments.spec, discrete_opts); + } + + std::string filename = arguments.file; + std::ifstream input(filename, std::ios::binary); + if(!input) { + std::cerr << "Error opening file: " << filename << std::endl; + return 1; + } + std::ofstream output(filename + ".ryl"); + if(!output) { + std::cerr << "Error creating output" << std::endl; + return 1; + } + std::cout << "Processing " << filename << std::endl; + std::cout << "---" << std::endl; + + uint32_t total_lines = 0; + input.read(reinterpret_cast(&total_lines), sizeof(total_lines)); + if(input.gcount() != sizeof(total_lines)) { + std::cerr << "File too small or corrupted." << std::endl; + return 1; + } + + using record_t = tx_binary_input_t; + static_assert( + std::is_trivially_copyable_v, + "record_t must be trivially copyable for binary I/O!"); + + const size_t record_size = sizeof(record_t); + + std::vector buffer(buffer_size); + std::vector records; + records.reserve(buffer_size / record_size); + + uint64_t count = 0; + uint64_t errn = 0; + + while(true) { + input.read(buffer.data(), static_cast(buffer.size())); + std::streamsize bytes_read = input.gcount(); + if(bytes_read <= 0) { + break; + } + + size_t num_records = bytes_read / record_size; + + const char* ptr = buffer.data(); + for(size_t i = 0; i < num_records; i++) { + record_t rec; + std::memcpy(&rec, ptr, record_size); + ptr += record_size; + + reelay::json result = monitor.update(rec); + if(not result.empty()) { + errn++; + std::cout << result << std::endl; + } + count++; + } + } + std::cout << "---" << std::endl; + + return 0; +} diff --git a/benchmarks/rybinx-benchmark.sh b/benchmarks/rybinx-benchmark.sh new file mode 100644 index 00000000..a06c7f3b --- /dev/null +++ b/benchmarks/rybinx-benchmark.sh @@ -0,0 +1,73 @@ +#!/usr/bin/env bash +set -xeuo pipefail + +commit_hash=$(git rev-parse HEAD) +echo "Running time series discrete benchmark for commit ${commit_hash}" + +RYBINX_FLAGS=${RYBINX_FLAGS:-"-x"} +TESTDATA_DIR="${TESTDATA_DIR:-$1}" + +hyperfine \ + --warmup 3 \ + --runs 25 \ + --export-json "${TESTDATA_DIR}/rybinx.${commit_hash}.dx.results.json" \ + --command-name AbsentAQ10 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:10]{q}) -> ((not{p}) since {q}))' ${TESTDATA_DIR}/AbsentAQ10.row.bin" \ + --command-name AbsentAQ100 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:100]{q}) -> ((not{p}) since {q}))' ${TESTDATA_DIR}/AbsentAQ100.row.bin" \ + --command-name AbsentAQ1000 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:1000]{q}) -> ((not{p}) since {q}))' ${TESTDATA_DIR}/AbsentAQ1000.row.bin" \ + --command-name AbsentBQR10 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} && !{q} && once{q}) -> ((not{p}) since[3:10] {q})' ${TESTDATA_DIR}/AbsentBQR10.row.bin" \ + --command-name AbsentBQR100 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} && !{q} && once{q}) -> ((not{p}) since[30:100] {q})' ${TESTDATA_DIR}/AbsentBQR100.row.bin" \ + --command-name AbsentBQR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} && !{q} && once{q}) -> ((not{p}) since[300:1000] {q})' ${TESTDATA_DIR}/AbsentBQR1000.row.bin" \ + --command-name AbsentBR10 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:10](not{p})))' ${TESTDATA_DIR}/AbsentBR10.row.bin" \ + --command-name AbsentBR100 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:100](not{p})))' ${TESTDATA_DIR}/AbsentBR100.row.bin" \ + --command-name AbsentBR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:1000](not{p})))' ${TESTDATA_DIR}/AbsentBR1000.row.bin" \ + --command-name AlwaysAQ10 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:10]{q}) -> ({p} since {q}))' ${TESTDATA_DIR}/AlwaysAQ10.row.bin" \ + --command-name AlwaysAQ100 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:100]{q}) -> ({p} since {q}))' ${TESTDATA_DIR}/AlwaysAQ100.row.bin" \ + --command-name AlwaysAQ1000 \ + "rybinx ${RYBINX_FLAGS} 'historically((once[:1000]{q}) -> ({p} since {q}))' ${TESTDATA_DIR}/AlwaysAQ1000.row.bin" \ + --command-name AlwaysBQR10 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ({p} since[3:10] {q}))' ${TESTDATA_DIR}/AlwaysBQR10.row.bin" \ + --command-name AlwaysBQR100 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ({p} since[30:100] {q}))' ${TESTDATA_DIR}/AlwaysBQR100.row.bin" \ + --command-name AlwaysBQR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ({p} since[300:1000] {q}))' ${TESTDATA_DIR}/AlwaysBQR1000.row.bin" \ + --command-name AlwaysBR10 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:10]{p}))' ${TESTDATA_DIR}/AlwaysBR10.row.bin" \ + --command-name AlwaysBR100 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:100]{p}))' ${TESTDATA_DIR}/AlwaysBR100.row.bin" \ + --command-name AlwaysBR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically({r} -> (historically[:1000]{p}))' ${TESTDATA_DIR}/AlwaysBR1000.row.bin" \ + --command-name RecurBQR10 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ((once[:10]({p} or {q})) since {q}))' ${TESTDATA_DIR}/RecurBQR10.row.bin" \ + --command-name RecurBQR100 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ((once[:100]({p} or {q})) since {q}))' ${TESTDATA_DIR}/RecurBQR100.row.bin" \ + --command-name RecurBQR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ((once[:1000]({p} or {q})) since {q}))' ${TESTDATA_DIR}/RecurBQR1000.row.bin" \ + --command-name RecurGLB10 \ + "rybinx ${RYBINX_FLAGS} 'historically(once[:10]{p})' ${TESTDATA_DIR}/RecurGLB10.row.bin" \ + --command-name RecurGLB100 \ + "rybinx ${RYBINX_FLAGS} 'historically(once[:100]{p})' ${TESTDATA_DIR}/RecurGLB100.row.bin" \ + --command-name RecurGLB1000 \ + "rybinx ${RYBINX_FLAGS} 'historically(once[:1000]{p})' ${TESTDATA_DIR}/RecurGLB1000.row.bin" \ + --command-name RespondBQR10 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ( (({s} -> once[3:10]{p}) and not((not {s}) since[10:] {p})) since {q}))' ${TESTDATA_DIR}/RespondBQR10.row.bin" \ + --command-name RespondBQR100 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ( (({s} -> once[30:100]{p}) and not((not {s}) since[100:] {p})) since {q}))' ${TESTDATA_DIR}/RespondBQR100.row.bin" \ + --command-name RespondBQR1000 \ + "rybinx ${RYBINX_FLAGS} 'historically(({r} && !{q} && once{q}) -> ( (({s} -> once[300:1000]{p}) and not((not {s}) since[1000:] {p})) since {q}))' ${TESTDATA_DIR}/RespondBQR1000.row.bin" \ + --command-name RespondGLB10 \ + "rybinx ${RYBINX_FLAGS} 'historically(({s} -> once[3:10]{p}) and not((not {s}) since[10:] {p}))' ${TESTDATA_DIR}/RespondGLB10.row.bin" \ + --command-name RespondGLB100 \ + "rybinx ${RYBINX_FLAGS} 'historically(({s} -> once[30:100]{p}) and not((not {s}) since[100:] {p}))' ${TESTDATA_DIR}/RespondGLB100.row.bin" \ + --command-name RespondGLB1000 \ + "rybinx ${RYBINX_FLAGS} 'historically(({s} -> once[300:1000]{p}) and not((not {s}) since[1000:] {p}))' ${TESTDATA_DIR}/RespondGLB1000.row.bin"