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