denshooter/genode
1
0
Fork 0
mirror of https://github.com/mmueller41/genode.git synced 2026-01-21 12:32:56 +01:00
Code Issues Packages Projects Releases Wiki Activity

Networked version of B-link tree benchmark.

Browse Source
This commit is contained in:
Michael Mueller
2024-10-24 14:24:36 +02:00
parent ee5502ed34
commit 635d1a637c
13 changed files with 1284 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
repos/mml/src/app/blinktree_server/README.md Normal file
View File

@@ -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 <fill-file> <mixed-file>` 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 <NUMBER>` specifies the number of repetitions of each workload.
* `-s <NUMBER>` 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 <NUMBER>` 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 <FILE>` 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

199
repos/mml/src/app/blinktree_server/benchmark.cpp Normal file
View File

@@ -0,0 +1,199 @@
#include "benchmark.h"
#include <cstdlib>
#include <iostream>
#include <json.hpp>
#include <memory>
#include <mx/memory/global_heap.h>
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<db::index::blinktree::BLinkTree<std::uint64_t, std::int64_t>>(
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<RequestSchedulerTask>(
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<std::uint16_t>(static_cast<benchmark::phase>(this->_workload)),
this->_current_iteration + 1, this->_cores.current());
}
const mx::util::core_set &Benchmark::core_set()
{
if (this->_current_iteration == std::numeric_limits<std::uint16_t>::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<nlohmann::json>(*(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<std::uint16_t>(static_cast<benchmark::phase>(this->_workload))) + "-iteration-" +
std::to_string(this->_current_iteration) + ".json";
}

103
repos/mml/src/app/blinktree_server/benchmark.h Normal file
View File

@@ -0,0 +1,103 @@
#pragma once
#include "listener.h"
#include "request_scheduler.h"
#include <array>
#include <atomic>
#include <benchmark/chronometer.h>
#include <benchmark/cores.h>
#include <benchmark/workload.h>
#include <cstdint>
#include <db/index/blinktree/b_link_tree.h>
#include <memory>
#include <mx/util/core_set.h>
#include <string>
#include <vector>
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<std::uint16_t>::max();
// Workload to get requests from.
benchmark::Workload _workload;
// Tree to run requests on.
std::unique_ptr<db::index::blinktree::BLinkTree<std::uint64_t, std::int64_t>> _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<RequestSchedulerTask *> _request_scheduler;
// Chronometer for starting/stopping time and performance counter.
alignas(64) benchmark::Chronometer<std::uint16_t> _chronometer;
/**
* @return Name of the file to write profiling results to.
*/
[[nodiscard]] std::string profile_file_name() const;
};
} // namespace application::blinktree_benchmark

17
repos/mml/src/app/blinktree_server/config.h Normal file
View File

@@ -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

15
repos/mml/src/app/blinktree_server/listener.h Normal file
View File

@@ -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

153
repos/mml/src/app/blinktree_server/main.cpp Normal file
View File

@@ -0,0 +1,153 @@
#include "benchmark.h"
#include <argparse.hpp>
#include <benchmark/cores.h>
#include <mx/system/environment.h>
#include <mx/tasking/runtime.h>
#include <mx/util/core_set.h>
#include <tuple>
#include "server.h"
#include <cstring>
#include <cstdio>
/* Genode includes */
#include <libc/component.h>
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<Server *, std::uint16_t, bool> 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<Server *, std::uint16_t, bool> 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<bool>("-sco") ? mx::util::core_set::Order::Ascending : mx::util::core_set::Order::NUMAAware;
auto cores = mx::util::core_set::build(argument_parser.get<std::uint16_t>("cores")-1, order);
const auto isolation_level = argument_parser.get<bool>("--exclusive")
? mx::synchronization::isolation_level::Exclusive
: mx::synchronization::isolation_level::ExclusiveWriter;
auto preferred_synchronization_method = mx::synchronization::protocol::Queue;
if (argument_parser.get<bool>("--latched"))
{
preferred_synchronization_method = mx::synchronization::protocol::Latch;
}
else if (argument_parser.get<bool>("--olfit"))
{
preferred_synchronization_method = mx::synchronization::protocol::OLFIT;
}
else if (argument_parser.get<bool>("--sync4me"))
{
preferred_synchronization_method = mx::synchronization::protocol::None;
}
// Create the benchmark.
auto *server = new Server(argument_parser.get<std::uint64_t>("--port"), std::move(cores), argument_parser.get<std::uint16_t>("-pd"), isolation_level, preferred_synchronization_method);
return {server, argument_parser.get<std::uint16_t>("-pd"), argument_parser.get<bool>("--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);
});
}

