3 #ifdef CONFIG_HAVE_TIMERFD_CREATE
4 #include <sys/timerfd.h>
6 #ifdef CONFIG_VALGRIND_DEV
7 #include <valgrind/drd.h>
9 #define DRD_IGNORE_VAR(x) do { } while (0)
14 #include "helper_thread.h"
15 #include "steadystate.h"
18 static int sleep_accuracy_ms;
19 static int timerfd = -1;
27 static struct helper_data {
29 int pipe[2]; /* 0: read end; 1: write end. */
30 struct sk_out *sk_out;
32 struct fio_sem *startup_sem;
35 struct interval_timer {
37 struct timespec expires;
42 void helper_thread_destroy(void)
47 close(helper_data->pipe[0]);
48 close(helper_data->pipe[1]);
53 static void sock_init(void)
58 /* It is allowed to call WSAStartup() more than once. */
59 res = WSAStartup(MAKEWORD(2, 2), &wsaData);
63 static int make_nonblocking(int fd)
65 unsigned long arg = 1;
67 return ioctlsocket(fd, FIONBIO, &arg);
70 static int write_to_pipe(int fd, const void *buf, size_t len)
72 return send(fd, buf, len, 0);
75 static int read_from_pipe(int fd, void *buf, size_t len)
77 return recv(fd, buf, len, 0);
80 static void sock_init(void)
84 static int make_nonblocking(int fd)
86 return fcntl(fd, F_SETFL, O_NONBLOCK);
89 static int write_to_pipe(int fd, const void *buf, size_t len)
91 return write(fd, buf, len);
94 static int read_from_pipe(int fd, void *buf, size_t len)
96 return read(fd, buf, len);
100 static void block_signals(void)
102 #ifdef HAVE_PTHREAD_SIGMASK
105 ret = pthread_sigmask(SIG_UNBLOCK, NULL, &sigmask);
107 ret = pthread_sigmask(SIG_BLOCK, &sigmask, NULL);
112 static void submit_action(enum action a)
120 ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
124 void helper_reset(void)
126 submit_action(A_RESET);
130 * May be invoked in signal handler context and hence must only call functions
131 * that are async-signal-safe. See also
132 * https://pubs.opengroup.org/onlinepubs/9699919799/functions/V2_chap02.html#tag_15_04_03.
134 void helper_do_stat(void)
136 submit_action(A_DO_STAT);
139 bool helper_should_exit(void)
144 return helper_data->exit;
147 void helper_thread_exit(void)
152 helper_data->exit = 1;
153 submit_action(A_EXIT);
154 pthread_join(helper_data->thread, NULL);
157 /* Resets timers and returns the time in milliseconds until the next event. */
158 static int reset_timers(struct interval_timer timer[], int num_timers,
159 struct timespec *now)
161 uint32_t msec_to_next_event = INT_MAX;
164 for (i = 0; i < num_timers; ++i) {
165 timer[i].expires = *now;
166 timespec_add_msec(&timer[i].expires, timer[i].interval_ms);
167 msec_to_next_event = min_not_zero(msec_to_next_event,
168 timer[i].interval_ms);
171 return msec_to_next_event;
175 * Waits for an action from fd during at least timeout_ms. `fd` must be in
178 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
180 struct timeval timeout = {
181 .tv_sec = timeout_ms / 1000,
182 .tv_usec = (timeout_ms % 1000) * 1000,
189 res = read_from_pipe(fd, &action, sizeof(action));
190 if (res > 0 || timeout_ms == 0)
196 #ifdef CONFIG_HAVE_TIMERFD_CREATE
199 * If the timer frequency is 100 Hz, select() will round up
200 * `timeout` to the next multiple of 1 / 100 Hz = 10 ms. Hence
201 * use a high-resolution timer if possible to increase
202 * select() timeout accuracy.
204 struct itimerspec delta = {};
206 delta.it_value.tv_sec = timeout.tv_sec;
207 delta.it_value.tv_nsec = timeout.tv_usec * 1000;
208 res = timerfd_settime(timerfd, 0, &delta, NULL);
210 FD_SET(timerfd, &rfds);
213 res = select(max(fd, timerfd) + 1, &rfds, NULL, &efds,
214 timerfd >= 0 ? NULL : &timeout);
216 log_err("fio: select() call in helper thread failed: %s",
220 if (FD_ISSET(fd, &rfds))
221 read_from_pipe(fd, &action, sizeof(action));
222 if (timerfd >= 0 && FD_ISSET(timerfd, &rfds)) {
223 res = read(timerfd, &exp, sizeof(exp));
224 assert(res == sizeof(exp));
230 * Verify whether or not timer @it has expired. If timer @it has expired, call
231 * @it->func(). @now is the current time. @msec_to_next_event is an
232 * input/output parameter that represents the time until the next event.
234 static int eval_timer(struct interval_timer *it, const struct timespec *now,
235 unsigned int *msec_to_next_event)
240 /* interval == 0 means that the timer is disabled. */
241 if (it->interval_ms == 0)
244 delta_ms = rel_time_since(now, &it->expires);
245 expired = delta_ms <= sleep_accuracy_ms;
247 timespec_add_msec(&it->expires, it->interval_ms);
248 delta_ms = rel_time_since(now, &it->expires);
249 if (delta_ms < it->interval_ms - sleep_accuracy_ms ||
250 delta_ms > it->interval_ms + sleep_accuracy_ms) {
251 dprint(FD_HELPERTHREAD,
252 "%s: delta = %" PRIi64 " <> %u. Clock jump?\n",
253 it->name, delta_ms, it->interval_ms);
254 delta_ms = it->interval_ms;
256 timespec_add_msec(&it->expires, it->interval_ms);
259 *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
260 return expired ? it->func() : 0;
263 static void *helper_thread_main(void *data)
265 struct helper_data *hd = data;
266 unsigned int msec_to_next_event, next_log;
267 struct interval_timer timer[] = {
270 .interval_ms = DISK_UTIL_MSEC,
271 .func = update_io_ticks,
274 .name = "status_interval",
275 .interval_ms = status_interval,
276 .func = __show_running_run_stats,
279 .name = "steadystate",
280 .interval_ms = steadystate_enabled ? STEADYSTATE_MSEC :
282 .func = steadystate_check,
286 int clk_tck, ret = 0;
289 clk_tck = sysconf(_SC_CLK_TCK);
292 * The timer frequence is variable on Windows. Instead of trying to
293 * query it, use 64 Hz, the clock frequency lower bound. See also
294 * https://carpediemsystems.co.uk/2019/07/18/windows-system-timer-granularity/.
298 dprint(FD_HELPERTHREAD, "clk_tck = %d\n", clk_tck);
300 sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
302 #ifdef CONFIG_HAVE_TIMERFD_CREATE
303 timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
304 assert(timerfd >= 0);
305 sleep_accuracy_ms = 1;
308 sk_out_assign(hd->sk_out);
310 /* Let another thread handle signals. */
313 fio_get_mono_time(&ts);
314 msec_to_next_event = reset_timers(timer, ARRAY_SIZE(timer), &ts);
316 fio_sem_up(hd->startup_sem);
318 while (!ret && !hd->exit) {
322 action = wait_for_action(hd->pipe[0], msec_to_next_event);
323 if (action == A_EXIT)
326 fio_get_mono_time(&ts);
328 msec_to_next_event = INT_MAX;
330 if (action == A_RESET)
331 msec_to_next_event = reset_timers(timer,
332 ARRAY_SIZE(timer), &ts);
334 for (i = 0; i < ARRAY_SIZE(timer); ++i)
335 ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
337 if (action == A_DO_STAT)
338 __show_running_run_stats();
340 next_log = calc_log_samples();
342 next_log = DISK_UTIL_MSEC;
344 msec_to_next_event = min(next_log, msec_to_next_event);
345 dprint(FD_HELPERTHREAD,
346 "next_log: %u, msec_to_next_event: %u\n",
347 next_log, msec_to_next_event);
350 print_thread_status();
358 fio_writeout_logs(false);
365 * Connect two sockets to each other to emulate the pipe() system call on Windows.
367 int pipe_over_loopback(int fd[2])
369 struct sockaddr_in addr = { .sin_family = AF_INET };
370 socklen_t len = sizeof(addr);
373 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
377 fd[0] = socket(AF_INET, SOCK_STREAM, 0);
380 fd[1] = socket(AF_INET, SOCK_STREAM, 0);
383 res = bind(fd[0], (struct sockaddr *)&addr, len);
386 res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
389 res = listen(fd[0], 1);
392 res = connect(fd[1], (struct sockaddr *)&addr, len);
395 res = accept(fd[0], NULL, NULL);
412 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
414 struct helper_data *hd;
417 hd = scalloc(1, sizeof(*hd));
424 #if defined(CONFIG_PIPE2)
425 ret = pipe2(hd->pipe, O_CLOEXEC);
426 #elif defined(CONFIG_PIPE)
427 ret = pipe(hd->pipe);
429 ret = pipe_over_loopback(hd->pipe);
434 ret = make_nonblocking(hd->pipe[0]);
437 hd->startup_sem = startup_sem;
439 DRD_IGNORE_VAR(helper_data);
441 ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
443 log_err("Can't create helper thread: %s\n", strerror(ret));
449 dprint(FD_MUTEX, "wait on startup_sem\n");
450 fio_sem_down(startup_sem);
451 dprint(FD_MUTEX, "done waiting on startup_sem\n");