6 #ifdef CONFIG_HAVE_TIMERFD_CREATE
7 #include <sys/timerfd.h>
9 #ifdef CONFIG_VALGRIND_DEV
10 #include <valgrind/drd.h>
12 #define DRD_IGNORE_VAR(x) do { } while (0)
16 #include "os/os-windows.h"
21 #include "helper_thread.h"
22 #include "steadystate.h"
25 static int sleep_accuracy_ms;
26 static int timerfd = -1;
34 static struct helper_data {
36 int pipe[2]; /* 0: read end; 1: write end. */
37 struct sk_out *sk_out;
39 struct fio_sem *startup_sem;
42 struct interval_timer {
44 struct timespec expires;
49 void helper_thread_destroy(void)
54 close(helper_data->pipe[0]);
55 close(helper_data->pipe[1]);
60 static void sock_init(void)
65 /* It is allowed to call WSAStartup() more than once. */
66 res = WSAStartup(MAKEWORD(2, 2), &wsaData);
70 static int make_nonblocking(int fd)
72 unsigned long arg = 1;
74 return ioctlsocket(fd, FIONBIO, &arg);
77 static int write_to_pipe(int fd, const void *buf, size_t len)
79 return send(fd, buf, len, 0);
82 static int read_from_pipe(int fd, void *buf, size_t len)
84 return recv(fd, buf, len, 0);
87 static void sock_init(void)
91 static int make_nonblocking(int fd)
93 return fcntl(fd, F_SETFL, O_NONBLOCK);
96 static int write_to_pipe(int fd, const void *buf, size_t len)
98 return write(fd, buf, len);
101 static int read_from_pipe(int fd, void *buf, size_t len)
103 return read(fd, buf, len);
107 static void block_signals(void)
109 #ifdef CONFIG_PTHREAD_SIGMASK
114 ret = pthread_sigmask(SIG_UNBLOCK, NULL, &sigmask);
116 ret = pthread_sigmask(SIG_BLOCK, &sigmask, NULL);
120 static void submit_action(enum action a)
128 ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
130 log_err("failed to write action into pipe, err %i:%s", errno, strerror(errno));
135 void helper_reset(void)
137 submit_action(A_RESET);
141 * May be invoked in signal handler context and hence must only call functions
142 * that are async-signal-safe. See also
143 * https://pubs.opengroup.org/onlinepubs/9699919799/functions/V2_chap02.html#tag_15_04_03.
145 void helper_do_stat(void)
147 submit_action(A_DO_STAT);
150 bool helper_should_exit(void)
155 return helper_data->exit;
158 void helper_thread_exit(void)
163 helper_data->exit = 1;
164 submit_action(A_EXIT);
165 pthread_join(helper_data->thread, NULL);
168 /* Resets timers and returns the time in milliseconds until the next event. */
169 static int reset_timers(struct interval_timer timer[], int num_timers,
170 struct timespec *now)
172 uint32_t msec_to_next_event = INT_MAX;
175 for (i = 0; i < num_timers; ++i) {
176 timer[i].expires = *now;
177 timespec_add_msec(&timer[i].expires, timer[i].interval_ms);
178 msec_to_next_event = min_not_zero(msec_to_next_event,
179 timer[i].interval_ms);
182 return msec_to_next_event;
186 * Waits for an action from fd during at least timeout_ms. `fd` must be in
189 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
191 struct timeval timeout = {
192 .tv_sec = timeout_ms / 1000,
193 .tv_usec = (timeout_ms % 1000) * 1000,
200 res = read_from_pipe(fd, &action, sizeof(action));
201 if (res > 0 || timeout_ms == 0)
207 #ifdef CONFIG_HAVE_TIMERFD_CREATE
210 * If the timer frequency is 100 Hz, select() will round up
211 * `timeout` to the next multiple of 1 / 100 Hz = 10 ms. Hence
212 * use a high-resolution timer if possible to increase
213 * select() timeout accuracy.
215 struct itimerspec delta = {};
217 delta.it_value.tv_sec = timeout.tv_sec;
218 delta.it_value.tv_nsec = timeout.tv_usec * 1000;
219 res = timerfd_settime(timerfd, 0, &delta, NULL);
221 FD_SET(timerfd, &rfds);
224 res = select(max(fd, timerfd) + 1, &rfds, NULL, &efds,
225 timerfd >= 0 ? NULL : &timeout);
227 log_err("fio: select() call in helper thread failed: %s",
231 if (FD_ISSET(fd, &rfds))
232 read_from_pipe(fd, &action, sizeof(action));
233 if (timerfd >= 0 && FD_ISSET(timerfd, &rfds)) {
234 res = read(timerfd, &exp, sizeof(exp));
235 assert(res == sizeof(exp));
241 * Verify whether or not timer @it has expired. If timer @it has expired, call
242 * @it->func(). @now is the current time. @msec_to_next_event is an
243 * input/output parameter that represents the time until the next event.
245 static int eval_timer(struct interval_timer *it, const struct timespec *now,
246 unsigned int *msec_to_next_event)
251 /* interval == 0 means that the timer is disabled. */
252 if (it->interval_ms == 0)
255 delta_ms = rel_time_since(now, &it->expires);
256 expired = delta_ms <= sleep_accuracy_ms;
258 timespec_add_msec(&it->expires, it->interval_ms);
259 delta_ms = rel_time_since(now, &it->expires);
260 if (delta_ms < it->interval_ms - sleep_accuracy_ms ||
261 delta_ms > it->interval_ms + sleep_accuracy_ms) {
262 dprint(FD_HELPERTHREAD,
263 "%s: delta = %" PRIi64 " <> %u. Clock jump?\n",
264 it->name, delta_ms, it->interval_ms);
265 delta_ms = it->interval_ms;
267 timespec_add_msec(&it->expires, it->interval_ms);
270 *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
271 return expired ? it->func() : 0;
274 static void *helper_thread_main(void *data)
276 struct helper_data *hd = data;
277 unsigned int msec_to_next_event, next_log;
278 struct interval_timer timer[] = {
281 .interval_ms = DISK_UTIL_MSEC,
282 .func = update_io_ticks,
285 .name = "status_interval",
286 .interval_ms = status_interval,
287 .func = __show_running_run_stats,
290 .name = "steadystate",
291 .interval_ms = steadystate_enabled ? ss_check_interval :
293 .func = steadystate_check,
300 os_clk_tck(&clk_tck);
302 dprint(FD_HELPERTHREAD, "clk_tck = %ld\n", clk_tck);
304 sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
306 #ifdef CONFIG_HAVE_TIMERFD_CREATE
307 timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
308 assert(timerfd >= 0);
309 sleep_accuracy_ms = 1;
312 sk_out_assign(hd->sk_out);
314 /* Let another thread handle signals. */
317 fio_get_mono_time(&ts);
318 msec_to_next_event = reset_timers(timer, FIO_ARRAY_SIZE(timer), &ts);
320 fio_sem_up(hd->startup_sem);
322 while (!ret && !hd->exit) {
326 action = wait_for_action(hd->pipe[0], msec_to_next_event);
327 if (action == A_EXIT)
330 fio_get_mono_time(&ts);
332 msec_to_next_event = INT_MAX;
334 if (action == A_RESET)
335 msec_to_next_event = reset_timers(timer,
336 FIO_ARRAY_SIZE(timer), &ts);
338 for (i = 0; i < FIO_ARRAY_SIZE(timer); ++i)
339 ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
341 if (action == A_DO_STAT)
342 __show_running_run_stats();
344 next_log = calc_log_samples();
346 next_log = DISK_UTIL_MSEC;
348 msec_to_next_event = min(next_log, msec_to_next_event);
349 dprint(FD_HELPERTHREAD,
350 "next_log: %u, msec_to_next_event: %u\n",
351 next_log, msec_to_next_event);
354 print_thread_status();
362 fio_writeout_logs(false);
369 * Connect two sockets to each other to emulate the pipe() system call on Windows.
371 int pipe_over_loopback(int fd[2])
373 struct sockaddr_in addr = { .sin_family = AF_INET };
374 socklen_t len = sizeof(addr);
377 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
381 fd[0] = socket(AF_INET, SOCK_STREAM, 0);
384 fd[1] = socket(AF_INET, SOCK_STREAM, 0);
387 res = bind(fd[0], (struct sockaddr *)&addr, len);
390 res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
393 res = listen(fd[0], 1);
396 res = connect(fd[1], (struct sockaddr *)&addr, len);
399 res = accept(fd[0], NULL, NULL);
416 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
418 struct helper_data *hd;
421 hd = scalloc(1, sizeof(*hd));
428 #if defined(CONFIG_PIPE2)
429 ret = pipe2(hd->pipe, O_CLOEXEC);
430 #elif defined(CONFIG_PIPE)
431 ret = pipe(hd->pipe);
433 ret = pipe_over_loopback(hd->pipe);
438 ret = make_nonblocking(hd->pipe[0]);
441 hd->startup_sem = startup_sem;
443 DRD_IGNORE_VAR(helper_data);
445 ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
447 log_err("Can't create helper thread: %s\n", strerror(ret));
453 dprint(FD_MUTEX, "wait on startup_sem\n");
454 fio_sem_down(startup_sem);
455 dprint(FD_MUTEX, "done waiting on startup_sem\n");