9
repos/mml/src/app/blinktree_server/network/config.h Normal file
View File

@@ -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

278
repos/mml/src/app/blinktree_server/network/server.cpp Normal file
View File

@@ -0,0 +1,278 @@
#include "server.h"
#include <limits>
#include <mx/tasking/runtime.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <db/index/blinktree/lookup_task.h>
#include <db/index/blinktree/insert_value_task.h>
#include <db/index/blinktree/update_task.h>
#include <mx/system/topology.h>
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<std::uint64_t, std::int64_t, ResponseHandler>>(
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<std::uint64_t, std::int64_t, ResponseHandler>>(
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<std::uint64_t, std::int64_t, ResponseHandler>>(
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<ResponseHandler*>(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<RequestTask*>(std::malloc( sizeof(RequestTask) * config::max_connections()));
}
Server::~Server() {
::free(this->_response_handlers);
::free(this->_request_tasks);
}
bool Server::listen(db::index::blinktree::BLinkTree<std::uint64_t, std::int64_t>* 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<sockaddr *>(&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<sockaddr *>(&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<const void *>(&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<std::uint16_t>::max();
}
void Server::stop() noexcept
{
this->_is_running = false;
}

90
repos/mml/src/app/blinktree_server/network/server.h Normal file
View File

@@ -0,0 +1,90 @@
#pragma once
#include "config.h"
#include <array>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mx/memory/fixed_size_allocator.h>
#include <mx/tasking/config.h>
#include <mx/tasking/scheduler.h>
#include <mx/util/core_set.h>
#include <optional>
#include <string>
#include <db/index/blinktree/b_link_tree.h>
#include <db/index/blinktree/listener.h>
namespace application::blinktree_server::network {
class Server;
class alignas(64) ResponseHandler final : public db::index::blinktree::Listener<std::uint64_t, std::int64_t>
{
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<std::uint64_t, std::int64_t>* 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<std::uint64_t, std::int64_t>* 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<std::uint64_t, std::int64_t>* 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<std::uint64_t, std::int64_t>* _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<std::uint64_t, std::int64_t>* tree);
[[nodiscard]] bool is_running() const noexcept { return _is_running; }
private:
const std::uint64_t _port;
std::int32_t _socket;
std::array<std::uint32_t, config::max_connections()> _client_sockets;
std::array<char, 2048U> _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

252
repos/mml/src/app/blinktree_server/request_scheduler.h Normal file
View File

@@ -0,0 +1,252 @@
#pragma once
#include "config.h"
#include "listener.h"
#include <atomic>
#include <benchmark/workload.h>
#include <cstdint>
#include <db/index/blinktree/b_link_tree.h>
#include <db/index/blinktree/config.h>
#include <db/index/blinktree/insert_value_task.h>
#include <db/index/blinktree/lookup_task.h>
#include <db/index/blinktree/update_task.h>
#include <mx/resource/resource.h>
#include <mx/tasking/runtime.h>
#include <mx/tasking/task.h>
#include <mx/util/core_set.h>
#include <mx/util/reference_counter.h>
namespace application::blinktree_benchmark {
class RequestIndex
{
public:
static RequestIndex make_finished() { return RequestIndex{std::numeric_limits<decltype(_index)>::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<std::uint64_t, std::uint64_t> &&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<decltype(_index)>::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<std::uint64_t, std::int64_t> *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<RequestContainer>(
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<RequestContainer>(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<std::uint64_t, std::int64_t, RequestContainer>>(
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<std::uint64_t, std::int64_t, RequestContainer>>(
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<std::uint64_t, std::int64_t, RequestContainer>>(
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<RequestContainer>(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<std::uint64_t, std::int64_t> *_tree;
// Benchmark listener to notify on requests are done.
Listener *_listener;
};
} // namespace application::blinktree_benchmark

32
repos/mml/src/app/blinktree_server/server.cpp Normal file
View File

@@ -0,0 +1,32 @@
#include "server.h"
#include "network/server.h"
#include <iostream>
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<db::index::blinktree::BLinkTree<std::uint64_t, std::int64_t>>(
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;
}

30
repos/mml/src/app/blinktree_server/server.h Normal file
View File

@@ -0,0 +1,30 @@
#pragma once
#include <db/index/blinktree/b_link_tree.h>
#include <mx/util/core_set.h>
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<db::index::blinktree::BLinkTree<std::uint64_t, std::int64_t>> _tree;
};
}

27
repos/mml/src/app/blinktree_server/target.mk Normal file
View File

@@ -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++