Merge branch 'pulse-eval' into 'master'

Pulse: initial pulse evaluation commit

See merge request !376

eval/Pulse.cpp

+// SPDX-License-Identifier: LGPL-3.0-or-later
+// Copyright © 2022 Florian Fischer
+#include <boost/program_options.hpp>
+#include <chrono>
+#include <compare>
+#include <cstdlib>
+#include <exception>
+#include <fstream>	// IWYU pragma: keep
+#include <iostream>
+#include <string>
+
+#include "CountingPrivateSemaphore.hpp"
+#include "Runtime.hpp"
+#include "RuntimeBuilder.hpp"
+#include "emper-common.h"
+#include "emper.hpp"
+
+namespace po = boost::program_options;
+
+using Clock = std::chrono::high_resolution_clock;
+using TimePoint = std::chrono::time_point<Clock>;
+
+// Pulse of new work in seconds
+static unsigned pulse = 1;
+// Number of pulses
+static unsigned iterations = 30;
+// Utilization of the runtime in percent
+static unsigned utilization = 80;
+
+class Work {
+ public:
+	static Work** doneWork;
+	static thread_local Work** localDoneWork;
+
+	TimePoint start;
+	TimePoint end;
+	Work* next = nullptr;
+
+	Work() : start(Clock::now()) {}
+
+	void run() {
+		const auto deadline = Clock::now() + std::chrono::seconds(pulse);
+		while (Clock::now() < deadline) {
+		}
+
+		finish();
+	}
+
+	void finish() {
+		this->end = std::chrono::high_resolution_clock::now();
+		this->next = *localDoneWork;
+		*localDoneWork = this;
+	}
+};
+
+Work** Work::doneWork;
+thread_local Work** Work::localDoneWork;
+
+static void pulser(Runtime& runtime) {
+	const workerid_t workerCount = runtime.getWorkerCount();
+	const workerid_t workItems = (workerCount * utilization) / 100;
+
+	CPS cps;
+	for (unsigned i = 0; i < iterations; ++i) {
+		for (unsigned w = 0; w < workItems; ++w) {
+			Work* work = new Work();
+			spawn([=] { work->run(); }, cps);
+		}
+
+		// TODO: better calculate the time until the next pulse
+		emper::sleep(pulse);
+	}
+	cps.wait();
+	runtime.initiateTermination();
+}
+
+static auto eval(const po::variables_map& vm) -> int {
+	if (vm.count("pulse")) pulse = vm["pulse"].as<unsigned>();
+
+	if (vm.count("iterations")) iterations = vm["iterations"].as<unsigned>();
+
+	if (vm.count("utilization")) utilization = vm["utilization"].as<unsigned>();
+
+	std::cerr << "Starting pulse evaluation with pulse=" << pulse;
+	std::cerr << ", iterations=" << iterations;
+	std::cerr << " and utilization=" << utilization;
+	std::cerr << std::endl;
+
+	const workerid_t workerCount = Runtime::getDefaultWorkerCount();
+	Work::doneWork = new Work*[workerCount];
+
+	RuntimeBuilder runtimeBuilder;
+	Runtime runtime = runtimeBuilder.withWorkerCount(workerCount)
+												.newWorkerHook([](workerid_t id) {
+													Work::localDoneWork = &Work::doneWork[id];
+													*Work::localDoneWork = nullptr;
+												})
+												.build();
+
+	runtime.executeAndWait([&] { pulser(runtime); });
+
+	std::fstream fout;
+	std::ostream* out = &std::cout;
+	if (vm.count("latencies-file")) {
+		out = &fout;
+		fout.open(vm["latencies-file"].as<std::string>(), std::fstream::out);
+	}
+
+	for (unsigned i = 0; i < workerCount; ++i) {
+		auto* localDoneWork = Work::doneWork[i];
+		while (localDoneWork) {
+			auto* work = localDoneWork;
+			*out << i << ","
+					 << std::chrono::duration_cast<std::chrono::nanoseconds>(work->end - work->start).count()
+					 << std::endl;
+			localDoneWork = work->next;
+			delete work;
+		}
+	}
+	delete[] Work::doneWork;
+
+	if (fout.is_open()) fout.close();
+
+	return EXIT_SUCCESS;
+}
+
+auto main(int argc, char* argv[]) -> int {
+	po::options_description desc("Allowed options");
+	// clang-format off
+	desc.add_options()
+		("help", "Show help")
+		("pulse", po::value<unsigned>(), "The pulse of new work")
+		("iterations", po::value<unsigned>(), "The number of pulses")
+		("utilization", po::value<unsigned>(), "The target utilization of the runtime")
+		("latencies-file", po::value<std::string>(), "File to store all collected latencies")
+		;
+	// clang-format on
+
+	auto parse_result = po::command_line_parser(argc, argv).options(desc).run();
+
+	po::variables_map vm;
+	po::store(parse_result, vm);
+	po::notify(vm);
+
+	if (vm.count("help")) {
+		std::cout << desc << "\n";
+		return EXIT_SUCCESS;
+	}
+
+	try {
+		return eval(vm);
+	} catch (std::exception& e) {
+		std::cerr << e.what() << std::endl;
+		return EXIT_FAILURE;
+	}
+}

eval/meson.build

@@ -16,6 +16,14 @@ executable(
   dependencies: emper_dep,
 )
 
+if cpp_can_link_with_boost_program_options
+  executable(
+    'pulse',
+    'Pulse.cpp',
+    dependencies: [emper_dep, boost_program_options_dep],
+  )
+endif
+
 std_atomic_wait_notify_code = '''\
 #include <atomic>
 int main() {