EchoServer.cpp

// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright © 2020-2021 Florian Fischer
#include <sys/socket.h>

#include <atomic>
#include <cerrno>
#include <chrono>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <random>
#include <string>

#include "Common.hpp"
#include "Debug.hpp"
#include "Runtime.hpp"
#include "RuntimeBuilder.hpp"
#include "emper-common.h"
#include "emper-config.h"
#include "io.hpp"

#ifdef EMPER_HAS_COMPARE_H
#include <compare>
#endif

static const std::string HOST = "::";
static const std::string PORT = "12345";
static const int BACKLOG = 1024;

static const size_t BUF_SIZE = 1024;

static unsigned int computations_us = 0;
static unsigned int max_computations_us = 0;
static float max_computations_probability = -1;

static std::atomic<bool> quit = false;

static thread_local std::mt19937 randGenerator;
static auto getComputation() -> unsigned {
	// fixed computation is computations_us
	if (!max_computations_us) return computations_us;

	// computation is in range [computations_us, max_computations_us]
	if (max_computations_probability == -1) {
		std::uniform_int_distribution<unsigned int> distribution(computations_us, max_computations_us);
		return computations_us += distribution(randGenerator);
	}

	// computation is either computations_us or max_computations_us with probability
	// max_computations_probability
	std::uniform_real_distribution<float> distribution(0, 1);
	float p = distribution(randGenerator);
	return p >= max_computations_probability ? max_computations_us : computations_us;
}

auto main(int argc, char* argv[]) -> int {
	std::string host = HOST;
	std::string port = PORT;

	if (argc > 5) {
		std::cerr << "Usage: " << argv[0] << " [port] [computations_us]"
							<< " [max_computations_us] [max_computations_probability]" << std::endl;
		exit(EXIT_FAILURE);
	}

	if (argc > 1) {
		port = std::string(argv[1]);
	}

	if (argc > 2) {
		computations_us = std::stoi(argv[2]);
	}

	if (argc > 3) {
		max_computations_us = std::stoi(argv[3]);
		if (max_computations_us < computations_us)
			DIE_MSG("max_computations_us must be bigger than computations_us");
	}

	if (argc > 4) {
		max_computations_probability = std::stof(argv[4]);
		if (max_computations_probability < 0 || max_computations_probability > 1)
			DIE_MSG("max_computations_probability must be in [0,1]");
	}

	std::cout << "Echoserver listening on " << host << ":" << port;
	if (computations_us) {
		std::cout << " with " << computations_us;
		if (max_computations_us) std::cout << " - " << max_computations_us;
		std::cout << " us computations";
	}
	std::cout << std::endl;

	RuntimeBuilder runtimeBuilder;
	if (max_computations_us) {
		runtimeBuilder.newWorkerHook([](workerid_t id) { randGenerator.seed(id); });
	}

	auto runtime = runtimeBuilder.build();

	auto serverFunc = [](int socket) {
		// NOLINTNEXTLINE(modernize-avoid-c-arrays)
		char buf[BUF_SIZE];
		while (!quit.load(std::memory_order_consume)) {
			ssize_t bytes_recv = emper::io::recvAndWait(socket, buf, sizeof(buf), 0);
			if (unlikely(bytes_recv <= 0)) {
				// socket was shutdown
				if (bytes_recv < 0) {
					LOGE("server read failed:" << strerror(errno));
				}
				break;
			}

			if (unlikely(bytes_recv == 5 && strncmp("quit\n", buf, bytes_recv) == 0)) {
				quit = true;
				Runtime::getRuntime()->initiateTermination();
				break;
			}

			if (computations_us) {
				unsigned int computation = getComputation();
				const auto start = std::chrono::steady_clock::now();
				const auto deadline = start + std::chrono::microseconds(computation);
				// TODO: The suppressed linter error below may be a false positive
				// reported by clang-tidy.
				// NOLINTNEXTLINE(modernize-use-nullptr)
				while (std::chrono::steady_clock::now() < deadline) {
				}
			}

			ssize_t bytes_send = emper::io::sendAndWait(socket, buf, bytes_recv, MSG_NOSIGNAL, true);
			if (unlikely(bytes_recv != bytes_send)) {
				LOGE("server send failed: " << strerror(errno));
				break;
			}
		}

		emper::io::closeAndForget(socket);
	};

	auto* listener =
			emper::io::tcp_listener(host, port, serverFunc, BACKLOG, {emper::io::SockOpt::ReusePort});

	if (!listener) {
		exit(EXIT_FAILURE);
	}

	runtime.scheduleFromAnywhere(*listener);

	runtime.waitUntilFinished();

	return EXIT_SUCCESS;
}