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 pthread_join(helper_data->thread, NULL);
167 /* Resets timers and returns the time in milliseconds until the next event. */
168 static int reset_timers(struct interval_timer timer[], int num_timers,
169 struct timespec *now)
171 uint32_t msec_to_next_event = INT_MAX;
174 for (i = 0; i < num_timers; ++i) {
175 timer[i].expires = *now;
176 timespec_add_msec(&timer[i].expires, timer[i].interval_ms);
177 msec_to_next_event = min_not_zero(msec_to_next_event,
178 timer[i].interval_ms);
181 return msec_to_next_event;
185 * Waits for an action from fd during at least timeout_ms. `fd` must be in
188 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
190 struct timeval timeout = {
191 .tv_sec = timeout_ms / 1000,
192 .tv_usec = (timeout_ms % 1000) * 1000,
199 res = read_from_pipe(fd, &action, sizeof(action));
200 if (res > 0 || timeout_ms == 0)
206 #ifdef CONFIG_HAVE_TIMERFD_CREATE
209 * If the timer frequency is 100 Hz, select() will round up
210 * `timeout` to the next multiple of 1 / 100 Hz = 10 ms. Hence
211 * use a high-resolution timer if possible to increase
212 * select() timeout accuracy.
214 struct itimerspec delta = {};
216 delta.it_value.tv_sec = timeout.tv_sec;
217 delta.it_value.tv_nsec = timeout.tv_usec * 1000;
218 res = timerfd_settime(timerfd, 0, &delta, NULL);
220 FD_SET(timerfd, &rfds);
223 res = select(max(fd, timerfd) + 1, &rfds, NULL, &efds,
224 timerfd >= 0 ? NULL : &timeout);
226 log_err("fio: select() call in helper thread failed: %s",
230 if (FD_ISSET(fd, &rfds))
231 read_from_pipe(fd, &action, sizeof(action));
232 if (timerfd >= 0 && FD_ISSET(timerfd, &rfds)) {
233 res = read(timerfd, &exp, sizeof(exp));
234 assert(res == sizeof(exp));
240 * Verify whether or not timer @it has expired. If timer @it has expired, call
241 * @it->func(). @now is the current time. @msec_to_next_event is an
242 * input/output parameter that represents the time until the next event.
244 static int eval_timer(struct interval_timer *it, const struct timespec *now,
245 unsigned int *msec_to_next_event)
250 /* interval == 0 means that the timer is disabled. */
251 if (it->interval_ms == 0)
254 delta_ms = rel_time_since(now, &it->expires);
255 expired = delta_ms <= sleep_accuracy_ms;
257 timespec_add_msec(&it->expires, it->interval_ms);
258 delta_ms = rel_time_since(now, &it->expires);
259 if (delta_ms < it->interval_ms - sleep_accuracy_ms ||
260 delta_ms > it->interval_ms + sleep_accuracy_ms) {
261 dprint(FD_HELPERTHREAD,
262 "%s: delta = %" PRIi64 " <> %u. Clock jump?\n",
263 it->name, delta_ms, it->interval_ms);
264 delta_ms = it->interval_ms;
266 timespec_add_msec(&it->expires, it->interval_ms);
269 *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
270 return expired ? it->func() : 0;
273 static void *helper_thread_main(void *data)
275 struct helper_data *hd = data;
276 unsigned int msec_to_next_event, next_log;
277 struct interval_timer timer[] = {
280 .interval_ms = DISK_UTIL_MSEC,
281 .func = update_io_ticks,
284 .name = "status_interval",
285 .interval_ms = status_interval,
286 .func = __show_running_run_stats,
289 .name = "steadystate",
290 .interval_ms = steadystate_enabled ? ss_check_interval :
292 .func = steadystate_check,
299 os_clk_tck(&clk_tck);
301 dprint(FD_HELPERTHREAD, "clk_tck = %ld\n", clk_tck);
303 sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
305 #ifdef CONFIG_HAVE_TIMERFD_CREATE
306 timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
307 assert(timerfd >= 0);
308 sleep_accuracy_ms = 1;
311 sk_out_assign(hd->sk_out);
313 /* Let another thread handle signals. */
316 fio_get_mono_time(&ts);
317 msec_to_next_event = reset_timers(timer, FIO_ARRAY_SIZE(timer), &ts);
319 fio_sem_up(hd->startup_sem);
321 while (!ret && !hd->exit) {
325 action = wait_for_action(hd->pipe[0], msec_to_next_event);
326 if (action == A_EXIT)
329 fio_get_mono_time(&ts);
331 msec_to_next_event = INT_MAX;
333 if (action == A_RESET)
334 msec_to_next_event = reset_timers(timer,
335 FIO_ARRAY_SIZE(timer), &ts);
337 for (i = 0; i < FIO_ARRAY_SIZE(timer); ++i)
338 ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
340 if (action == A_DO_STAT)
341 __show_running_run_stats();
343 next_log = calc_log_samples();
345 next_log = DISK_UTIL_MSEC;
347 msec_to_next_event = min(next_log, msec_to_next_event);
348 dprint(FD_HELPERTHREAD,
349 "next_log: %u, msec_to_next_event: %u\n",
350 next_log, msec_to_next_event);
353 print_thread_status();
361 fio_writeout_logs(false);
368 * Connect two sockets to each other to emulate the pipe() system call on Windows.
370 int pipe_over_loopback(int fd[2])
372 struct sockaddr_in addr = { .sin_family = AF_INET };
373 socklen_t len = sizeof(addr);
376 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
380 fd[0] = socket(AF_INET, SOCK_STREAM, 0);
383 fd[1] = socket(AF_INET, SOCK_STREAM, 0);
386 res = bind(fd[0], (struct sockaddr *)&addr, len);
389 res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
392 res = listen(fd[0], 1);
395 res = connect(fd[1], (struct sockaddr *)&addr, len);
398 res = accept(fd[0], NULL, NULL);
415 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
417 struct helper_data *hd;
420 hd = scalloc(1, sizeof(*hd));
427 #if defined(CONFIG_PIPE2)
428 ret = pipe2(hd->pipe, O_CLOEXEC);
429 #elif defined(CONFIG_PIPE)
430 ret = pipe(hd->pipe);
432 ret = pipe_over_loopback(hd->pipe);
437 ret = make_nonblocking(hd->pipe[0]);
440 hd->startup_sem = startup_sem;
442 DRD_IGNORE_VAR(helper_data);
444 ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
446 log_err("Can't create helper thread: %s\n", strerror(ret));
452 dprint(FD_MUTEX, "wait on startup_sem\n");
453 fio_sem_down(startup_sem);
454 dprint(FD_MUTEX, "done waiting on startup_sem\n");