helper_thread.c

   1 #ifdef CONFIG_VALGRIND_DEV
   2 #include <valgrind/drd.h>
   3 #else
   4 #define DRD_IGNORE_VAR(x) do { } while (0)
   5 #endif
   6
   7 #include "fio.h"
   8 #include "smalloc.h"
   9 #include "helper_thread.h"
  10 #include "steadystate.h"
  11 #include "pshared.h"
  12
  13 enum action {
  14         A_EXIT          = 1,
  15         A_RESET         = 2,
  16         A_DO_STAT       = 3,
  17 };
  18
  19 static struct helper_data {
  20         volatile int exit;
  21         int pipe[2]; /* 0: read end; 1: write end. */
  22         struct sk_out *sk_out;
  23         pthread_t thread;
  24         struct fio_sem *startup_sem;
  25 } *helper_data;
  26
  27 void helper_thread_destroy(void)
  28 {
  29         if (!helper_data)
  30                 return;
  31
  32         close(helper_data->pipe[0]);
  33         close(helper_data->pipe[1]);
  34         sfree(helper_data);
  35 }
  36
  37 #ifdef _WIN32
  38 static void sock_init(void)
  39 {
  40         WSADATA wsaData;
  41         int res;
  42
  43         /* It is allowed to call WSAStartup() more than once. */
  44         res = WSAStartup(MAKEWORD(2, 2), &wsaData);
  45         assert(res == 0);
  46 }
  47
  48 static int make_nonblocking(int fd)
  49 {
  50         unsigned long arg = 1;
  51
  52         return ioctlsocket(fd, FIONBIO, &arg);
  53 }
  54
  55 static int write_to_pipe(int fd, const void *buf, size_t len)
  56 {
  57         return send(fd, buf, len, 0);
  58 }
  59
  60 static int read_from_pipe(int fd, void *buf, size_t len)
  61 {
  62         return recv(fd, buf, len, 0);
  63 }
  64 #else
  65 static void sock_init(void)
  66 {
  67 }
  68
  69 static int make_nonblocking(int fd)
  70 {
  71         return fcntl(fd, F_SETFL, O_NONBLOCK);
  72 }
  73
  74 static int write_to_pipe(int fd, const void *buf, size_t len)
  75 {
  76         return write(fd, buf, len);
  77 }
  78
  79 static int read_from_pipe(int fd, void *buf, size_t len)
  80 {
  81         return read(fd, buf, len);
  82 }
  83 #endif
  84
  85 static void submit_action(enum action a)
  86 {
  87         const char data = a;
  88         int ret;
  89
  90         if (!helper_data)
  91                 return;
  92
  93         ret = write_to_pipe(helper_data->pipe[1], &data, sizeof(data));
  94         assert(ret == 1);
  95 }
  96
  97 void helper_reset(void)
  98 {
  99         submit_action(A_RESET);
 100 }
 101
 102 /*
 103  * May be invoked in signal handler context and hence must only call functions
 104  * that are async-signal-safe. See also
 105  * https://pubs.opengroup.org/onlinepubs/9699919799/functions/V2_chap02.html#tag_15_04_03.
 106  */
 107 void helper_do_stat(void)
 108 {
 109         submit_action(A_DO_STAT);
 110 }
 111
 112 bool helper_should_exit(void)
 113 {
 114         if (!helper_data)
 115                 return true;
 116
 117         return helper_data->exit;
 118 }
 119
 120 void helper_thread_exit(void)
 121 {
 122         if (!helper_data)
 123                 return;
 124
 125         helper_data->exit = 1;
 126         submit_action(A_EXIT);
 127         pthread_join(helper_data->thread, NULL);
 128 }
 129
 130 static void *helper_thread_main(void *data)
 131 {
 132         struct helper_data *hd = data;
 133         unsigned int msec_to_next_event, next_log, next_ss = STEADYSTATE_MSEC;
 134         struct timespec ts, last_du, last_ss;
 135         char action;
 136         int ret = 0;
 137
 138         sk_out_assign(hd->sk_out);
 139
 140 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
 141         clock_gettime(CLOCK_MONOTONIC, &ts);
 142 #else
 143         clock_gettime(CLOCK_REALTIME, &ts);
 144 #endif
 145         memcpy(&last_du, &ts, sizeof(ts));
 146         memcpy(&last_ss, &ts, sizeof(ts));
 147
 148         fio_sem_up(hd->startup_sem);
 149
 150         msec_to_next_event = DISK_UTIL_MSEC;
 151         while (!ret && !hd->exit) {
 152                 uint64_t since_du, since_ss = 0;
 153                 struct timeval timeout = {
 154                         .tv_sec  = DISK_UTIL_MSEC / 1000,
 155                         .tv_usec = (DISK_UTIL_MSEC % 1000) * 1000,
 156                 };
 157                 fd_set rfds, efds;
 158
 159                 timespec_add_msec(&ts, msec_to_next_event);
 160
 161                 if (read_from_pipe(hd->pipe[0], &action, sizeof(action)) < 0) {
 162                         FD_ZERO(&rfds);
 163                         FD_SET(hd->pipe[0], &rfds);
 164                         FD_ZERO(&efds);
 165                         FD_SET(hd->pipe[0], &efds);
 166                         select(1, &rfds, NULL, &efds, &timeout);
 167                         if (read_from_pipe(hd->pipe[0], &action, sizeof(action)) <
 168                             0)
 169                                 action = 0;
 170                 }
 171
 172 #ifdef CONFIG_PTHREAD_CONDATTR_SETCLOCK
 173                 clock_gettime(CLOCK_MONOTONIC, &ts);
 174 #else
 175                 clock_gettime(CLOCK_REALTIME, &ts);
 176 #endif
 177
 178                 if (action == A_RESET) {
 179                         last_du = ts;
 180                         last_ss = ts;
 181                 }
 182
 183                 since_du = mtime_since(&last_du, &ts);
 184                 if (since_du >= DISK_UTIL_MSEC || DISK_UTIL_MSEC - since_du < 10) {
 185                         ret = update_io_ticks();
 186                         timespec_add_msec(&last_du, DISK_UTIL_MSEC);
 187                         msec_to_next_event = DISK_UTIL_MSEC;
 188                         if (since_du >= DISK_UTIL_MSEC)
 189                                 msec_to_next_event -= (since_du - DISK_UTIL_MSEC);
 190                 } else
 191                         msec_to_next_event = DISK_UTIL_MSEC - since_du;
 192
 193                 if (action == A_DO_STAT)
 194                         __show_running_run_stats();
 195
 196                 next_log = calc_log_samples();
 197                 if (!next_log)
 198                         next_log = DISK_UTIL_MSEC;
 199
 200                 if (steadystate_enabled) {
 201                         since_ss = mtime_since(&last_ss, &ts);
 202                         if (since_ss >= STEADYSTATE_MSEC || STEADYSTATE_MSEC - since_ss < 10) {
 203                                 steadystate_check();
 204                                 timespec_add_msec(&last_ss, since_ss);
 205                                 if (since_ss > STEADYSTATE_MSEC)
 206                                         next_ss = STEADYSTATE_MSEC - (since_ss - STEADYSTATE_MSEC);
 207                                 else
 208                                         next_ss = STEADYSTATE_MSEC;
 209                         } else
 210                                 next_ss = STEADYSTATE_MSEC - since_ss;
 211                 }
 212
 213                 msec_to_next_event = min(min(next_log, msec_to_next_event), next_ss);
 214                 dprint(FD_HELPERTHREAD, "since_ss: %llu, next_ss: %u, next_log: %u, msec_to_next_event: %u\n", (unsigned long long)since_ss, next_ss, next_log, msec_to_next_event);
 215
 216                 if (!is_backend)
 217                         print_thread_status();
 218         }
 219
 220         fio_writeout_logs(false);
 221
 222         sk_out_drop();
 223         return NULL;
 224 }
 225
 226 /*
 227  * Connect two sockets to each other to emulate the pipe() system call on Windows.
 228  */
 229 int pipe_over_loopback(int fd[2])
 230 {
 231         struct sockaddr_in addr = { .sin_family = AF_INET };
 232         socklen_t len = sizeof(addr);
 233         int res;
 234
 235         addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
 236
 237         sock_init();
 238
 239         fd[0] = socket(AF_INET, SOCK_STREAM, 0);
 240         if (fd[0] < 0)
 241                 goto err;
 242         fd[1] = socket(AF_INET, SOCK_STREAM, 0);
 243         if (fd[1] < 0)
 244                 goto close_fd_0;
 245         res = bind(fd[0], (struct sockaddr *)&addr, len);
 246         if (res < 0)
 247                 goto close_fd_1;
 248         res = getsockname(fd[0], (struct sockaddr *)&addr, &len);
 249         if (res < 0)
 250                 goto close_fd_1;
 251         res = listen(fd[0], 1);
 252         if (res < 0)
 253                 goto close_fd_1;
 254         res = connect(fd[1], (struct sockaddr *)&addr, len);
 255         if (res < 0)
 256                 goto close_fd_1;
 257         res = accept(fd[0], NULL, NULL);
 258         if (res < 0)
 259                 goto close_fd_1;
 260         close(fd[0]);
 261         fd[0] = res;
 262         return 0;
 263
 264 close_fd_1:
 265         close(fd[1]);
 266
 267 close_fd_0:
 268         close(fd[0]);
 269
 270 err:
 271         return -1;
 272 }
 273
 274 int helper_thread_create(struct fio_sem *startup_sem, struct sk_out *sk_out)
 275 {
 276         struct helper_data *hd;
 277         int ret;
 278
 279         hd = scalloc(1, sizeof(*hd));
 280
 281         setup_disk_util();
 282         steadystate_setup();
 283
 284         hd->sk_out = sk_out;
 285
 286 #if defined(CONFIG_PIPE2)
 287         ret = pipe2(hd->pipe, O_CLOEXEC);
 288 #elif defined(CONFIG_PIPE)
 289         ret = pipe(hd->pipe);
 290 #else
 291         ret = pipe_over_loopback(hd->pipe);
 292 #endif
 293         if (ret)
 294                 return 1;
 295
 296         ret = make_nonblocking(hd->pipe[0]);
 297         assert(ret >= 0);
 298
 299         hd->startup_sem = startup_sem;
 300
 301         DRD_IGNORE_VAR(helper_data);
 302
 303         ret = pthread_create(&hd->thread, NULL, helper_thread_main, hd);
 304         if (ret) {
 305                 log_err("Can't create helper thread: %s\n", strerror(ret));
 306                 return 1;
 307         }
 308
 309         helper_data = hd;
 310
 311         dprint(FD_MUTEX, "wait on startup_sem\n");
 312         fio_sem_down(startup_sem);
 313         dprint(FD_MUTEX, "done waiting on startup_sem\n");
 314         return 0;
 315 }