tools/perf/bench/futex-wake-parallel.c

   1 /*
   2  * Copyright (C) 2015 Davidlohr Bueso.
   3  *
   4  * Block a bunch of threads and let parallel waker threads wakeup an
   5  * equal amount of them. The program output reflects the avg latency
   6  * for each individual thread to service its share of work. Ultimately
   7  * it can be used to measure futex_wake() changes.
   8  */
   9
  10 /* For the CLR_() macros */
  11 #include <pthread.h>
  12
  13 #include <signal.h>
  14 #include "../util/stat.h"
  15 #include <subcmd/parse-options.h>
  16 #include <linux/compiler.h>
  17 #include <linux/kernel.h>
  18 #include <errno.h>
  19 #include "bench.h"
  20 #include "futex.h"
  21
  22 #include <err.h>
  23 #include <stdlib.h>
  24 #include <sys/time.h>
  25
  26 struct thread_data {
  27         pthread_t worker;
  28         unsigned int nwoken;
  29         struct timeval runtime;
  30 };
  31
  32 static unsigned int nwakes = 1;
  33
  34 /* all threads will block on the same futex -- hash bucket chaos ;) */
  35 static u_int32_t futex = 0;
  36
  37 static pthread_t *blocked_worker;
  38 static bool done = false, silent = false, fshared = false;
  39 static unsigned int nblocked_threads = 0, nwaking_threads = 0;
  40 static pthread_mutex_t thread_lock;
  41 static pthread_cond_t thread_parent, thread_worker;
  42 static struct stats waketime_stats, wakeup_stats;
  43 static unsigned int ncpus, threads_starting;
  44 static int futex_flag = 0;
  45
  46 static const struct option options[] = {
  47         OPT_UINTEGER('t', "threads", &nblocked_threads, "Specify amount of threads"),
  48         OPT_UINTEGER('w', "nwakers", &nwaking_threads, "Specify amount of waking threads"),
  49         OPT_BOOLEAN( 's', "silent",  &silent,   "Silent mode: do not display data/details"),
  50         OPT_BOOLEAN( 'S', "shared",  &fshared,  "Use shared futexes instead of private ones"),
  51         OPT_END()
  52 };
  53
  54 static const char * const bench_futex_wake_parallel_usage[] = {
  55         "perf bench futex wake-parallel <options>",
  56         NULL
  57 };
  58
  59 static void *waking_workerfn(void *arg)
  60 {
  61         struct thread_data *waker = (struct thread_data *) arg;
  62         struct timeval start, end;
  63
  64         gettimeofday(&start, NULL);
  65
  66         waker->nwoken = futex_wake(&futex, nwakes, futex_flag);
  67         if (waker->nwoken != nwakes)
  68                 warnx("couldn't wakeup all tasks (%d/%d)",
  69                       waker->nwoken, nwakes);
  70
  71         gettimeofday(&end, NULL);
  72         timersub(&end, &start, &waker->runtime);
  73
  74         pthread_exit(NULL);
  75         return NULL;
  76 }
  77
  78 static void wakeup_threads(struct thread_data *td, pthread_attr_t thread_attr)
  79 {
  80         unsigned int i;
  81
  82         pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_JOINABLE);
  83
  84         /* create and block all threads */
  85         for (i = 0; i < nwaking_threads; i++) {
  86                 /*
  87                  * Thread creation order will impact per-thread latency
  88                  * as it will affect the order to acquire the hb spinlock.
  89                  * For now let the scheduler decide.
  90                  */
  91                 if (pthread_create(&td[i].worker, &thread_attr,
  92                                    waking_workerfn, (void *)&td[i]))
  93                         err(EXIT_FAILURE, "pthread_create");
  94         }
  95
  96         for (i = 0; i < nwaking_threads; i++)
  97                 if (pthread_join(td[i].worker, NULL))
  98                         err(EXIT_FAILURE, "pthread_join");
  99 }
 100
 101 static void *blocked_workerfn(void *arg __maybe_unused)
 102 {
 103         pthread_mutex_lock(&thread_lock);
 104         threads_starting--;
 105         if (!threads_starting)
 106                 pthread_cond_signal(&thread_parent);
 107         pthread_cond_wait(&thread_worker, &thread_lock);
 108         pthread_mutex_unlock(&thread_lock);
 109
 110         while (1) { /* handle spurious wakeups */
 111                 if (futex_wait(&futex, 0, NULL, futex_flag) != EINTR)
 112                         break;
 113         }
 114
 115         pthread_exit(NULL);
 116         return NULL;
 117 }
 118
 119 static void block_threads(pthread_t *w, pthread_attr_t thread_attr)
 120 {
 121         cpu_set_t cpu;
 122         unsigned int i;
 123
 124         threads_starting = nblocked_threads;
 125
 126         /* create and block all threads */
 127         for (i = 0; i < nblocked_threads; i++) {
 128                 CPU_ZERO(&cpu);
 129                 CPU_SET(i % ncpus, &cpu);
 130
 131                 if (pthread_attr_setaffinity_np(&thread_attr, sizeof(cpu_set_t), &cpu))
 132                         err(EXIT_FAILURE, "pthread_attr_setaffinity_np");
 133
 134                 if (pthread_create(&w[i], &thread_attr, blocked_workerfn, NULL))
 135                         err(EXIT_FAILURE, "pthread_create");
 136         }
 137 }
 138
 139 static void print_run(struct thread_data *waking_worker, unsigned int run_num)
 140 {
 141         unsigned int i, wakeup_avg;
 142         double waketime_avg, waketime_stddev;
 143         struct stats __waketime_stats, __wakeup_stats;
 144
 145         init_stats(&__wakeup_stats);
 146         init_stats(&__waketime_stats);
 147
 148         for (i = 0; i < nwaking_threads; i++) {
 149                 update_stats(&__waketime_stats, waking_worker[i].runtime.tv_usec);
 150                 update_stats(&__wakeup_stats, waking_worker[i].nwoken);
 151         }
 152
 153         waketime_avg = avg_stats(&__waketime_stats);
 154         waketime_stddev = stddev_stats(&__waketime_stats);
 155         wakeup_avg = avg_stats(&__wakeup_stats);
 156
 157         printf("[Run %d]: Avg per-thread latency (waking %d/%d threads) "
 158                "in %.4f ms (+-%.2f%%)\n", run_num + 1, wakeup_avg,
 159                nblocked_threads, waketime_avg/1e3,
 160                rel_stddev_stats(waketime_stddev, waketime_avg));
 161 }
 162
 163 static void print_summary(void)
 164 {
 165         unsigned int wakeup_avg;
 166         double waketime_avg, waketime_stddev;
 167
 168         waketime_avg = avg_stats(&waketime_stats);
 169         waketime_stddev = stddev_stats(&waketime_stats);
 170         wakeup_avg = avg_stats(&wakeup_stats);
 171
 172         printf("Avg per-thread latency (waking %d/%d threads) in %.4f ms (+-%.2f%%)\n",
 173                wakeup_avg,
 174                nblocked_threads,
 175                waketime_avg/1e3,
 176                rel_stddev_stats(waketime_stddev, waketime_avg));
 177 }
 178
 179
 180 static void do_run_stats(struct thread_data *waking_worker)
 181 {
 182         unsigned int i;
 183
 184         for (i = 0; i < nwaking_threads; i++) {
 185                 update_stats(&waketime_stats, waking_worker[i].runtime.tv_usec);
 186                 update_stats(&wakeup_stats, waking_worker[i].nwoken);
 187         }
 188
 189 }
 190
 191 static void toggle_done(int sig __maybe_unused,
 192                         siginfo_t *info __maybe_unused,
 193                         void *uc __maybe_unused)
 194 {
 195         done = true;
 196 }
 197
 198 int bench_futex_wake_parallel(int argc, const char **argv,
 199                               const char *prefix __maybe_unused)
 200 {
 201         int ret = 0;
 202         unsigned int i, j;
 203         struct sigaction act;
 204         pthread_attr_t thread_attr;
 205         struct thread_data *waking_worker;
 206
 207         argc = parse_options(argc, argv, options,
 208                              bench_futex_wake_parallel_usage, 0);
 209         if (argc) {
 210                 usage_with_options(bench_futex_wake_parallel_usage, options);
 211                 exit(EXIT_FAILURE);
 212         }
 213
 214         sigfillset(&act.sa_mask);
 215         act.sa_sigaction = toggle_done;
 216         sigaction(SIGINT, &act, NULL);
 217
 218         ncpus = sysconf(_SC_NPROCESSORS_ONLN);
 219         if (!nblocked_threads)
 220                 nblocked_threads = ncpus;
 221
 222         /* some sanity checks */
 223         if (nwaking_threads > nblocked_threads || !nwaking_threads)
 224                 nwaking_threads = nblocked_threads;
 225
 226         if (nblocked_threads % nwaking_threads)
 227                 errx(EXIT_FAILURE, "Must be perfectly divisible");
 228         /*
 229          * Each thread will wakeup nwakes tasks in
 230          * a single futex_wait call.
 231          */
 232         nwakes = nblocked_threads/nwaking_threads;
 233
 234         blocked_worker = calloc(nblocked_threads, sizeof(*blocked_worker));
 235         if (!blocked_worker)
 236                 err(EXIT_FAILURE, "calloc");
 237
 238         if (!fshared)
 239                 futex_flag = FUTEX_PRIVATE_FLAG;
 240
 241         printf("Run summary [PID %d]: blocking on %d threads (at [%s] "
 242                "futex %p), %d threads waking up %d at a time.\n\n",
 243                getpid(), nblocked_threads, fshared ? "shared":"private",
 244                &futex, nwaking_threads, nwakes);
 245
 246         init_stats(&wakeup_stats);
 247         init_stats(&waketime_stats);
 248
 249         pthread_attr_init(&thread_attr);
 250         pthread_mutex_init(&thread_lock, NULL);
 251         pthread_cond_init(&thread_parent, NULL);
 252         pthread_cond_init(&thread_worker, NULL);
 253
 254         for (j = 0; j < bench_repeat && !done; j++) {
 255                 waking_worker = calloc(nwaking_threads, sizeof(*waking_worker));
 256                 if (!waking_worker)
 257                         err(EXIT_FAILURE, "calloc");
 258
 259                 /* create, launch & block all threads */
 260                 block_threads(blocked_worker, thread_attr);
 261
 262                 /* make sure all threads are already blocked */
 263                 pthread_mutex_lock(&thread_lock);
 264                 while (threads_starting)
 265                         pthread_cond_wait(&thread_parent, &thread_lock);
 266                 pthread_cond_broadcast(&thread_worker);
 267                 pthread_mutex_unlock(&thread_lock);
 268
 269                 usleep(100000);
 270
 271                 /* Ok, all threads are patiently blocked, start waking folks up */
 272                 wakeup_threads(waking_worker, thread_attr);
 273
 274                 for (i = 0; i < nblocked_threads; i++) {
 275                         ret = pthread_join(blocked_worker[i], NULL);
 276                         if (ret)
 277                                 err(EXIT_FAILURE, "pthread_join");
 278                 }
 279
 280                 do_run_stats(waking_worker);
 281                 if (!silent)
 282                         print_run(waking_worker, j);
 283
 284                 free(waking_worker);
 285         }
 286
 287         /* cleanup & report results */
 288         pthread_cond_destroy(&thread_parent);
 289         pthread_cond_destroy(&thread_worker);
 290         pthread_mutex_destroy(&thread_lock);
 291         pthread_attr_destroy(&thread_attr);
 292
 293         print_summary();
 294
 295         free(blocked_worker);
 296         return ret;
 297 }