lustre/tests/iabf/iabf.c

   1 #include <stdbool.h>
   2 #include <stddef.h>
   3 #include <inttypes.h>
   4 #include <stdio.h>
   5 #include <stdlib.h>
   6 #include <errno.h>
   7 #include <limits.h>
   8 #include <string.h>
   9 #include <time.h>
  10 #include <assert.h>
  11 #include <pthread.h>
  12 #include <unistd.h>
  13 #include <sys/syscall.h>
  14 #include <sys/wait.h>
  15 #include "callvpe.h"
  16
  17 enum {
  18         NSEC_PER_SEC = 1000000000L,
  19 };
  20
  21 static int iabf_debug = false;
  22 static const char iabf_delim[] = "---";
  23 static __thread pid_t iabf_tid;
  24
  25 #define DEBUG(fmt, args...)                                             \
  26         do {                                                            \
  27                 if (iabf_debug)                                         \
  28                         fprintf(stderr, "DEBUG [%d] %s:%d: "fmt, iabf_tid, __func__, __LINE__, ##args); \
  29         } while (0)
  30
  31 #define DEBUG_B(x) DEBUG("%s = %s\n", #x, (x) ? "true" : "false")
  32 #define DEBUG_D(x) DEBUG("%s = %"PRIdMAX"\n", #x, (intmax_t)(x))
  33 #define DEBUG_P(x) DEBUG("%s = %p\n", #x, (x))
  34 #define DEBUG_S(x) DEBUG("%s = '%s'\n", #x, (x))
  35 #define DEBUG_U(x) DEBUG("%s = %"PRIuMAX"\n", #x, (uintmax_t)(x))
  36 #define DEBUG_X(x) DEBUG("%s = %"PRIxMAX"\n", #x, (uintmax_t)(x))
  37
  38 #define ERROR(fmt, args...)                                             \
  39         fprintf(stderr, "%s: "fmt, program_invocation_short_name, ##args)
  40
  41 #define FATAL(fmt, args...)                         \
  42         do {                                    \
  43                 ERROR("fatal: "fmt, ##args);        \
  44                 exit(EXIT_FAILURE);                 \
  45         } while (0)
  46
  47 #define xstrerror(e) strerror(abs(e))
  48
  49 static struct timespec
  50 timespec_sub(struct timespec a, struct timespec b)
  51 {
  52         struct timespec r = {
  53                 .tv_sec = a.tv_sec - b.tv_sec,
  54                 .tv_nsec = a.tv_nsec - b.tv_nsec,
  55         };
  56
  57         while (r.tv_nsec >= NSEC_PER_SEC) {
  58                 r.tv_sec++;
  59                 r.tv_nsec -= NSEC_PER_SEC;
  60         }
  61
  62         while (r.tv_nsec < 0) {
  63                 r.tv_sec--;
  64                 r.tv_nsec += NSEC_PER_SEC;
  65         }
  66
  67         return r;
  68 }
  69
  70 static struct timespec
  71 timespec_from_ns(long ns)
  72 {
  73         return (struct timespec) {
  74                 .tv_sec = ns / NSEC_PER_SEC,
  75                 .tv_nsec = ns % NSEC_PER_SEC,
  76         };
  77 }
  78
  79 static long timespec_to_ns(struct timespec tv)
  80 {
  81         return tv.tv_sec * NSEC_PER_SEC + tv.tv_nsec;
  82 }
  83
  84 struct iabf_control {
  85         char **ic_init;
  86         char **ic_fini;
  87         cpu_set_t *ic_affinity;
  88         long ic_delay_begin_ns;
  89         long ic_delay_end_ns;
  90         long ic_delay_step_ns;
  91         long ic_step_count;
  92         long ic_autotune_count;
  93
  94         pthread_barrier_t ic_barrier[2];
  95         int ic_should_stop;
  96 };
  97
  98 struct iabf_task {
  99         struct iabf_control *it_control;
 100         const char *it_name;
 101         pthread_t it_thread;
 102         struct timespec it_delay;
 103         struct timespec it_elapsed;
 104         char **it_argv;
 105 };
 106
 107 long iabf_getenvl(const char *name, long def)
 108 {
 109         const char *s = getenv(name);
 110         return s != NULL ? atol(s) : def;
 111 }
 112
 113 static void iabf_barrier_wait(struct iabf_control *ic, int which)
 114 {
 115         int rc;
 116
 117         assert(PTHREAD_BARRIER_SERIAL_THREAD == -1);
 118
 119         rc = pthread_barrier_wait(&ic->ic_barrier[which]);
 120         if (rc > 0)
 121                 FATAL("cannot wait on barrier: %s\n", xstrerror(rc));
 122 }
 123
 124 static void *iabf_task_thread(void *data)
 125 {
 126         struct iabf_task *it = data;
 127         struct iabf_control *ic = it->it_control;
 128         int rc;
 129
 130         iabf_tid = syscall(SYS_gettid);
 131
 132         assert(PTHREAD_BARRIER_SERIAL_THREAD == -1);
 133
 134         while (1) {
 135                 struct timespec ts[2];
 136                 pid_t pid, pid2;
 137                 int status;
 138
 139                 iabf_barrier_wait(ic, 0);
 140
 141                 DEBUG_D(ic->ic_should_stop);
 142                 if (ic->ic_should_stop)
 143                         break;
 144
 145                 rc = clock_nanosleep(CLOCK_MONOTONIC, 0, &it->it_delay, NULL);
 146                 if (rc != 0)
 147                         FATAL("%s: cannot sleep: %s\n", it->it_name, xstrerror(rc));
 148
 149                 rc = clock_gettime(CLOCK_MONOTONIC, &ts[0]);
 150                 if (rc != 0)
 151                         FATAL("%s: cannot get time: %s\n", it->it_name, xstrerror(errno));
 152
 153                 pid = fork();
 154                 if (pid < 0)
 155                         FATAL("%s: cannot fork: %s\n", it->it_name, strerror(errno));
 156
 157                 if (pid == 0) {
 158                         execvpe(it->it_argv[0], it->it_argv, environ);
 159                         _exit(127);
 160                 }
 161
 162                 pid2 = waitpid(pid, &status, 0);
 163                 if (pid2 < 0)
 164                         FATAL("%s: cannot wait for pid %d: %s\n", it->it_name, (int)pid, strerror(errno));
 165
 166                 rc = clock_gettime(CLOCK_MONOTONIC, &ts[1]);
 167                 if (rc != 0)
 168                         FATAL("%s: cannot get time: %s\n", it->it_name, xstrerror(errno));
 169
 170                 it->it_elapsed = timespec_sub(ts[1], ts[0]);
 171
 172                 assert(pid == pid2);
 173
 174                 DEBUG("%s: cmd = '%s', pid = %d, status = %d, elapsed_ns = %ld\n",
 175                       it->it_name, it->it_argv[0], pid, status, timespec_to_ns(it->it_elapsed));
 176
 177                 if (WIFEXITED(status) && WEXITSTATUS(status) == 127)
 178                         FATAL("%s: command '%s' (pid %d) exited with status 127\n", it->it_name, it->it_argv[0], pid);
 179
 180                 iabf_barrier_wait(ic, 1);
 181         }
 182
 183         return NULL;
 184 }
 185
 186 /* Run I, A+B, F once. Task threads must already be started.
 187  *
 188  * If delay_ns < 0 then
 189  *     delay exec of A by labs(delay_ns) nsec
 190  * else
 191  *     delay exec of B by labs(delay_ns) nsec.
 192  */
 193 static int iabf_step(struct iabf_control *ic,
 194                      struct iabf_task it[2],
 195                      long delay_ns)
 196 {
 197         int rc;
 198
 199         if (ic->ic_init != NULL && ic->ic_init[0] != NULL) {
 200                 rc = callvpe(ic->ic_init[0], ic->ic_init, environ);
 201                 DEBUG_D(rc); /* waitpid status */
 202                 if (rc != 0)
 203                         FATAL("initializer '%s' terminated with status %d\n", ic->ic_init[0], rc);
 204         }
 205
 206         DEBUG_D(delay_ns);
 207
 208         if (delay_ns < 0) {
 209                 it[0].it_delay = timespec_from_ns(labs(delay_ns));
 210                 it[1].it_delay = timespec_from_ns(0);
 211         } else {
 212                 it[0].it_delay = timespec_from_ns(0);
 213                 it[1].it_delay = timespec_from_ns(labs(delay_ns));
 214         }
 215
 216         iabf_barrier_wait(ic, 0);
 217
 218         /* A+B run here. */
 219
 220         iabf_barrier_wait(ic, 1);
 221
 222         if (ic->ic_fini != NULL && ic->ic_fini[0] != NULL) {
 223                 rc = callvpe(ic->ic_fini[0], ic->ic_fini, environ);
 224                 DEBUG_D(rc); /* waitpid status */
 225                 if (rc != 0)
 226                         FATAL("finalizer '%s' terminated with status %d\n", ic->ic_fini[0], rc);
 227         }
 228
 229         return 0;
 230 }
 231
 232 /* Run (I, A+B, F) step $IABF_AUTOTUNE_COUNT times. Task threads must
 233  * already be started. Get the average elapsed times for A and B.  We
 234  * want to choose delay_begin and delay_end to try to arrange all
 235  * possible overlaps given the expected elapsed times of A and B.
 236  *
 237  *     AAAAAAAAAA       delay(A) is approx elapsed(B)
 238  * BBBBB                delay(B) == 0
 239  *
 240  * AAAAAAAAAA           delay(A) == 0
 241  * BBBBB                delay(B) == 0
 242  *
 243  * AAAAAAAAAA           delay(A) == 0
 244  *          BBBBB       delay(B) is approx elapsed(A)
 245  *
 246  * Note that to delay task A we use a negative delay_ns.
 247  */
 248 static int iabf_autotune(struct iabf_control *ic,
 249                          struct iabf_task it[2])
 250 {
 251         long elapsed_ns[2] = { 0, 0 };
 252         long i, j;
 253
 254         DEBUG("begin autotune\n");
 255
 256         assert(ic->ic_autotune_count >= 0);
 257
 258         if (ic->ic_autotune_count == 0)
 259                 return 0;
 260
 261         for (i = 0; i < ic->ic_autotune_count; i++) {
 262                 iabf_step(ic, it, 0);
 263
 264                 for (j = 0; j < 2; j++)
 265                         elapsed_ns[j] += timespec_to_ns(it[j].it_elapsed);
 266         }
 267
 268         elapsed_ns[0] /= ic->ic_autotune_count;
 269         elapsed_ns[1] /= ic->ic_autotune_count;
 270
 271         DEBUG_D(elapsed_ns[0]);
 272         DEBUG_D(elapsed_ns[1]);
 273
 274         assert(0 <= elapsed_ns[0]);
 275         assert(0 <= elapsed_ns[1]);
 276
 277         /* TODO Apply a multiplier to endpoints. */
 278
 279         if (ic->ic_delay_begin_ns == LONG_MIN)
 280                 ic->ic_delay_begin_ns = -elapsed_ns[1];
 281
 282         if (ic->ic_delay_end_ns == LONG_MAX)
 283                 ic->ic_delay_end_ns = +elapsed_ns[0];
 284
 285         assert(ic->ic_delay_begin_ns <= ic->ic_delay_end_ns);
 286         assert(0 <= ic->ic_step_count);
 287
 288         if (ic->ic_step_count != 0)
 289                 ic->ic_delay_step_ns = (ic->ic_delay_end_ns - ic->ic_delay_begin_ns) / ic->ic_step_count;
 290
 291         if (ic->ic_delay_step_ns == 0)
 292                 ic->ic_delay_step_ns = 1; /* Or just leave it 0? */
 293
 294         DEBUG("end autotune\n");
 295
 296         return 0;
 297 }
 298
 299 /* Start A and B threads, autotune delay parameters if needed, run
 300  * iabf_step() however many times. */
 301 static int iabf(struct iabf_control *ic, char **a, char **b)
 302 {
 303         struct iabf_task it[2] = {
 304                 [0] = {
 305                         .it_control = ic,
 306                         .it_name = "A",
 307                         .it_argv = a,
 308                 },
 309                 [1] = {
 310                         .it_control = ic,
 311                         .it_name = "B",
 312                         .it_argv = b,
 313                 },
 314         };
 315         pthread_attr_t attr_, *attr = NULL;
 316         long i;
 317         int rc;
 318
 319         rc = pthread_attr_init(&attr_);
 320         if (rc != 0)
 321                 FATAL("cannot initialize thread attributes: %s\n", xstrerror(rc));
 322
 323         attr = &attr_;
 324
 325         for (i = 0; i < 2; i++) {
 326                 rc = pthread_barrier_init(&ic->ic_barrier[i], NULL, 3);
 327                 if (rc != 0)
 328                         FATAL("cannot initialize barrier: %s\n", xstrerror(rc));
 329         }
 330
 331         if (ic->ic_affinity != NULL) {
 332                 rc = pthread_setaffinity_np(pthread_self(), sizeof(ic->ic_affinity[2]), &ic->ic_affinity[2]);
 333                 if (rc != 0)
 334                         FATAL("cannot set CPU affinity : %s\n", xstrerror(rc));
 335         }
 336
 337         for (i = 0; i < 2; i++) {
 338                 if (ic->ic_affinity != NULL) {
 339                         rc = pthread_attr_setaffinity_np(attr, sizeof(ic->ic_affinity[i]), &ic->ic_affinity[i]);
 340                         if (rc != 0)
 341                                 FATAL("cannot set thread attr CPU affinity : %s\n", xstrerror(rc));
 342                 }
 343
 344                 rc = pthread_create(&it[i].it_thread,
 345                                     attr,
 346                                     iabf_task_thread,
 347                                     &it[i]);
 348                 if (rc != 0)
 349                         FATAL("cannot create thread: %s\n", xstrerror(rc));
 350         }
 351
 352         if (ic->ic_delay_begin_ns == LONG_MIN ||
 353             ic->ic_delay_end_ns == LONG_MAX ||
 354             ic->ic_delay_step_ns == 0)
 355                 iabf_autotune(ic, it);
 356
 357         DEBUG_D(ic->ic_delay_begin_ns);
 358         DEBUG_D(ic->ic_delay_end_ns);
 359         DEBUG_D(ic->ic_delay_step_ns);
 360
 361         long delay_ns;
 362         for (delay_ns = ic->ic_delay_begin_ns;
 363              delay_ns < ic->ic_delay_end_ns;
 364              delay_ns += ic->ic_delay_step_ns)
 365                 iabf_step(ic, it, delay_ns);
 366
 367         ic->ic_should_stop = 1;
 368         DEBUG_D(ic->ic_should_stop);
 369
 370         iabf_barrier_wait(ic, 0);
 371
 372         for (i = 0; i < 2; i++) {
 373                 rc = pthread_join(it[i].it_thread, NULL);
 374                 if (rc != 0)
 375                         FATAL("cannot join thread %s: %s\n", it[i].it_name, xstrerror(rc));
 376         }
 377
 378         for (i = 0; i < 2; i++) {
 379                 rc = pthread_barrier_destroy(&ic->ic_barrier[i]);
 380                 if (rc != 0)
 381                         FATAL("cannot destroy barrier: %s\n", xstrerror(rc));
 382         }
 383
 384         if (attr != NULL)
 385                 pthread_attr_destroy(attr);
 386
 387         return 0;
 388 }
 389
 390 /* strsep() for argvs */
 391 char **arg_sep(char ***pargs, const char *delim)
 392 {
 393         char **begin, **end;
 394
 395         begin = *pargs;
 396         if (begin == NULL)
 397                 return NULL;
 398
 399         /* Find the end of the token.  */
 400         /* end = begin + strcspn (begin, delim); */
 401
 402         for (end = begin; *end != NULL && strcmp(*end, delim) != 0; end++)
 403                 ;
 404
 405         if (*end != NULL) {
 406                 /* Terminate the token and set *STRINGP past NUL character. */
 407                 *end++ = NULL;
 408                 *pargs = end;
 409         } else {
 410                 /* No more delimiters; this is the last token. */
 411                 *pargs = NULL;
 412         }
 413
 414         return begin;
 415 }
 416
 417 static cpu_set_t *iabf_affinity(const char *str)
 418 {
 419         cpu_set_t *cpu_sets = NULL;
 420         char *str1 = NULL;
 421         char *p;
 422         char *q;
 423         char *r;
 424         int i;
 425
 426         if (str == NULL)
 427                 return NULL;
 428
 429         cpu_sets = calloc(3, sizeof(cpu_sets[0]));
 430         p = str1 = strdup(str);
 431
 432         for (i = 0; i < 3; i++) {
 433                 CPU_ZERO(&cpu_sets[i]);
 434
 435                 q = strsep(&p, " ");
 436                 if (q == NULL)
 437                         FATAL("invalid affinity '%s'\n", str);
 438
 439                 while ((r = strsep(&q, ",")) != NULL)
 440                         CPU_SET(atoi(r), &cpu_sets[i]);
 441         }
 442
 443         if (p != NULL)
 444                 FATAL("invalid affinity '%s'\n", str);
 445
 446         free(str1);
 447
 448         return cpu_sets;
 449 }
 450
 451 int main(int argc, char **argv)
 452 {
 453         struct iabf_control ic = {
 454                 .ic_should_stop = 0,
 455         };
 456         char **args = argv + 1;
 457         char **a;
 458         char **b;
 459
 460         iabf_tid = syscall(SYS_gettid);
 461
 462         iabf_debug = atoi(getenv("IABF_DEBUG") ?: "0");
 463
 464         ic.ic_init = arg_sep(&args, iabf_delim);
 465         a = arg_sep(&args, iabf_delim);
 466         b = arg_sep(&args, iabf_delim);
 467         ic.ic_fini = arg_sep(&args, iabf_delim);
 468
 469         if (ic.ic_init == NULL ||
 470             a == NULL ||
 471             b == NULL ||
 472             ic.ic_fini == NULL)
 473                 FATAL("missing '%s' in argv\n", iabf_delim);
 474
 475         int i;
 476         for (i = 0; ic.ic_init[i] != NULL; i++)
 477                 DEBUG_S(ic.ic_init[i]);
 478
 479         for (i = 0; a[i] != NULL; i++)
 480                 DEBUG_S(a[i]);
 481
 482         for (i = 0; b[i] != NULL; i++)
 483                 DEBUG_S(b[i]);
 484
 485         for (i = 0; ic.ic_fini[i] != NULL; i++)
 486                 DEBUG_S(ic.ic_fini[i]);
 487
 488         ic.ic_affinity = iabf_affinity(getenv("IABF_AFFINITY"));
 489         ic.ic_delay_begin_ns = iabf_getenvl("IABF_DELAY_BEGIN_NS", LONG_MIN);
 490         ic.ic_delay_end_ns = iabf_getenvl("IABF_DELAY_END_NS", LONG_MAX);
 491         ic.ic_delay_step_ns = iabf_getenvl("IABF_DELAY_STEP_NS", 0);
 492         ic.ic_step_count = iabf_getenvl("IABF_STEP_COUNT", 0);
 493         ic.ic_autotune_count = iabf_getenvl("IABF_AUTOTUNE_COUNT", 16);
 494
 495         DEBUG_D(ic.ic_delay_begin_ns);
 496         DEBUG_D(ic.ic_delay_end_ns);
 497         DEBUG_D(ic.ic_delay_step_ns);
 498         DEBUG_D(ic.ic_step_count);
 499         DEBUG_D(ic.ic_autotune_count);
 500
 501         assert(ic.ic_delay_begin_ns <= ic.ic_delay_end_ns);
 502         assert(0 <= ic.ic_delay_step_ns);
 503         assert(0 <= ic.ic_step_count);
 504
 505         iabf(&ic, a, b);
 506
 507         return 0;
 508 }