EchoServer.cpp 4.19 KB
Newer Older
1
2
// SPDX-License-Identifier: LGPL-3.0-or-later
// Copyright © 2020-2021 Florian Fischer
Florian Fischer's avatar
Florian Fischer committed
3
#include <sys/socket.h>
4

5
#include <atomic>
6
#include <cerrno>
7
#include <chrono>
8
9
10
#include <cstdlib>
#include <cstring>
#include <iostream>
11
#include <random>
12
13
14
#include <string>

#include "Common.hpp"
Florian Fischer's avatar
Florian Fischer committed
15
#include "Debug.hpp"
16
#include "Runtime.hpp"
17
18
#include "RuntimeBuilder.hpp"
#include "emper-common.h"
19
#include "emper-config.h"
20
21
#include "io.hpp"

22
23
24
25
#ifdef EMPER_HAS_COMPARE_H
#include <compare>
#endif

26
27
28
static const std::string HOST = "::";
static const std::string PORT = "12345";
static const int BACKLOG = 1024;
29

30
31
static const size_t BUF_SIZE = 1024;

32
static unsigned int computations_us = 0;
33
static unsigned int max_computations_us = 0;
34
static float max_computations_probability = -1;
35

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

38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
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;
}

56
57
auto main(int argc, char* argv[]) -> int {
	std::string host = HOST;
Florian Fischer's avatar
Florian Fischer committed
58
	std::string port = PORT;
59

60
61
62
	if (argc > 5) {
		std::cerr << "Usage: " << argv[0] << " [port] [computations_us]"
							<< " [max_computations_us] [max_computations_probability]" << std::endl;
63
64
65
66
		exit(EXIT_FAILURE);
	}

	if (argc > 1) {
Florian Fischer's avatar
Florian Fischer committed
67
		port = std::string(argv[1]);
68
69
	}

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

74
75
76
77
78
79
	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");
	}

80
81
82
83
84
85
	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]");
	}

86
87
88
89
90
91
92
	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;
93

94
95
96
97
98
99
100
	RuntimeBuilder runtimeBuilder;
	if (max_computations_us) {
		runtimeBuilder.newWorkerHook([](workerid_t id) { randGenerator.seed(id); });
	}

	auto runtime = runtimeBuilder.build();

101
	auto serverFunc = [](int socket) {
102
		// NOLINTNEXTLINE(modernize-avoid-c-arrays)
103
		char buf[BUF_SIZE];
104
		while (!quit.load(std::memory_order_consume)) {
105
106
107
108
			ssize_t bytes_recv = emper::io::recvAndWait(socket, buf, sizeof(buf), 0);
			if (unlikely(bytes_recv <= 0)) {
				// socket was shutdown
				if (bytes_recv < 0) {
109
					LOGE("server read failed:" << strerror(errno));
110
				}
111
				break;
112
113
114
			}

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

120
			if (computations_us) {
121
				unsigned int computation = getComputation();
122
123
124
125
126
127
128
				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) {
				}
129
130
			}

131
132
133
			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));
134
				break;
135
			}
136
		}
137
138

		emper::io::closeAndForget(socket);
139
140
141
142
	};

	auto* listener =
			emper::io::tcp_listener(host, port, serverFunc, BACKLOG, {emper::io::SockOpt::ReusePort});
143
144
145
146
147
148
149
150
151

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

	runtime.scheduleFromAnywhere(*listener);

	runtime.waitUntilFinished();

152
	return EXIT_SUCCESS;
153
}