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