3 #ifdef CONFIG_VALGRIND_DEV
4 #include <valgrind/drd.h>
6 #define DRD_IGNORE_VAR(x) do { } while (0)
11 #include "helper_thread.h"
12 #include "steadystate.h"
15 static int sleep_accuracy_ms;
23 static struct helper_data {
25 int pipe[2]; /* 0: read end; 1: write end. */
26 struct sk_out *sk_out;
28 struct fio_sem *startup_sem;
31 struct interval_timer {
33 struct timespec expires;
38 void helper_thread_destroy(void)
43 close(helper_data->pipe[0]);
44 close(helper_data->pipe[1]);
49 static void sock_init(void)
54 /* It is allowed to call WSAStartup() more than once. */
55 res = WSAStartup(MAKEWORD(2, 2), &wsaData);
59 static int make_nonblocking(int fd)
61 unsigned long arg = 1;
63 return ioctlsocket(fd, FIONBIO, &arg);
66 static int write_to_pipe(int fd, const void *buf, size_t len)
68 return send(fd, buf, len, 0);
71 static int read_from_pipe(int fd, void *buf, size_t len)
73 return recv(fd, buf, len, 0);
76 static void sock_init(void)
80 static int make_nonblocking(int fd)
82 return fcntl(fd, F_SETFL, O_NONBLOCK);
85 static int write_to_pipe(int fd, const void *buf, size_t len)
87 return write(fd, buf, len);
90 static int read_from_pipe(int fd, void *buf, size_t len)
92 return read(fd, buf, len);
96 static void block_signals(void)
98 #ifdef HAVE_PTHREAD_SIGMASK
101 ret = pthread_sigmask(SIG_UNBLOCK, NULL, &sigmask);
103 ret = pthread_sigmask(SIG_BLOCK, &sigmask, NULL);
108 static void submit_action(enum action a)
116 ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
120 void helper_reset(void)
122 submit_action(A_RESET);
126 * May be invoked in signal handler context and hence must only call functions
127 * that are async-signal-safe. See also
128 * https://pubs.opengroup.org/onlinepubs/9699919799/functions/V2_chap02.html#tag_15_04_03.
130 void helper_do_stat(void)
132 submit_action(A_DO_STAT);
135 bool helper_should_exit(void)
140 return helper_data->exit;
143 void helper_thread_exit(void)
148 helper_data->exit = 1;
149 submit_action(A_EXIT);
150 pthread_join(helper_data->thread, NULL);
153 /* Resets timers and returns the time in milliseconds until the next event. */
154 static int reset_timers(struct interval_timer timer[], int num_timers,
155 struct timespec *now)
157 uint32_t msec_to_next_event = INT_MAX;
160 for (i = 0; i < num_timers; ++i) {
161 timer[i].expires = *now;
162 timespec_add_msec(&timer[i].expires, timer[i].interval_ms);
163 msec_to_next_event = min_not_zero(msec_to_next_event,
164 timer[i].interval_ms);
167 return msec_to_next_event;
171 * Waits for an action from fd during at least timeout_ms. `fd` must be in
174 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
176 struct timeval timeout = {
177 .tv_sec = timeout_ms / 1000,
178 .tv_usec = (timeout_ms % 1000) * 1000,
184 res = read_from_pipe(fd, &action, sizeof(action));
185 if (res > 0 || timeout_ms == 0)
191 res = select(fd + 1, &rfds, NULL, &efds, &timeout);
193 log_err("fio: select() call in helper thread failed: %s",
197 if (FD_ISSET(fd, &rfds))
198 read_from_pipe(fd, &action, sizeof(action));
203 * Verify whether or not timer @it has expired. If timer @it has expired, call
204 * @it->func(). @now is the current time. @msec_to_next_event is an
205 * input/output parameter that represents the time until the next event.
207 static int eval_timer(struct interval_timer *it, const struct timespec *now,
208 unsigned int *msec_to_next_event)
213 /* interval == 0 means that the timer is disabled. */
214 if (it->interval_ms == 0)
217 delta_ms = rel_time_since(now, &it->expires);
218 expired = delta_ms <= sleep_accuracy_ms;
220 timespec_add_msec(&it->expires, it->interval_ms);
221 delta_ms = rel_time_since(now, &it->expires);
222 if (delta_ms < it->interval_ms - sleep_accuracy_ms ||
223 delta_ms > it->interval_ms + sleep_accuracy_ms) {
224 dprint(FD_HELPERTHREAD,
225 "%s: delta = %" PRIi64 " <> %u. Clock jump?\n",
226 it->name, delta_ms, it->interval_ms);
227 delta_ms = it->interval_ms;
229 timespec_add_msec(&it->expires, it->interval_ms);
232 *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
233 return expired ? it->func() : 0;
236 static void *helper_thread_main(void *data)
238 struct helper_data *hd = data;
239 unsigned int msec_to_next_event, next_log;
240 struct interval_timer timer[] = {
243 .interval_ms = DISK_UTIL_MSEC,
244 .func = update_io_ticks,
247 .name = "status_interval",
248 .interval_ms = status_interval,
249 .func = __show_running_run_stats,
252 .name = "steadystate",
253 .interval_ms = steadystate_enabled ? STEADYSTATE_MSEC :
255 .func = steadystate_check,
259 int clk_tck, ret = 0;
262 clk_tck = sysconf(_SC_CLK_TCK);
265 * The timer frequence is variable on Windows. Instead of trying to
266 * query it, use 64 Hz, the clock frequency lower bound. See also
267 * https://carpediemsystems.co.uk/2019/07/18/windows-system-timer-granularity/.
271 dprint(FD_HELPERTHREAD, "clk_tck = %d\n", clk_tck);
273 sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
275 sk_out_assign(hd->sk_out);
277 /* Let another thread handle signals. */
280 fio_get_mono_time(&ts);
281 msec_to_next_event = reset_timers(timer, ARRAY_SIZE(timer), &ts);
283 fio_sem_up(hd->startup_sem);
285 while (!ret && !hd->exit) {
289 action = wait_for_action(hd->pipe[0], msec_to_next_event);
290 if (action == A_EXIT)
293 fio_get_mono_time(&ts);
295 msec_to_next_event = INT_MAX;
297 if (action == A_RESET)
298 msec_to_next_event = reset_timers(timer,
299 ARRAY_SIZE(timer), &ts);
301 for (i = 0; i < ARRAY_SIZE(timer); ++i)
302 ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
304 if (action == A_DO_STAT)
305 __show_running_run_stats();
307 next_log = calc_log_samples();
309 next_log = DISK_UTIL_MSEC;
311 msec_to_next_event = min(next_log, msec_to_next_event);
312 dprint(FD_HELPERTHREAD,
313 "next_log: %u, msec_to_next_event: %u\n",
314 next_log, msec_to_next_event);
317 print_thread_status();
320 fio_writeout_logs(false);
327 * Connect two sockets to each other to emulate the pipe() system call on Windows.
329 int pipe_over_loopback(int fd[2])
331 struct sockaddr_in addr = { .sin_family = AF_INET };
332 socklen_t len = sizeof(addr);
335 addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
339 fd[0] = socket(AF_INET, SOCK_STREAM, 0);
342 fd[1] = socket(AF_INET, SOCK_STREAM, 0);
345 res = bind(fd[0], (struct sockaddr *)&addr, len);
348 res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
351 res = listen(fd[0], 1);
354 res = connect(fd[1], (struct sockaddr *)&addr, len);
357 res = accept(fd[0], NULL, NULL);
374 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
376 struct helper_data *hd;
379 hd = scalloc(1, sizeof(*hd));
386 #if defined(CONFIG_PIPE2)
387 ret = pipe2(hd->pipe, O_CLOEXEC);
388 #elif defined(CONFIG_PIPE)
389 ret = pipe(hd->pipe);
391 ret = pipe_over_loopback(hd->pipe);
396 ret = make_nonblocking(hd->pipe[0]);
399 hd->startup_sem = startup_sem;
401 DRD_IGNORE_VAR(helper_data);
403 ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
405 log_err("Can't create helper thread: %s\n", strerror(ret));
411 dprint(FD_MUTEX, "wait on startup_sem\n");
412 fio_sem_down(startup_sem);
413 dprint(FD_MUTEX, "done waiting on startup_sem\n");