helper_thread.c

   1 #include <errno.h>
   2 #include <signal.h>
   3 #include <stdio.h>
   4 #include <string.h>
   5 #include <unistd.h>
   6 #ifdef CONFIG_HAVE_TIMERFD_CREATE
   7 #include <sys/timerfd.h>
   8 #endif
   9 #ifdef CONFIG_VALGRIND_DEV
  10 #include <valgrind/drd.h>
  11 #else
  12 #define DRD_IGNORE_VAR(x) do { } while (0)
  13 #endif
  14
  15 #ifdef WIN32
  16 #include "os/os-windows.h"
  17 #endif
  18
  19 #include "fio.h"
  20 #include "smalloc.h"
  21 #include "helper_thread.h"
  22 #include "steadystate.h"
  23 #include "pshared.h"
  24
  25 static int sleep_accuracy_ms;
  26 static int timerfd = -1;
  27
  28 enum action {
  29         A_EXIT          = 1,
  30         A_RESET         = 2,
  31         A_DO_STAT       = 3,
  32 };
  33
  34 static struct helper_data {
  35         volatile int exit;
  36         int pipe[2]; /* 0: read end; 1: write end. */
  37         struct sk_out *sk_out;
  38         pthread_t thread;
  39         struct fio_sem *startup_sem;
  40 } *helper_data;
  41
  42 struct interval_timer {
  43         const char      *name;
  44         struct timespec expires;
  45         uint32_t        interval_ms;
  46         int             (*func)(void);
  47 };
  48
  49 void helper_thread_destroy(void)
  50 {
  51         if (!helper_data)
  52                 return;
  53
  54         close(helper_data->pipe[0]);
  55         close(helper_data->pipe[1]);
  56         sfree(helper_data);
  57 }
  58
  59 #ifdef _WIN32
  60 static void sock_init(void)
  61 {
  62         WSADATA wsaData;
  63         int res;
  64
  65         /* It is allowed to call WSAStartup() more than once. */
  66         res = WSAStartup(MAKEWORD(2, 2), &wsaData);
  67         assert(res == 0);
  68 }
  69
  70 static int make_nonblocking(int fd)
  71 {
  72         unsigned long arg = 1;
  73
  74         return ioctlsocket(fd, FIONBIO, &arg);
  75 }
  76
  77 static int write_to_pipe(int fd, const void *buf, size_t len)
  78 {
  79         return send(fd, buf, len, 0);
  80 }
  81
  82 static int read_from_pipe(int fd, void *buf, size_t len)
  83 {
  84         return recv(fd, buf, len, 0);
  85 }
  86 #else
  87 static void sock_init(void)
  88 {
  89 }
  90
  91 static int make_nonblocking(int fd)
  92 {
  93         return fcntl(fd, F_SETFL, O_NONBLOCK);
  94 }
  95
  96 static int write_to_pipe(int fd, const void *buf, size_t len)
  97 {
  98         return write(fd, buf, len);
  99 }
 100
 101 static int read_from_pipe(int fd, void *buf, size_t len)
 102 {
 103         return read(fd, buf, len);
 104 }
 105 #endif
 106
 107 static void block_signals(void)
 108 {
 109 #ifdef CONFIG_PTHREAD_SIGMASK
 110         sigset_t sigmask;
 111
 112         int ret;
 113
 114         ret = pthread_sigmask(SIG_UNBLOCK, NULL, &sigmask);
 115         assert(ret == 0);
 116         ret = pthread_sigmask(SIG_BLOCK, &sigmask, NULL);
 117 #endif
 118 }
 119
 120 static void submit_action(enum action a)
 121 {
 122         const char data = a;
 123         int ret;
 124
 125         if (!helper_data)
 126                 return;
 127
 128         ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
 129         if (ret != 1) {
 130                 log_err("failed to write action into pipe, err %i:%s", errno, strerror(errno));
 131                 assert(0);
 132         }
 133 }
 134
 135 void helper_reset(void)
 136 {
 137         submit_action(A_RESET);
 138 }
 139
 140 /*
 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.
 144  */
 145 void helper_do_stat(void)
 146 {
 147         submit_action(A_DO_STAT);
 148 }
 149
 150 bool helper_should_exit(void)
 151 {
 152         if (!helper_data)
 153                 return true;
 154
 155         return helper_data->exit;
 156 }
 157
 158 void helper_thread_exit(void)
 159 {
 160         if (!helper_data)
 161                 return;
 162
 163         helper_data->exit = 1;
 164         submit_action(A_EXIT);
 165         pthread_join(helper_data->thread, NULL);
 166 }
 167
 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)
 171 {
 172         uint32_t msec_to_next_event = INT_MAX;
 173         int i;
 174
 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);
 180         }
 181
 182         return msec_to_next_event;
 183 }
 184
 185 /*
 186  * Waits for an action from fd during at least timeout_ms. `fd` must be in
 187  * non-blocking mode.
 188  */
 189 static uint8_t wait_for_action(int fd, unsigned int timeout_ms)
 190 {
 191         struct timeval timeout = {
 192                 .tv_sec  = timeout_ms / 1000,
 193                 .tv_usec = (timeout_ms % 1000) * 1000,
 194         };
 195         fd_set rfds, efds;
 196         uint8_t action = 0;
 197         uint64_t exp;
 198         int res;
 199
 200         res = read_from_pipe(fd, &action, sizeof(action));
 201         if (res > 0 || timeout_ms == 0)
 202                 return action;
 203         FD_ZERO(&rfds);
 204         FD_SET(fd, &rfds);
 205         FD_ZERO(&efds);
 206         FD_SET(fd, &efds);
 207 #ifdef CONFIG_HAVE_TIMERFD_CREATE
 208         {
 209                 /*
 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.
 214                  */
 215                 struct itimerspec delta = {};
 216
 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);
 220                 assert(res == 0);
 221                 FD_SET(timerfd, &rfds);
 222         }
 223 #endif
 224         res = select(max(fd, timerfd) + 1, &rfds, NULL, &efds,
 225                      timerfd >= 0 ? NULL : &timeout);
 226         if (res < 0) {
 227                 log_err("fio: select() call in helper thread failed: %s",
 228                         strerror(errno));
 229                 return A_EXIT;
 230         }
 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));
 236         }
 237         return action;
 238 }
 239
 240 /*
 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.
 244  */
 245 static int eval_timer(struct interval_timer *it, const struct timespec *now,
 246                       unsigned int *msec_to_next_event)
 247 {
 248         int64_t delta_ms;
 249         bool expired;
 250
 251         /* interval == 0 means that the timer is disabled. */
 252         if (it->interval_ms == 0)
 253                 return 0;
 254
 255         delta_ms = rel_time_since(now, &it->expires);
 256         expired = delta_ms <= sleep_accuracy_ms;
 257         if (expired) {
 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;
 266                         it->expires = *now;
 267                         timespec_add_msec(&it->expires, it->interval_ms);
 268                 }
 269         }
 270         *msec_to_next_event = min((unsigned int)delta_ms, *msec_to_next_event);
 271         return expired ? it->func() : 0;
 272 }
 273
 274 static void *helper_thread_main(void *data)
 275 {
 276         struct helper_data *hd = data;
 277         unsigned int msec_to_next_event, next_log;
 278         struct interval_timer timer[] = {
 279                 {
 280                         .name = "disk_util",
 281                         .interval_ms = DISK_UTIL_MSEC,
 282                         .func = update_io_ticks,
 283                 },
 284                 {
 285                         .name = "status_interval",
 286                         .interval_ms = status_interval,
 287                         .func = __show_running_run_stats,
 288                 },
 289                 {
 290                         .name = "steadystate",
 291                         .interval_ms = steadystate_enabled ? ss_check_interval :
 292                                 0,
 293                         .func = steadystate_check,
 294                 }
 295         };
 296         struct timespec ts;
 297         long clk_tck;
 298         int ret = 0;
 299
 300         os_clk_tck(&clk_tck);
 301
 302         dprint(FD_HELPERTHREAD, "clk_tck = %ld\n", clk_tck);
 303         assert(clk_tck > 0);
 304         sleep_accuracy_ms = (1000 + clk_tck - 1) / clk_tck;
 305
 306 #ifdef CONFIG_HAVE_TIMERFD_CREATE
 307         timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK);
 308         assert(timerfd >= 0);
 309         sleep_accuracy_ms = 1;
 310 #endif
 311
 312         sk_out_assign(hd->sk_out);
 313
 314         /* Let another thread handle signals. */
 315         block_signals();
 316
 317         fio_get_mono_time(&ts);
 318         msec_to_next_event = reset_timers(timer, FIO_ARRAY_SIZE(timer), &ts);
 319
 320         fio_sem_up(hd->startup_sem);
 321
 322         while (!ret && !hd->exit) {
 323                 uint8_t action;
 324                 int i;
 325
 326                 action = wait_for_action(hd->pipe[0], msec_to_next_event);
 327                 if (action == A_EXIT)
 328                         break;
 329
 330                 fio_get_mono_time(&ts);
 331
 332                 msec_to_next_event = INT_MAX;
 333
 334                 if (action == A_RESET)
 335                         msec_to_next_event = reset_timers(timer,
 336                                                 FIO_ARRAY_SIZE(timer), &ts);
 337
 338                 for (i = 0; i < FIO_ARRAY_SIZE(timer); ++i)
 339                         ret = eval_timer(&timer[i], &ts, &msec_to_next_event);
 340
 341                 if (action == A_DO_STAT)
 342                         __show_running_run_stats();
 343
 344                 next_log = calc_log_samples();
 345                 if (!next_log)
 346                         next_log = DISK_UTIL_MSEC;
 347
 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);
 352
 353                 if (!is_backend)
 354                         print_thread_status();
 355         }
 356
 357         if (timerfd >= 0) {
 358                 close(timerfd);
 359                 timerfd = -1;
 360         }
 361
 362         fio_writeout_logs(false);
 363
 364         sk_out_drop();
 365         return NULL;
 366 }
 367
 368 /*
 369  * Connect two sockets to each other to emulate the pipe() system call on Windows.
 370  */
 371 int pipe_over_loopback(int fd[2])
 372 {
 373         struct sockaddr_in addr = { .sin_family = AF_INET };
 374         socklen_t len = sizeof(addr);
 375         int res;
 376
 377         addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
 378
 379         sock_init();
 380
 381         fd[0] = socket(AF_INET, SOCK_STREAM, 0);
 382         if (fd[0] < 0)
 383                 goto err;
 384         fd[1] = socket(AF_INET, SOCK_STREAM, 0);
 385         if (fd[1] < 0)
 386                 goto close_fd_0;
 387         res = bind(fd[0], (struct sockaddr *)&addr, len);
 388         if (res < 0)
 389                 goto close_fd_1;
 390         res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
 391         if (res < 0)
 392                 goto close_fd_1;
 393         res = listen(fd[0], 1);
 394         if (res < 0)
 395                 goto close_fd_1;
 396         res = connect(fd[1], (struct sockaddr *)&addr, len);
 397         if (res < 0)
 398                 goto close_fd_1;
 399         res = accept(fd[0], NULL, NULL);
 400         if (res < 0)
 401                 goto close_fd_1;
 402         close(fd[0]);
 403         fd[0] = res;
 404         return 0;
 405
 406 close_fd_1:
 407         close(fd[1]);
 408
 409 close_fd_0:
 410         close(fd[0]);
 411
 412 err:
 413         return -1;
 414 }
 415
 416 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
 417 {
 418         struct helper_data *hd;
 419         int ret;
 420
 421         hd = scalloc(1, sizeof(*hd));
 422
 423         setup_disk_util();
 424         steadystate_setup();
 425
 426         hd->sk_out = sk_out;
 427
 428 #if defined(CONFIG_PIPE2)
 429         ret = pipe2(hd->pipe, O_CLOEXEC);
 430 #elif defined(CONFIG_PIPE)
 431         ret = pipe(hd->pipe);
 432 #else
 433         ret = pipe_over_loopback(hd->pipe);
 434 #endif
 435         if (ret)
 436                 return 1;
 437
 438         ret = make_nonblocking(hd->pipe[0]);
 439         assert(ret >= 0);
 440
 441         hd->startup_sem = startup_sem;
 442
 443         DRD_IGNORE_VAR(helper_data);
 444
 445         ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
 446         if (ret) {
 447                 log_err("Can't create helper thread: %s\n", strerror(ret));
 448                 return 1;
 449         }
 450
 451         helper_data = hd;
 452
 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");
 456         return 0;
 457 }