2 * Copyright (C) 2015 Davidlohr Bueso.
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.
10 /* For the CLR_() macros */
14 #include "../util/util.h"
15 #include "../util/stat.h"
16 #include <subcmd/parse-options.h>
17 #include "../util/header.h"
28 struct timeval runtime
;
31 static unsigned int nwakes
= 1;
33 /* all threads will block on the same futex -- hash bucket chaos ;) */
34 static u_int32_t futex
= 0;
36 static pthread_t
*blocked_worker
;
37 static bool done
= false, silent
= false, fshared
= false;
38 static unsigned int nblocked_threads
= 0, nwaking_threads
= 0;
39 static pthread_mutex_t thread_lock
;
40 static pthread_cond_t thread_parent
, thread_worker
;
41 static struct stats waketime_stats
, wakeup_stats
;
42 static unsigned int ncpus
, threads_starting
;
43 static int futex_flag
= 0;
45 static const struct option options
[] = {
46 OPT_UINTEGER('t', "threads", &nblocked_threads
, "Specify amount of threads"),
47 OPT_UINTEGER('w', "nwakers", &nwaking_threads
, "Specify amount of waking threads"),
48 OPT_BOOLEAN( 's', "silent", &silent
, "Silent mode: do not display data/details"),
49 OPT_BOOLEAN( 'S', "shared", &fshared
, "Use shared futexes instead of private ones"),
53 static const char * const bench_futex_wake_parallel_usage
[] = {
54 "perf bench futex wake-parallel <options>",
58 static void *waking_workerfn(void *arg
)
60 struct thread_data
*waker
= (struct thread_data
*) arg
;
61 struct timeval start
, end
;
63 gettimeofday(&start
, NULL
);
65 waker
->nwoken
= futex_wake(&futex
, nwakes
, futex_flag
);
66 if (waker
->nwoken
!= nwakes
)
67 warnx("couldn't wakeup all tasks (%d/%d)",
68 waker
->nwoken
, nwakes
);
70 gettimeofday(&end
, NULL
);
71 timersub(&end
, &start
, &waker
->runtime
);
77 static void wakeup_threads(struct thread_data
*td
, pthread_attr_t thread_attr
)
81 pthread_attr_setdetachstate(&thread_attr
, PTHREAD_CREATE_JOINABLE
);
83 /* create and block all threads */
84 for (i
= 0; i
< nwaking_threads
; i
++) {
86 * Thread creation order will impact per-thread latency
87 * as it will affect the order to acquire the hb spinlock.
88 * For now let the scheduler decide.
90 if (pthread_create(&td
[i
].worker
, &thread_attr
,
91 waking_workerfn
, (void *)&td
[i
]))
92 err(EXIT_FAILURE
, "pthread_create");
95 for (i
= 0; i
< nwaking_threads
; i
++)
96 if (pthread_join(td
[i
].worker
, NULL
))
97 err(EXIT_FAILURE
, "pthread_join");
100 static void *blocked_workerfn(void *arg __maybe_unused
)
102 pthread_mutex_lock(&thread_lock
);
104 if (!threads_starting
)
105 pthread_cond_signal(&thread_parent
);
106 pthread_cond_wait(&thread_worker
, &thread_lock
);
107 pthread_mutex_unlock(&thread_lock
);
109 while (1) { /* handle spurious wakeups */
110 if (futex_wait(&futex
, 0, NULL
, futex_flag
) != EINTR
)
118 static void block_threads(pthread_t
*w
, pthread_attr_t thread_attr
)
123 threads_starting
= nblocked_threads
;
125 /* create and block all threads */
126 for (i
= 0; i
< nblocked_threads
; i
++) {
128 CPU_SET(i
% ncpus
, &cpu
);
130 if (pthread_attr_setaffinity_np(&thread_attr
, sizeof(cpu_set_t
), &cpu
))
131 err(EXIT_FAILURE
, "pthread_attr_setaffinity_np");
133 if (pthread_create(&w
[i
], &thread_attr
, blocked_workerfn
, NULL
))
134 err(EXIT_FAILURE
, "pthread_create");
138 static void print_run(struct thread_data
*waking_worker
, unsigned int run_num
)
140 unsigned int i
, wakeup_avg
;
141 double waketime_avg
, waketime_stddev
;
142 struct stats __waketime_stats
, __wakeup_stats
;
144 init_stats(&__wakeup_stats
);
145 init_stats(&__waketime_stats
);
147 for (i
= 0; i
< nwaking_threads
; i
++) {
148 update_stats(&__waketime_stats
, waking_worker
[i
].runtime
.tv_usec
);
149 update_stats(&__wakeup_stats
, waking_worker
[i
].nwoken
);
152 waketime_avg
= avg_stats(&__waketime_stats
);
153 waketime_stddev
= stddev_stats(&__waketime_stats
);
154 wakeup_avg
= avg_stats(&__wakeup_stats
);
156 printf("[Run %d]: Avg per-thread latency (waking %d/%d threads) "
157 "in %.4f ms (+-%.2f%%)\n", run_num
+ 1, wakeup_avg
,
158 nblocked_threads
, waketime_avg
/1e3
,
159 rel_stddev_stats(waketime_stddev
, waketime_avg
));
162 static void print_summary(void)
164 unsigned int wakeup_avg
;
165 double waketime_avg
, waketime_stddev
;
167 waketime_avg
= avg_stats(&waketime_stats
);
168 waketime_stddev
= stddev_stats(&waketime_stats
);
169 wakeup_avg
= avg_stats(&wakeup_stats
);
171 printf("Avg per-thread latency (waking %d/%d threads) in %.4f ms (+-%.2f%%)\n",
175 rel_stddev_stats(waketime_stddev
, waketime_avg
));
179 static void do_run_stats(struct thread_data
*waking_worker
)
183 for (i
= 0; i
< nwaking_threads
; i
++) {
184 update_stats(&waketime_stats
, waking_worker
[i
].runtime
.tv_usec
);
185 update_stats(&wakeup_stats
, waking_worker
[i
].nwoken
);
190 static void toggle_done(int sig __maybe_unused
,
191 siginfo_t
*info __maybe_unused
,
192 void *uc __maybe_unused
)
197 int bench_futex_wake_parallel(int argc
, const char **argv
,
198 const char *prefix __maybe_unused
)
202 struct sigaction act
;
203 pthread_attr_t thread_attr
;
204 struct thread_data
*waking_worker
;
206 argc
= parse_options(argc
, argv
, options
,
207 bench_futex_wake_parallel_usage
, 0);
209 usage_with_options(bench_futex_wake_parallel_usage
, options
);
213 sigfillset(&act
.sa_mask
);
214 act
.sa_sigaction
= toggle_done
;
215 sigaction(SIGINT
, &act
, NULL
);
217 ncpus
= sysconf(_SC_NPROCESSORS_ONLN
);
218 if (!nblocked_threads
)
219 nblocked_threads
= ncpus
;
221 /* some sanity checks */
222 if (nwaking_threads
> nblocked_threads
|| !nwaking_threads
)
223 nwaking_threads
= nblocked_threads
;
225 if (nblocked_threads
% nwaking_threads
)
226 errx(EXIT_FAILURE
, "Must be perfectly divisible");
228 * Each thread will wakeup nwakes tasks in
229 * a single futex_wait call.
231 nwakes
= nblocked_threads
/nwaking_threads
;
233 blocked_worker
= calloc(nblocked_threads
, sizeof(*blocked_worker
));
235 err(EXIT_FAILURE
, "calloc");
238 futex_flag
= FUTEX_PRIVATE_FLAG
;
240 printf("Run summary [PID %d]: blocking on %d threads (at [%s] "
241 "futex %p), %d threads waking up %d at a time.\n\n",
242 getpid(), nblocked_threads
, fshared
? "shared":"private",
243 &futex
, nwaking_threads
, nwakes
);
245 init_stats(&wakeup_stats
);
246 init_stats(&waketime_stats
);
248 pthread_attr_init(&thread_attr
);
249 pthread_mutex_init(&thread_lock
, NULL
);
250 pthread_cond_init(&thread_parent
, NULL
);
251 pthread_cond_init(&thread_worker
, NULL
);
253 for (j
= 0; j
< bench_repeat
&& !done
; j
++) {
254 waking_worker
= calloc(nwaking_threads
, sizeof(*waking_worker
));
256 err(EXIT_FAILURE
, "calloc");
258 /* create, launch & block all threads */
259 block_threads(blocked_worker
, thread_attr
);
261 /* make sure all threads are already blocked */
262 pthread_mutex_lock(&thread_lock
);
263 while (threads_starting
)
264 pthread_cond_wait(&thread_parent
, &thread_lock
);
265 pthread_cond_broadcast(&thread_worker
);
266 pthread_mutex_unlock(&thread_lock
);
270 /* Ok, all threads are patiently blocked, start waking folks up */
271 wakeup_threads(waking_worker
, thread_attr
);
273 for (i
= 0; i
< nblocked_threads
; i
++) {
274 ret
= pthread_join(blocked_worker
[i
], NULL
);
276 err(EXIT_FAILURE
, "pthread_join");
279 do_run_stats(waking_worker
);
281 print_run(waking_worker
, j
);
286 /* cleanup & report results */
287 pthread_cond_destroy(&thread_parent
);
288 pthread_cond_destroy(&thread_worker
);
289 pthread_mutex_destroy(&thread_lock
);
290 pthread_attr_destroy(&thread_attr
);
294 free(blocked_worker
);