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)
13 #include "os/os-windows.h"
18 #include "helper_thread.h"
19 #include "steadystate.h"
22 static int sleep_accuracy_ms;
23 static int timerfd = -1;
31 static struct helper_data {
33 int pipe[2]; /* 0: read end; 1: write end. */
34 struct sk_out *sk_out;
36 struct fio_sem *startup_sem;
39 struct interval_timer {
41 struct timespec expires;
46 void helper_thread_destroy(void)
51 close(helper_data->pipe[0]);
52 close(helper_data->pipe[1]);
57 static void sock_init(void)
62 /* It is allowed to call WSAStartup() more than once. */
63 res = WSAStartup(MAKEWORD(2, 2), &wsaData);
67 static int make_nonblocking(int fd)
69 unsigned long arg = 1;
71 return ioctlsocket(fd, FIONBIO, &arg);
74 static int write_to_pipe(int fd, const void *buf, size_t len)
76 return send(fd, buf, len, 0);
79 static int read_from_pipe(int fd, void *buf, size_t len)
81 return recv(fd, buf, len, 0);
84 static void sock_init(void)
88 static int make_nonblocking(int fd)
90 return fcntl(fd, F_SETFL, O_NONBLOCK);
93 static int write_to_pipe(int fd, const void *buf, size_t len)
95 return write(fd, buf, len);
98 static int read_from_pipe(int fd, void *buf, size_t len)
100 return read(fd, buf, len);
104 static void block_signals(void)
106 #ifdef HAVE_PTHREAD_SIGMASK
109 ret = pthread_sigmask(SIG_UNBLOCK, NULL, &sigmask);
111 ret = pthread_sigmask(SIG_BLOCK, &sigmask, NULL);
116 static void submit_action(enum action a)
124 ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
128 void helper_reset(void)
130 submit_action(A_RESET);
134 * May be invoked in signal handler context and hence must only call functions
135 * that are async-signal-safe. See also
136 * https://pubs.opengroup.org/onlinepubs/9699919799/functions/V2_chap02.html#tag_15_04_03.
138 void helper_do_stat(void)
140 submit_action(A_DO_STAT);
143 bool helper_should_exit(void)
148 return helper_data->exit;
151 void helper_thread_exit(void)
156 helper_data->exit = 1;
157 submit_action(A_EXIT);
158 pthread_join(helper_data->thread, NULL);
161 /* Resets timers and returns the time in milliseconds until the next event. */
162 static int reset_timers(struct interval_timer timer[], int num_timers,
163 struct timespec *now)
165 uint32_t msec_to_next_event = INT_MAX;
168 for (i = 0; i < num_timers; ++i) {
169 timer[i].expires = *now;
170 timespec_add_msec(&timer[i].expires, timer[i].interval_ms);
171 msec_to_next_event = min_not_zero(msec_to_next_event,
172 timer[i].interval_ms);
175 return msec_to_next_event;
179 * Waits for an action from fd during at least timeout_ms. `fd` must be in
182 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
184 struct timeval timeout = {
185 .tv_sec = timeout_ms / 1000,
186 .tv_usec = (timeout_ms % 1000) * 1000,
193 res = read_from_pipe(fd, &action, sizeof(action));
194 if (res > 0 || timeout_ms == 0)
200 #ifdef CONFIG_HAVE_TIMERFD_CREATE
203 * If the timer frequency is 100 Hz, select() will round up
204 * `timeout` to the next multiple of 1 / 100 Hz = 10 ms. Hence
205 * use a high-resolution timer if possible to increase
206 * select() timeout accuracy.
208 struct itimerspec delta = {};
210 delta.it_value.tv_sec = timeout.tv_sec;
211 delta.it_value.tv_nsec = timeout.tv_usec * 1000;
212 res = timerfd_settime(timerfd, 0, &delta, NULL);
214 FD_SET(timerfd, &rfds);
217 res = select(max(fd, timerfd) + 1, &rfds, NULL, &efds,
218 timerfd >= 0 ? NULL : &timeout);
220 log_err("fio: select() call in helper thread failed: %s",
224 if (FD_ISSET(fd, &rfds))
225 read_from_pipe(fd, &action, sizeof(action));
226 if (timerfd >= 0 && FD_ISSET(timerfd, &rfds)) {
227 res = read(timerfd, &exp, sizeof(exp));
228 assert(res == sizeof(exp));
234 * Verify whether or not timer @it has expired. If timer @it has expired, call
235 * @it->func(). @now is the current time. @msec_to_next_event is an
236 * input/output parameter that represents the time until the next event.
238 static int eval_timer(struct interval_timer *it, const struct timespec *now,
239 unsigned int *msec_to_next_event)
244 /* interval == 0 means that the timer is disabled. */
245 if (it->interval_ms == 0)
248 delta_ms = rel_time_since(now, &it->expires);
249 expired = delta_ms <= sleep_accuracy_ms;
251 timespec_add_msec(&it->expires, it->interval_ms);
252 delta_ms = rel_time_since(now, &it->expires);
253 if (delta_ms < it->interval_ms - sleep_accuracy_ms ||
254 delta_ms > it->interval_ms + sleep_accuracy_ms) {
255 dprint(FD_HELPERTHREAD,
256 "%s: delta = %" PRIi64 " <> %u. Clock jump?\n",
257 it->name, delta_ms, it->interval_ms);
258 delta_ms = it->interval_ms;
260 timespec_add_msec(&it->expires, it->interval_ms);
263 *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
264 return expired ? it->func() : 0;
267 static void *helper_thread_main(void *data)
269 struct helper_data *hd = data;
270 unsigned int msec_to_next_event, next_log;
271 struct interval_timer timer[] = {
274 .interval_ms = DISK_UTIL_MSEC,
275 .func = update_io_ticks,
278 .name = "status_interval",
279 .interval_ms = status_interval,
280 .func = __show_running_run_stats,
283 .name = "steadystate",
284 .interval_ms = steadystate_enabled ? ss_check_interval :
286 .func = steadystate_check,
293 os_clk_tck(&clk_tck);
295 dprint(FD_HELPERTHREAD, "clk_tck = %ld\n", clk_tck);
297 sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
299 #ifdef CONFIG_HAVE_TIMERFD_CREATE
300 timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
301 assert(timerfd >= 0);
302 sleep_accuracy_ms = 1;
305 sk_out_assign(hd->sk_out);
307 /* Let another thread handle signals. */
310 fio_get_mono_time(&ts);
311 msec_to_next_event = reset_timers(timer, FIO_ARRAY_SIZE(timer), &ts);
313 fio_sem_up(hd->startup_sem);
315 while (!ret && !hd->exit) {
319 action = wait_for_action(hd->pipe[0], msec_to_next_event);
320 if (action == A_EXIT)
323 fio_get_mono_time(&ts);
325 msec_to_next_event = INT_MAX;
327 if (action == A_RESET)
328 msec_to_next_event = reset_timers(timer,
329 FIO_ARRAY_SIZE(timer), &ts);
331 for (i = 0; i < FIO_ARRAY_SIZE(timer); ++i)
332 ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
334 if (action == A_DO_STAT)
335 __show_running_run_stats();
337 next_log = calc_log_samples();
339 next_log = DISK_UTIL_MSEC;
341 msec_to_next_event = min(next_log, msec_to_next_event);
342 dprint(FD_HELPERTHREAD,
343 "next_log: %u, msec_to_next_event: %u\n",
344 next_log, msec_to_next_event);
347 print_thread_status();
355 fio_writeout_logs(false);
362 * Connect two sockets to each other to emulate the pipe() system call on Windows.
364 int pipe_over_loopback(int fd[2])
366 struct sockaddr_in addr = { .sin_family = AF_INET };
367 socklen_t len = sizeof(addr);
370 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
374 fd[0] = socket(AF_INET, SOCK_STREAM, 0);
377 fd[1] = socket(AF_INET, SOCK_STREAM, 0);
380 res = bind(fd[0], (struct sockaddr *)&addr, len);
383 res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
386 res = listen(fd[0], 1);
389 res = connect(fd[1], (struct sockaddr *)&addr, len);
392 res = accept(fd[0], NULL, NULL);
409 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
411 struct helper_data *hd;
414 hd = scalloc(1, sizeof(*hd));
421 #if defined(CONFIG_PIPE2)
422 ret = pipe2(hd->pipe, O_CLOEXEC);
423 #elif defined(CONFIG_PIPE)
424 ret = pipe(hd->pipe);
426 ret = pipe_over_loopback(hd->pipe);
431 ret = make_nonblocking(hd->pipe[0]);
434 hd->startup_sem = startup_sem;
436 DRD_IGNORE_VAR(helper_data);
438 ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
440 log_err("Can't create helper thread: %s\n", strerror(ret));
446 dprint(FD_MUTEX, "wait on startup_sem\n");
447 fio_sem_down(startup_sem);
448 dprint(FD_MUTEX, "done waiting on startup_sem\n");