From 635d1a637cf79925588788320889473b165ec175 Mon Sep 17 00:00:00 2001 From: Michael Mueller Date: Thu, 24 Oct 2024 14:24:36 +0200 Subject: [PATCH] Networked version of B-link tree benchmark. --- repos/mml/src/app/blinktree_server/README.md | 79 +++++ .../src/app/blinktree_server/benchmark.cpp | 199 +++++++++++++ .../mml/src/app/blinktree_server/benchmark.h | 103 +++++++ repos/mml/src/app/blinktree_server/config.h | 17 ++ repos/mml/src/app/blinktree_server/listener.h | 15 + repos/mml/src/app/blinktree_server/main.cpp | 153 ++++++++++ .../src/app/blinktree_server/network/config.h | 9 + .../app/blinktree_server/network/server.cpp | 278 ++++++++++++++++++ .../src/app/blinktree_server/network/server.h | 90 ++++++ .../app/blinktree_server/request_scheduler.h | 252 ++++++++++++++++ repos/mml/src/app/blinktree_server/server.cpp | 32 ++ repos/mml/src/app/blinktree_server/server.h | 30 ++ repos/mml/src/app/blinktree_server/target.mk | 27 ++ 13 files changed, 1284 insertions(+) create mode 100644 repos/mml/src/app/blinktree_server/README.md create mode 100644 repos/mml/src/app/blinktree_server/benchmark.cpp create mode 100644 repos/mml/src/app/blinktree_server/benchmark.h create mode 100644 repos/mml/src/app/blinktree_server/config.h create mode 100644 repos/mml/src/app/blinktree_server/listener.h create mode 100644 repos/mml/src/app/blinktree_server/main.cpp create mode 100644 repos/mml/src/app/blinktree_server/network/config.h create mode 100644 repos/mml/src/app/blinktree_server/network/server.cpp create mode 100644 repos/mml/src/app/blinktree_server/network/server.h create mode 100644 repos/mml/src/app/blinktree_server/request_scheduler.h create mode 100644 repos/mml/src/app/blinktree_server/server.cpp create mode 100644 repos/mml/src/app/blinktree_server/server.h create mode 100644 repos/mml/src/app/blinktree_server/target.mk diff --git a/repos/mml/src/app/blinktree_server/README.md b/repos/mml/src/app/blinktree_server/README.md new file mode 100644 index 0000000000..06f1770410 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/README.md @@ -0,0 +1,79 @@ +# BLinkTree Benchmark +The BLinkTree-benchmark stores `8` byte numeric keys and values. +Call `./bin/blinktree_benchmark -h` for help and parameters. + +## How to generate YCSB workload +* Workload specifications are done by files in `workloads_specification/`. +* Call `make ycsb-a` and `make ycsb-c` to generate workloads **A** and **C**. +* Workload files are stored in `workloads/` +* Use `./bin/blinktree_benchmark -f ` to pass the desired workload. +* Default (if not specified) is `-f workloads/fill_randint_workloada workloads/mixed_randint_workloada`. + +## Important CLI arguments +* The first argument is the number of cores: + * `./bin/blinktree_benchmark 1` for using a single core. + * `./bin/blinktree_benchmark 1:24` for using cores `1` up to `24`. +* `-i ` specifies the number of repetitions of each workload. +* `-s ` steps of the cores: + * `-s 1` will increase the used cores by one (core ids: `0,1,2,3,4,5,6,7,..,23`). + * `-s 2` will skip every second core (core ids: `0,1,3,5,7,..23`). +* `-pd ` specifies the prefetch distance. +* `-p` or `--perf` will activate performance counter (result will be printed to console and output file). +* `--latched` will enable latches for synchronization (default off). +* `--exclusive` forces the tasks to access tree nodes exclusively (e.g. by using spinlocks or core-based sequencing) (default off). +* `--sync4me` will use built-in synchronization selection to choose the matching primitive based on annotations. +* `-o ` will write the results in **json** format to the given file. + +## Understanding the output +After started, the benchmark will print a summary of configured cores and workload: + + core configuration: + 1: 0 + 2: 0 1 + 4: 0 1 2 3 + workload: fill: 5m / readonly: 5m + +Here, we configured the benchmark to use one to four cores; each line of the core configuration displays the number of cores and the core identifiers. + +Following, the benchmark will be started and print the results for every iteration: + + 1 1 0 1478 ms 3.38295e+06 op/s + 1 1 1 1237 ms 4.04204e+06 op/s + 2 1 0 964 ms 5.18672e+06 op/s + 2 1 1 675 ms 7.40741e+06 op/s + 4 1 0 935 ms 5.34759e+06 op/s + 4 1 1 532 ms 9.3985e+06 op/s + +* The first column is the number of used cores. +* The second column displays the iteration of the benchmark (configured by `-i X`). +* Thirdly, the phase-identifier will be printed: `0` for initialization phase (which will be only inserts) and `1` for the workload phase (which is read-only here). +* After that, the time and throughput are written. +* If `--perf` is enabled, the output will be extended by some perf counters, which are labeled (like throughput). + +## Plot the results +When using `-o FILE`, the results will be written to the given file, using `JSON` format. +The plot script `scripts/plot_blinktree_benchmark INPUT_FILE [INPUT_FILE ...]` will aggregate and plot the results using one or more of those `JSON` files. + +## Examples + +###### Running workload A using optimistic synchronization + + ./bin/blinktree_benchmark 1: -s 2 -i 3 -pd 3 -p -f workloads/fill_randint_workloada workloads/mixed_randint_workloada -o optimistic.json + +###### Running workload A using best matching synchronization + + ./bin/blinktree_benchmark 1: -s 2 -i 3 -pd 3 -p --sync4me -f workloads/fill_randint_workloada workloads/mixed_randint_workloada -o sync4me.json + +###### Running workload A using reader/writer-locks + + ./bin/blinktree_benchmark 1: -s 2 -i 3 -pd 3 -p --latched -f workloads/fill_randint_workloada workloads/mixed_randint_workloada -o rwlocked.json + +###### Running workload A using core-based sequencing + + ./bin/blinktree_benchmark 1: -s 2 -i 3 -pd 3 -p --exclusive -f workloads/fill_randint_workloada workloads/mixed_randint_workloada -o core-sequenced.json + +###### Running workload A using spin-locks + + ./bin/blinktree_benchmark 1: -s 2 -i 3 -pd 3 -p --latched --exclusive -f workloads/fill_randint_workloada workloads/mixed_randint_workloada -o spinlocked.json + + \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/benchmark.cpp b/repos/mml/src/app/blinktree_server/benchmark.cpp new file mode 100644 index 0000000000..24117ba79f --- /dev/null +++ b/repos/mml/src/app/blinktree_server/benchmark.cpp @@ -0,0 +1,199 @@ +#include "benchmark.h" +#include +#include +#include +#include +#include + +using namespace application::blinktree_benchmark; + +Benchmark::Benchmark(benchmark::Cores &&cores, const std::uint16_t iterations, std::string &&fill_workload_file, + std::string &&mixed_workload_file, const bool use_performance_counter, + const mx::synchronization::isolation_level node_isolation_level, + const mx::synchronization::protocol preferred_synchronization_method, + const bool print_tree_statistics, const bool check_tree, std::string &&result_file_name, + std::string &&statistic_file_name, std::string &&tree_file_name, const bool profile) + : _cores(std::move(cores)), _iterations(iterations), _node_isolation_level(node_isolation_level), + _preferred_synchronization_method(preferred_synchronization_method), + _print_tree_statistics(print_tree_statistics), _check_tree(check_tree), + _result_file_name(std::move(result_file_name)), _statistic_file_name(std::move(statistic_file_name)), + _tree_file_name(std::move(tree_file_name)), _profile(profile) +{ + if (use_performance_counter) + { + this->_chronometer.add(benchmark::Perf::CYCLES); + this->_chronometer.add(benchmark::Perf::INSTRUCTIONS); + this->_chronometer.add(benchmark::Perf::STALLS_MEM_ANY); + this->_chronometer.add(benchmark::Perf::SW_PREFETCH_ACCESS_NTA); + this->_chronometer.add(benchmark::Perf::SW_PREFETCH_ACCESS_WRITE); + } + + std::cout << "core configuration: \n" << this->_cores.dump(2) << std::endl; + + this->_workload.build(fill_workload_file, mixed_workload_file); + if (this->_workload.empty(benchmark::phase::FILL) && this->_workload.empty(benchmark::phase::MIXED)) + { + std::exit(1); + } + + std::cout << "workload: " << this->_workload << "\n" << std::endl; +} + +void Benchmark::start() +{ + // Reset tree. + if (this->_tree == nullptr) + { + this->_tree = std::make_unique>( + this->_node_isolation_level, this->_preferred_synchronization_method); + } + + // Reset request scheduler. + if (this->_request_scheduler.empty() == false) + { + this->_request_scheduler.clear(); + } + + // Create one request scheduler per core. + for (auto core_index = 0U; core_index < this->_cores.current().size(); core_index++) + { + const auto channel_id = core_index; + auto *request_scheduler = mx::tasking::runtime::new_task( + 0U, core_index, channel_id, this->_workload, this->_cores.current(), this->_tree.get(), this); + mx::tasking::runtime::spawn(*request_scheduler, 0U); + this->_request_scheduler.push_back(request_scheduler); + } + this->_open_requests = this->_request_scheduler.size(); + + // Start measurement. + if (this->_profile) + { + mx::tasking::runtime::profile(this->profile_file_name()); + } + this->_chronometer.start(static_cast(static_cast(this->_workload)), + this->_current_iteration + 1, this->_cores.current()); +} + +const mx::util::core_set &Benchmark::core_set() +{ + if (this->_current_iteration == std::numeric_limits::max()) + { + // This is the very first time we start the benchmark. + this->_current_iteration = 0U; + return this->_cores.next(); + } + + // Switch from fill to mixed phase. + if (this->_workload == benchmark::phase::FILL && this->_workload.empty(benchmark::phase::MIXED) == false) + { + this->_workload.reset(benchmark::phase::MIXED); + return this->_cores.current(); + } + this->_workload.reset(benchmark::phase::FILL); + + // Run the next iteration. + if (++this->_current_iteration < this->_iterations) + { + return this->_cores.current(); + } + this->_current_iteration = 0U; + + // At this point, all phases and all iterations for the current core configuration + // are done. Increase the cores. + return this->_cores.next(); +} + +void Benchmark::requests_finished() +{ + const auto open_requests = --this->_open_requests; + + if (open_requests == 0U) // All request schedulers are done. + { + // Stop and print time (and performance counter). + const auto result = this->_chronometer.stop(this->_workload.size()); + mx::tasking::runtime::stop(); + std::cout << result << std::endl; + + // Dump results to file. + if (this->_result_file_name.empty() == false) + { + std::ofstream result_file_stream(this->_result_file_name, std::ofstream::app); + result_file_stream << result.to_json().dump() << std::endl; + } + + // Dump statistics to file. + if constexpr (mx::tasking::config::task_statistics()) + { + if (this->_statistic_file_name.empty() == false) + { + std::ofstream statistic_file_stream(this->_statistic_file_name, std::ofstream::app); + nlohmann::json statistic_json; + statistic_json["iteration"] = result.iteration(); + statistic_json["cores"] = result.core_count(); + statistic_json["phase"] = result.phase(); + statistic_json["scheduled"] = nlohmann::json(); + statistic_json["scheduled-on-channel"] = nlohmann::json(); + statistic_json["scheduled-off-channel"] = nlohmann::json(); + statistic_json["executed"] = nlohmann::json(); + statistic_json["executed-reader"] = nlohmann::json(); + statistic_json["executed-writer"] = nlohmann::json(); + statistic_json["buffer-fills"] = nlohmann::json(); + for (auto i = 0U; i < this->_cores.current().size(); i++) + { + const auto core_id = std::int32_t{this->_cores.current()[i]}; + const auto core_id_string = std::to_string(core_id); + statistic_json["scheduled"][core_id_string] = + result.scheduled_tasks(core_id) / double(result.operation_count()); + statistic_json["scheduled-on-core"][core_id_string] = + result.scheduled_tasks_on_core(core_id) / double(result.operation_count()); + statistic_json["scheduled-off-core"][core_id_string] = + result.scheduled_tasks_off_core(core_id) / double(result.operation_count()); + statistic_json["executed"][core_id_string] = + result.executed_tasks(core_id) / double(result.operation_count()); + statistic_json["executed-reader"][core_id_string] = + result.executed_reader_tasks(core_id) / double(result.operation_count()); + statistic_json["executed-writer"][core_id_string] = + result.executed_writer_tasks(core_id) / double(result.operation_count()); + statistic_json["fill"][core_id_string] = + result.worker_fills(core_id) / double(result.operation_count()); + } + + statistic_file_stream << statistic_json.dump(2) << std::endl; + } + } + + // Check and print the tree. + if (this->_check_tree) + { + this->_tree->check(); + } + + if (this->_print_tree_statistics) + { + this->_tree->print_statistics(); + } + + const auto is_last_phase = + this->_workload == benchmark::phase::MIXED || this->_workload.empty(benchmark::phase::MIXED); + + // Dump the tree. + if (this->_tree_file_name.empty() == false && is_last_phase) + { + std::ofstream tree_file_stream(this->_tree_file_name); + tree_file_stream << static_cast(*(this->_tree)).dump() << std::endl; + } + + // Delete the tree to free the hole memory. + if (is_last_phase) + { + this->_tree.reset(nullptr); + } + } +} + +std::string Benchmark::profile_file_name() const +{ + return "profiling-" + std::to_string(this->_cores.current().size()) + "-cores" + "-phase-" + + std::to_string(static_cast(static_cast(this->_workload))) + "-iteration-" + + std::to_string(this->_current_iteration) + ".json"; +} \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/benchmark.h b/repos/mml/src/app/blinktree_server/benchmark.h new file mode 100644 index 0000000000..ae0b789b4b --- /dev/null +++ b/repos/mml/src/app/blinktree_server/benchmark.h @@ -0,0 +1,103 @@ +#pragma once + +#include "listener.h" +#include "request_scheduler.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace application::blinktree_benchmark { +/** + * Benchmark executing the task-based BLink-Tree. + */ +class Benchmark final : public Listener +{ +public: + Benchmark(benchmark::Cores &&, std::uint16_t iterations, std::string &&fill_workload_file, + std::string &&mixed_workload_file, bool use_performance_counter, + mx::synchronization::isolation_level node_isolation_level, + mx::synchronization::protocol preferred_synchronization_method, bool print_tree_statistics, + bool check_tree, std::string &&result_file_name, std::string &&statistic_file_name, + std::string &&tree_file_name, bool profile); + + ~Benchmark() noexcept override = default; + + /** + * @return Core set the benchmark should run in the current iteration. + */ + const mx::util::core_set &core_set(); + + /** + * Callback for request tasks to notify they are out of + * new requests. + */ + void requests_finished() override; + + /** + * Starts the benchmark after initialization. + */ + void start(); + +private: + // Collection of cores the benchmark should run on. + benchmark::Cores _cores; + + // Number of iterations the benchmark should use. + const std::uint16_t _iterations; + + // Current iteration within the actual core set. + std::uint16_t _current_iteration = std::numeric_limits::max(); + + // Workload to get requests from. + benchmark::Workload _workload; + + // Tree to run requests on. + std::unique_ptr> _tree; + + // The synchronization mechanism to use for tree nodes. + const mx::synchronization::isolation_level _node_isolation_level; + + // Preferred synchronization method. + const mx::synchronization::protocol _preferred_synchronization_method; + + // If true, the tree statistics (height, number of nodes, ...) will be printed. + const bool _print_tree_statistics; + + // If true, the tree will be checked for consistency after each iteration. + const bool _check_tree; + + // Name of the file to print results to. + const std::string _result_file_name; + + // Name of the file to print further statistics. + const std::string _statistic_file_name; + + // Name of the file to serialize the tree to. + const std::string _tree_file_name; + + // If true, use idle profiling. + const bool _profile; + + // Number of open request tasks; used for tracking the benchmark. + alignas(64) std::atomic_uint16_t _open_requests = 0; + + // List of request schedulers. + alignas(64) std::vector _request_scheduler; + + // Chronometer for starting/stopping time and performance counter. + alignas(64) benchmark::Chronometer _chronometer; + + /** + * @return Name of the file to write profiling results to. + */ + [[nodiscard]] std::string profile_file_name() const; +}; +} // namespace application::blinktree_benchmark \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/config.h b/repos/mml/src/app/blinktree_server/config.h new file mode 100644 index 0000000000..2144075be8 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/config.h @@ -0,0 +1,17 @@ +#pragma once + +namespace application::blinktree_benchmark { +class config +{ +public: + /** + * @return Number of requests that will be started at a time by the request scheduler. + */ + static constexpr auto batch_size() noexcept { return 500U; } + + /** + * @return Number of maximal open requests, system-wide. + */ + static constexpr auto max_parallel_requests() noexcept { return 1500U; } +}; +} // namespace application::blinktree_benchmark \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/listener.h b/repos/mml/src/app/blinktree_server/listener.h new file mode 100644 index 0000000000..5a911fa6b5 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/listener.h @@ -0,0 +1,15 @@ +#pragma once + +namespace application::blinktree_benchmark { +/** + * The listener will be used to notify the benchmark that request tasks are + * done and no more work is available. + */ +class Listener +{ +public: + constexpr Listener() = default; + virtual ~Listener() = default; + virtual void requests_finished() = 0; +}; +} // namespace application::blinktree_benchmark \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/main.cpp b/repos/mml/src/app/blinktree_server/main.cpp new file mode 100644 index 0000000000..e9e4046467 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/main.cpp @@ -0,0 +1,153 @@ +#include "benchmark.h" +#include +#include +#include +#include +#include +#include +#include "server.h" +#include +#include + +/* Genode includes */ +#include + +using namespace application::blinktree_server; + +/** + * Instantiates the BLink-Tree server with CLI arguments. + * @param count_arguments Number of CLI arguments. + * @param arguments Arguments itself. + * + * @return Instance of the server. + */ +std::tuple create_server(int count_arguments, char **arguments); + +/** + * Starts the server. + * + * @param count_arguments Number of CLI arguments. + * @param arguments Arguments itself. + * + * @return Return code of the application. + */ +int bt_main(int count_arguments, char **arguments) +{ + if (mx::system::Environment::is_numa_balancing_enabled()) + { + std::cout << "[Warn] NUMA balancing may be enabled, set '/proc/sys/kernel/numa_balancing' to '0'" << std::endl; + } + + auto [server, prefetch_distance, use_system_allocator] = create_server(count_arguments, arguments); + + if (server != nullptr) + { + /// Wait for the server to finish. + server->run(); + + delete server; + } + + return 0; +} + +std::tuple create_server(int count_arguments, char **arguments) +{ + // Set up arguments. + argparse::ArgumentParser argument_parser("blinktree_server"); + argument_parser.add_argument("cores") + .help("Number of cores to use.") + .default_value(std::uint16_t(1)) + .action([](const std::string &value) { return std::uint16_t(std::stoi(value)); }); + argument_parser.add_argument("--port") + .help("Port of the server") + .default_value(std::uint64_t(12345)) + .action([](const std::string &value) { return std::uint64_t(std::stoi(value)); }); + argument_parser.add_argument("-sco", "--system-core-order") + .help("Use systems core order. If not, cores are ordered by node id (should be preferred).") + .implicit_value(true) + .default_value(false); + argument_parser.add_argument("--exclusive") + .help("Are all node accesses exclusive?") + .implicit_value(true) + .default_value(false); + argument_parser.add_argument("--latched") + .help("Prefer latch for synchronization?") + .implicit_value(true) + .default_value(false); + argument_parser.add_argument("--olfit") + .help("Prefer OLFIT for synchronization?") + .implicit_value(true) + .default_value(false); + argument_parser.add_argument("--sync4me") + .help("Let the tasking layer decide the synchronization primitive.") + .implicit_value(true) + .default_value(false); + argument_parser.add_argument("-pd", "--prefetch-distance") + .help("Distance of prefetched data objects (0 = disable prefetching).") + .default_value(std::uint16_t(0)) + .action([](const std::string &value) { return std::uint16_t(std::stoi(value)); }); + argument_parser.add_argument("--system-allocator") + .help("Use the systems malloc interface to allocate tasks (default disabled).") + .implicit_value(true) + .default_value(false); + + // Parse arguments. + try + { + argument_parser.parse_args(count_arguments, arguments); + } + catch (std::runtime_error &e) + { + std::cout << argument_parser << std::endl; + return {nullptr, 0U, false}; + } + + auto order = + argument_parser.get("-sco") ? mx::util::core_set::Order::Ascending : mx::util::core_set::Order::NUMAAware; + auto cores = mx::util::core_set::build(argument_parser.get("cores")-1, order); + const auto isolation_level = argument_parser.get("--exclusive") + ? mx::synchronization::isolation_level::Exclusive + : mx::synchronization::isolation_level::ExclusiveWriter; + auto preferred_synchronization_method = mx::synchronization::protocol::Queue; + if (argument_parser.get("--latched")) + { + preferred_synchronization_method = mx::synchronization::protocol::Latch; + } + else if (argument_parser.get("--olfit")) + { + preferred_synchronization_method = mx::synchronization::protocol::OLFIT; + } + else if (argument_parser.get("--sync4me")) + { + preferred_synchronization_method = mx::synchronization::protocol::None; + } + + // Create the benchmark. + auto *server = new Server(argument_parser.get("--port"), std::move(cores), argument_parser.get("-pd"), isolation_level, preferred_synchronization_method); + + return {server, argument_parser.get("-pd"), argument_parser.get("--system-allocator")}; +} + +void Libc::Component::construct(Libc::Env &env) { + + mx::system::Environment::set_env(&env); + + auto sys_cores = mx::util::core_set::build(64); + mx::system::Environment::set_cores(&sys_cores); + + mx::memory::GlobalHeap::myself(); + std::uint16_t cores = 64; + //env.cpu().affinity_space().total(); + + char cores_arg[10]; + sprintf(cores_arg, "%d", cores); + + char *args[] = {"blinktree_server", "-pd", "3", cores_arg}; + + Libc::with_libc([&]() + { + std::cout << "Starting B-link tree server" << std::endl; + bt_main(4, args); + }); +} \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/network/config.h b/repos/mml/src/app/blinktree_server/network/config.h new file mode 100644 index 0000000000..026c610a96 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/network/config.h @@ -0,0 +1,9 @@ +#pragma once + +namespace application::blinktree_server::network { +class config +{ +public: + static constexpr auto max_connections() noexcept { return 64U; } +}; +} // namespace mx::io::network \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/network/server.cpp b/repos/mml/src/app/blinktree_server/network/server.cpp new file mode 100644 index 0000000000..40734f83ed --- /dev/null +++ b/repos/mml/src/app/blinktree_server/network/server.cpp @@ -0,0 +1,278 @@ +#include "server.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace application::blinktree_server::network; + +mx::tasking::TaskResult RequestTask::execute(const std::uint16_t core_id, const std::uint16_t channel_id) +{ + mx::tasking::TaskInterface* request_task; + + if (this->_type == Type::Insert) + { + request_task = mx::tasking::runtime::new_task< + db::index::blinktree::InsertValueTask>( + core_id, this->_key, this->_value, this->_tree, this->_response_handler); + + request_task->annotate(this->_tree->root(), db::index::blinktree::config::node_size() / 4U); + request_task->is_readonly(true); + } + else if (this->_type == Type::Lookup) + { + request_task = mx::tasking::runtime::new_task< + db::index::blinktree::LookupTask>( + core_id, this->_key, this->_response_handler); + + request_task->annotate(this->_tree->root(), db::index::blinktree::config::node_size() / 4U); + request_task->is_readonly(true); + } + else if(this->_type == Type::Update) + { + request_task = mx::tasking::runtime::new_task< + db::index::blinktree::UpdateTask>( + core_id, this->_key, this->_value, this->_response_handler); + + request_task->annotate(this->_tree->root(), db::index::blinktree::config::node_size() / 4U); + request_task->is_readonly(true); + } + else + { + this->_tree->check(); + this->_tree->print_statistics(); + return mx::tasking::TaskResult::make_null(); + } + + return mx::tasking::TaskResult::make_succeed(request_task); +} + +void ResponseHandler::inserted(const std::uint16_t /*core_id*/, const std::uint64_t key, const std::int64_t /*value*/) +{ + _server-> send(_client_id, std::to_string(key)); +} + +void ResponseHandler::updated(const std::uint16_t /*core_id*/, const std::uint64_t key, const std::int64_t /*value*/) +{ + _server-> send(_client_id, std::to_string(key)); +} + +void ResponseHandler::removed(const std::uint16_t /*core_id*/, const std::uint64_t key) +{ + _server-> send(_client_id, std::to_string(key)); +} + +void ResponseHandler::found(const std::uint16_t /*core_id*/, const std::uint64_t /*key*/, const std::int64_t value) +{ + _server-> send(_client_id, std::to_string(value)); +} + +void ResponseHandler::missing(const std::uint16_t /*core_id*/, const std::uint64_t key) +{ + _server-> send(_client_id, std::to_string(key)); +} + + + + +Server::Server(const std::uint64_t port, + const std::uint16_t count_channels) noexcept + : _port(port), _socket(-1), _client_sockets({0U}), + _count_channels(count_channels) +{ + this->_buffer.fill('\0'); + + this->_response_handlers = reinterpret_cast(std::malloc(sizeof(ResponseHandler) * config::max_connections())); + for (auto client_id = 0U; client_id < config::max_connections(); ++client_id) { + new (&this->_response_handlers[client_id]) ResponseHandler{this, client_id}; + } + + this->_request_tasks = reinterpret_cast(std::malloc( sizeof(RequestTask) * config::max_connections())); +} + +Server::~Server() { + ::free(this->_response_handlers); + ::free(this->_request_tasks); +} + +bool Server::listen(db::index::blinktree::BLinkTree* tree) +{ + this->_socket = socket(AF_INET, SOCK_STREAM, 0); + if (this->_socket == 0) + { + return false; + } + + auto option = std::int32_t{1}; + if (setsockopt(this->_socket, SOL_SOCKET, SO_REUSEADDR, &option, socklen_t{sizeof(std::int32_t)}) < 0) + { + return false; + } + + auto address = sockaddr_in{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = INADDR_ANY; + address.sin_port = htons(this->_port); + + if (bind(this->_socket, reinterpret_cast(&address), sizeof(sockaddr_in)) < 0) + { + return false; + } + + if (::listen(this->_socket, 1024) < 0) + { + return false; + } + + auto address_length = socklen_t{sizeof(sockaddr_in)}; + auto socket_descriptors = fd_set{}; + auto max_socket_descriptor = this->_socket; + auto client_socket = std::int32_t{-1}; + + while (this->_is_running) + { + FD_ZERO(&socket_descriptors); // NOLINT + FD_SET(this->_socket, &socket_descriptors); + + for (auto &socket_descriptor : this->_client_sockets) + { + if (socket_descriptor > 0) + { + FD_SET(socket_descriptor, &socket_descriptors); + } + + max_socket_descriptor = std::max(max_socket_descriptor, std::int32_t(socket_descriptor)); + } + + auto timeout = timeval{}; + timeout.tv_usec = 10000; + const auto count_ready_selectors = + select(max_socket_descriptor + 1, &socket_descriptors, nullptr, nullptr, &timeout); + + if (count_ready_selectors > 0) + { + if (FD_ISSET(this->_socket, &socket_descriptors)) + { + if ((client_socket = accept(this->_socket, reinterpret_cast(&address), &address_length)) < + 0) + { + return false; + } + this->add_client(client_socket); + } + + for (auto i = 0U; i < this->_client_sockets.size(); ++i) + { + const auto client = this->_client_sockets[i]; + if (FD_ISSET(client, &socket_descriptors)) + { + const auto read_bytes = read(client, this->_buffer.data(), this->_buffer.size()); + if (read_bytes == 0U) + { + ::close(client); + this->_client_sockets[i] = 0U; + } + else + { + // Copy incoming data locally. + RequestTask::Type request_type; + auto message = std::string(this->_buffer.data(), read_bytes); + + if (message[0] == 'D') + { + auto *request_task = new (&this->_request_tasks[i]) RequestTask{tree, this->_response_handlers[i]}; + request_task->annotate(std::uint16_t(0U)); + mx::tasking::runtime::spawn(*request_task); + } + else + { + switch(message[0]) + { + case 'I': request_type = RequestTask::Type::Insert; break; + case 'U': request_type = RequestTask::Type::Update; break; + default: request_type = RequestTask::Type::Lookup; + } + + auto key = 0ULL; + auto index = 2U; // Skip request type and comma. + while(message[index] >= '0' && message[index] <= '9') { + key = key * 10 + (message[index++] - '0'); + } + + auto channel_id = std::uint16_t(this->_next_worker_id.fetch_add(1U) % this->_count_channels); + if (request_type == RequestTask::Type::Insert || request_type == RequestTask::Type::Lookup) + { + auto value = 0LL; + ++index; + while (message[index] >= '0' && message[index] <= '9') + { + value = value * 10 + (message[index++] - '0'); + } + + auto *request_task = new (&this->_request_tasks[i]) RequestTask{tree, request_type, key, value, this->_response_handlers[i]}; + request_task->annotate(channel_id); + mx::tasking::runtime::spawn(*request_task); + } + else + { + auto *request_task = new (&this->_request_tasks[i]) RequestTask{tree, RequestTask::Type::Lookup, key, this->_response_handlers[i]}; + request_task->annotate(channel_id); + mx::tasking::runtime::spawn(*request_task); + } + //mx::tasking::runtime::scheduler().allocate_cores(64); + } + } + } + } + } + } + + for (const auto client : this->_client_sockets) + { + if (client > 0) + { + ::close(client); + } + } + ::close(this->_socket); + + return true; +} + +void Server::send(const std::uint32_t client_id, std::string &&message) +{ + const auto length = std::uint64_t(message.size()); + auto response = std::string(length + sizeof(length), '\0'); + + // Write header + std::memcpy(response.data(), static_cast(&length), sizeof(length)); + + // Write data + std::memmove(response.data() + sizeof(length), message.data(), length); + + ::send(this->_client_sockets[client_id], response.c_str(), response.length(), 0); +} + +std::uint16_t Server::add_client(const std::int32_t client_socket) +{ + for (auto i = 0U; i < this->_client_sockets.size(); ++i) + { + if (this->_client_sockets[i] == 0U) + { + this->_client_sockets[i] = client_socket; + return i; + } + } + + return std::numeric_limits::max(); +} + +void Server::stop() noexcept +{ + this->_is_running = false; +} \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/network/server.h b/repos/mml/src/app/blinktree_server/network/server.h new file mode 100644 index 0000000000..1a754d293f --- /dev/null +++ b/repos/mml/src/app/blinktree_server/network/server.h @@ -0,0 +1,90 @@ +#pragma once + +#include "config.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace application::blinktree_server::network { + +class Server; +class alignas(64) ResponseHandler final : public db::index::blinktree::Listener +{ +public: + ResponseHandler(Server* server, const std::uint32_t client_id) : _server(server), _client_id(client_id) { } + ResponseHandler(ResponseHandler&&) noexcept = default; + ~ResponseHandler() = default; + + void inserted(std::uint16_t core_id, const std::uint64_t key, const std::int64_t value) override; + void updated(std::uint16_t core_id, const std::uint64_t key, const std::int64_t value) override; + void removed(std::uint16_t core_id, const std::uint64_t key) override; + void found(std::uint16_t core_id, const std::uint64_t key, const std::int64_t value) override; + void missing(std::uint16_t core_id, const std::uint64_t key) override; + +private: + Server* _server; + std::uint32_t _client_id; +}; + +class alignas(64) RequestTask final : public mx::tasking::TaskInterface +{ +public: + enum Type { Insert, Update, Lookup, Debug }; + + RequestTask(db::index::blinktree::BLinkTree* tree, const Type type, const std::uint64_t key, ResponseHandler& response_handler) noexcept + : _tree(tree), _type(type), _key(key), _response_handler(response_handler) { } + RequestTask(db::index::blinktree::BLinkTree* tree, const Type type, const std::uint64_t key, const std::int64_t value, ResponseHandler& response_handler) noexcept + : _tree(tree), _type(type), _key(key), _value(value), _response_handler(response_handler) { } + RequestTask(db::index::blinktree::BLinkTree* tree, ResponseHandler& response_handler) noexcept + : _tree(tree), _type(Type::Debug), _response_handler(response_handler) { } + ~RequestTask() noexcept = default; + + mx::tasking::TaskResult execute(std::uint16_t core_id, std::uint16_t channel_id) override; + +private: + db::index::blinktree::BLinkTree* _tree; + Type _type; + std::uint64_t _key; + std::uint64_t _value; + ResponseHandler& _response_handler; +}; + +class Server +{ +public: + Server(std::uint64_t port, + std::uint16_t count_channels) noexcept; + ~Server(); + + [[nodiscard]] std::uint16_t port() const noexcept { return _port; } + void stop() noexcept; + void send(std::uint32_t client_id, std::string &&message); + bool listen(db::index::blinktree::BLinkTree* tree); + + [[nodiscard]] bool is_running() const noexcept { return _is_running; } + +private: + const std::uint64_t _port; + std::int32_t _socket; + std::array _client_sockets; + std::array _buffer; + + ResponseHandler* _response_handlers; + RequestTask *_request_tasks; + + alignas(64) bool _is_running = true; + alignas(64) std::atomic_uint64_t _next_worker_id{0U}; + const std::uint16_t _count_channels; + + std::uint16_t add_client(std::int32_t client_socket); +}; +} // namespace mx::io::network \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/request_scheduler.h b/repos/mml/src/app/blinktree_server/request_scheduler.h new file mode 100644 index 0000000000..677a29f4d0 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/request_scheduler.h @@ -0,0 +1,252 @@ +#pragma once + +#include "config.h" +#include "listener.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace application::blinktree_benchmark { + +class RequestIndex +{ +public: + static RequestIndex make_finished() { return RequestIndex{std::numeric_limits::max(), 0UL}; } + static RequestIndex make_no_new() { return RequestIndex{0UL, 0UL}; } + + RequestIndex(const std::uint64_t index, const std::uint64_t count) noexcept : _index(index), _count(count) {} + explicit RequestIndex(std::pair &&index_and_count) noexcept + : _index(std::get<0>(index_and_count)), _count(std::get<1>(index_and_count)) + { + } + RequestIndex(RequestIndex &&) noexcept = default; + RequestIndex(const RequestIndex &) = default; + ~RequestIndex() noexcept = default; + + RequestIndex &operator=(RequestIndex &&) noexcept = default; + + [[nodiscard]] std::uint64_t index() const noexcept { return _index; } + [[nodiscard]] std::uint64_t count() const noexcept { return _count; } + + [[nodiscard]] bool is_finished() const noexcept { return _index == std::numeric_limits::max(); } + [[nodiscard]] bool has_new() const noexcept { return _count > 0UL; } + + RequestIndex &operator-=(const std::uint64_t count) noexcept + { + _count -= count; + _index += count; + return *this; + } + +private: + std::uint64_t _index; + std::uint64_t _count; +}; + +/** + * The RequestContainer manages the workload and allocates new batches of requests + * that will be scheduled by the request scheduler. + */ +class RequestContainer +{ +public: + RequestContainer(const std::uint16_t core_id, const std::uint64_t max_open_requests, + benchmark::Workload &workload) noexcept + : _finished_requests(core_id), _local_buffer(workload.next(config::batch_size())), + _max_pending_requests(max_open_requests), _workload(workload) + { + } + + ~RequestContainer() noexcept = default; + + /** + * Allocates the next requests to spawn. + * + * @return Pair of workload-index and number of tuples to request. + * When the number is negative, no more requests are available. + */ + RequestIndex next() noexcept + { + const auto finished_requests = _finished_requests.load(); + + const auto pending_requests = _scheduled_requests - finished_requests; + if (pending_requests >= _max_pending_requests) + { + // Too many open requests somewhere in the system. + return RequestIndex::make_no_new(); + } + + if (_local_buffer.has_new() == false) + { + _local_buffer = RequestIndex{_workload.next(config::batch_size())}; + } + + if (_local_buffer.has_new()) + { + // How many requests can be scheduled without reaching the request limit? + const auto free_requests = _max_pending_requests - pending_requests; + + // Try to spawn all free requests, but at least those in the local buffer. + const auto count = std::min(free_requests, _local_buffer.count()); + + _scheduled_requests += count; + + const auto index = RequestIndex{_local_buffer.index(), count}; + _local_buffer -= count; + + return index; + } + + // Do we have to wait for pending requests or are we finished? + return pending_requests > 0UL ? RequestIndex::make_no_new() : RequestIndex::make_finished(); + } + + /** + * Callback after inserted a value. + */ + void inserted(const std::uint16_t core_id, const std::uint64_t /*key*/, const std::int64_t /*value*/) noexcept + { + task_finished(core_id); + } + + /** + * Callback after updated a value. + */ + void updated(const std::uint16_t core_id, const std::uint64_t /*key*/, const std::int64_t /*value*/) noexcept + { + task_finished(core_id); + } + + /** + * Callback after removed a value. + */ + void removed(const std::uint16_t core_id, const std::uint64_t /*key*/) noexcept { task_finished(core_id); } + + /** + * Callback after found a value. + */ + void found(const std::uint16_t core_id, const std::uint64_t /*key*/, const std::int64_t /*value*/) noexcept + { + task_finished(core_id); + } + + /** + * Callback on missing a value. + */ + void missing(const std::uint16_t core_id, const std::uint64_t /*key*/) noexcept { task_finished(core_id); } + + const benchmark::NumericTuple &operator[](const std::size_t index) const noexcept { return _workload[index]; } + +private: + // Number of requests finished by tasks. + mx::util::reference_counter_64 _finished_requests; + + // Number of tasks scheduled by the owning request scheduler. + std::uint64_t _scheduled_requests = 0UL; + + // Local buffer holding not scheduled, but from global worker owned request items. + RequestIndex _local_buffer; + + // Number of requests that can be distributed by this scheduler, + // due to system-wide maximal parallel requests. + const std::uint64_t _max_pending_requests; + + // Workload to get requests from. + benchmark::Workload &_workload; + + /** + * Updates the counter of finished requests. + */ + void task_finished(const std::uint16_t core_id) { _finished_requests.add(core_id); } +}; + +/** + * The RequestScheduler own its own request container and sets up requests for the BLink-Tree. + */ +class RequestSchedulerTask final : public mx::tasking::TaskInterface +{ +public: + RequestSchedulerTask(const std::uint16_t core_id, const std::uint16_t channel_id, benchmark::Workload &workload, + const mx::util::core_set &core_set, + db::index::blinktree::BLinkTree *tree, Listener *listener) + : _tree(tree), _listener(listener) + { + this->annotate(mx::tasking::priority::low); + this->is_readonly(false); + + const auto container = mx::tasking::runtime::new_resource( + sizeof(RequestContainer), mx::resource::hint{channel_id}, core_id, + config::max_parallel_requests() / core_set.size(), workload); + this->annotate(container, sizeof(RequestContainer)); + } + + ~RequestSchedulerTask() final = default; + + mx::tasking::TaskResult execute(const std::uint16_t core_id, const std::uint16_t channel_id) override + { + // Get some new requests from the container. + auto &request_container = *mx::resource::ptr_cast(this->annotated_resource()); + const auto next_requests = request_container.next(); + + if (next_requests.has_new()) + { + for (auto i = next_requests.index(); i < next_requests.index() + next_requests.count(); ++i) + { + mx::tasking::TaskInterface *task{nullptr}; + const auto &tuple = request_container[i]; + if (tuple == benchmark::NumericTuple::INSERT) + { + task = mx::tasking::runtime::new_task< + db::index::blinktree::InsertValueTask>( + core_id, tuple.key(), tuple.value(), _tree, request_container); + task->is_readonly(_tree->height() > 1U); + } + else if (tuple == benchmark::NumericTuple::LOOKUP) + { + task = mx::tasking::runtime::new_task< + db::index::blinktree::LookupTask>( + core_id, tuple.key(), request_container); + + task->is_readonly(true); + } + else if (tuple == benchmark::NumericTuple::UPDATE) + { + task = mx::tasking::runtime::new_task< + db::index::blinktree::UpdateTask>( + core_id, tuple.key(), tuple.value(), request_container); + task->is_readonly(_tree->height() > 1U); + } + + task->annotate(_tree->root(), db::index::blinktree::config::node_size() / 4U); + mx::tasking::runtime::spawn(*task, channel_id); + } + } + else if (next_requests.is_finished()) + { + // All requests are done. Notify the benchmark and die. + _listener->requests_finished(); + mx::tasking::runtime::delete_resource(this->annotated_resource()); + return mx::tasking::TaskResult::make_remove(); + } + + return mx::tasking::TaskResult::make_succeed(this); + } + +private: + // The tree to send requests to. + db::index::blinktree::BLinkTree *_tree; + + // Benchmark listener to notify on requests are done. + Listener *_listener; +}; +} // namespace application::blinktree_benchmark \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/server.cpp b/repos/mml/src/app/blinktree_server/server.cpp new file mode 100644 index 0000000000..bc279484fc --- /dev/null +++ b/repos/mml/src/app/blinktree_server/server.cpp @@ -0,0 +1,32 @@ +#include "server.h" +#include "network/server.h" +#include + +using namespace application::blinktree_server; + +Server::Server(const std::uint64_t port, mx::util::core_set &&cores, const std::uint16_t prefetch_distance, const mx::synchronization::isolation_level node_isolation_level, const mx::synchronization::protocol preferred_synchronization_method) + : _port(port), _cores(std::move(cores)), _prefetch_distance(prefetch_distance), _node_isolation_level(node_isolation_level), _preferred_synchronization_method(preferred_synchronization_method) +{ +} + +void Server::run() +{ + network::Server* server; + + mx::tasking::runtime::init(this->_cores, this->_prefetch_distance, /* use mx tasking's task allocator*/ false); + + this->_tree = std::make_unique>( + this->_node_isolation_level, this->_preferred_synchronization_method); + + server = new network::Server{this->_port, mx::tasking::runtime::channels()}; + + std::cout << "Waiting for requests on port :" << this->_port << std::endl; + auto network_thread = std::thread{[server, tree = this->_tree.get()]() { + server->listen(tree); + }}; + mx::tasking::runtime::start_and_wait(); + + //network_thread.join(); + + delete server; +} \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/server.h b/repos/mml/src/app/blinktree_server/server.h new file mode 100644 index 0000000000..af0d868cbf --- /dev/null +++ b/repos/mml/src/app/blinktree_server/server.h @@ -0,0 +1,30 @@ +#pragma once + +#include +#include + +namespace application::blinktree_server { +class Server +{ +public: + Server(std::uint64_t port, mx::util::core_set&& cores, std::uint16_t prefetch_distance, mx::synchronization::isolation_level node_isolation_level, mx::synchronization::protocol preferred_synchronization_method); + + void run(); +private: + const std::uint64_t _port; + + const std::uint16_t _prefetch_distance; + + /// Cores. + mx::util::core_set _cores; + + // The synchronization mechanism to use for tree nodes. + const mx::synchronization::isolation_level _node_isolation_level; + + // Preferred synchronization method. + const mx::synchronization::protocol _preferred_synchronization_method; + + /// Tree. + std::unique_ptr> _tree; +}; +} \ No newline at end of file diff --git a/repos/mml/src/app/blinktree_server/target.mk b/repos/mml/src/app/blinktree_server/target.mk new file mode 100644 index 0000000000..f0cf215d64 --- /dev/null +++ b/repos/mml/src/app/blinktree_server/target.mk @@ -0,0 +1,27 @@ +MXINC_DIR=$(REP_DIR)/src/app/blinktree_server +MXINC_DIR+=-I$(REP_DIR)/src/app/blinktree +GENODE_GCC_TOOLCHAIN_DIR ?= /usr/local/genode/tool/21.05 +MXBENCH_DIR=$(REP_DIR)/src/app/blinktree + +TARGET = blinktree_daemon +# soure file for benchmark framework +SRC_MXBENCH = ${MXBENCH_DIR}/benchmark/workload_set.cpp +SRC_MXBENCH += ${MXBENCH_DIR}/benchmark/workload.cpp +SRC_MXBENCH += ${MXBENCH_DIR}/benchmark/cores.cpp +SRC_MXBENCH += ${MXBENCH_DIR}/benchmark/string_util.cpp +SRC_MXBENCH += ${MXBENCH_DIR}/benchmark/perf.cpp +# source files for blinktree benchmark +SRC_BTREE += main.cpp +SRC_BTREE += server.cpp +SRC_BTREE += network/server.cpp + +SRC_CC = ${SRC_MXBENCH} ${SRC_BTREE} +LIBS += base libc stdcxx mxtasking +EXT_OBJECTS += /usr/local/genode/tool/lib/clang/14.0.5/lib/linux/libclang_rt.builtins-x86_64.a /usr/local/genode/tool/lib/libatomic.a +CUSTOM_CC = /usr/local/genode/tool/bin/clang +CUSTOM_CXX = /usr/local/genode/tool/bin/clang++ +CC_OPT += --target=x86_64-genode --sysroot=/does/not/exist --gcc-toolchain=$(GENODE_GCC_TOOLCHAIN_DIR) -Wno-error -O2 -g -fno-aligned-new -DNDEBUG -I$(MXINC_DIR) -std=c++20 #-D_GLIBCXX_ATOMIC_BUILTINS_8 -D__GCC_HAVE_SYNC_COMPARE_AND_SWAP_8 +CC_OPT += -femulated-tls -DCLANG_CXX11_ATOMICS +CC_CXX_WARN_STRICT = +CUSTOM_CXX_LIB := $(CROSS_DEV_PREFIX)g++ +#CXX_LD += $(CROSS_DEV_PREFIX)g++