[RFC PATCH v1 00/37] perf: use workqueue for evlist operations

@@ -265,11 +265,18 @@ bool perf_cpu_map__empty(const struct perf_cpu_map *map)
 
 int perf_cpu_map__idx(struct perf_cpu_map *cpus, int cpu)
 {
-	int i;
+	int low = 0, high = cpus->nr, idx, cpu_at_idx;
 
-	for (i = 0; i < cpus->nr; ++i) {
-		if (cpus->map[i] == cpu)
-			return i;
+	while (low < high) {
+		idx = (low + high) / 2;
+		cpu_at_idx = cpus->map[idx];
+
+		if (cpu_at_idx == cpu)
+			return idx;
+		else if (cpu_at_idx > cpu)
+			high = idx;
+		else
+			low = idx+1;
 	}
 
 	return -1;

@@ -284,14 +284,10 @@ int perf_cpu_map__idx(struct perf_cpu_map *cpus, int cpu)
 
 int perf_cpu_map__max(struct perf_cpu_map *map)
 {
-	int i, max = -1;
-
-	for (i = 0; i < map->nr; i++) {
-		if (map->map[i] > max)
-			max = map->map[i];
-	}
-
-	return max;
+	if (map->nr > 0)
+		return map->map[map->nr-1];
+	else
+		return -1;
 }
 
 /*

@@ -407,7 +407,7 @@ static int read_affinity_counters(struct timespec *rs)
 {
 	struct evsel *counter;
 	struct affinity affinity;
-	int i, ncpus, cpu;
+	int i, ncpus, cpu, cpu_idx;
 
 	if (all_counters_use_bpf)
 		return 0;

@@ -424,13 +424,14 @@ static int read_affinity_counters(struct timespec *rs)
 		affinity__set(&affinity, cpu);
 
 		evlist__for_each_entry(evsel_list, counter) {
-			if (evsel__cpu_iter_skip(counter, cpu))
+			cpu_idx = evsel__find_cpu(counter, cpu);
+			if (cpu_idx < 0)
 				continue;
 			if (evsel__is_bpf(counter))
 				continue;
 			if (!counter->err) {
 				counter->err = read_counter_cpu(counter, rs,
-								counter->cpu_iter - 1);
+								cpu_idx);
 			}
 		}
 	}

@@ -789,7 +790,7 @@ static int __run_perf_stat(int argc, const char **argv, int run_idx)
 	const bool forks = (argc > 0);
 	bool is_pipe = STAT_RECORD ? perf_stat.data.is_pipe : false;
 	struct affinity affinity;
-	int i, cpu, err;
+	int i, cpu, cpu_idx, err;
 	bool second_pass = false;
 
 	if (forks) {

@@ -823,7 +824,8 @@ static int __run_perf_stat(int argc, const char **argv, int run_idx)
 		affinity__set(&affinity, cpu);
 
 		evlist__for_each_entry(evsel_list, counter) {
-			if (evsel__cpu_iter_skip(counter, cpu))
+			cpu_idx = evsel__find_cpu(counter, cpu);
+			if (cpu_idx < 0)
 				continue;
 			if (counter->reset_group || counter->errored)
 				continue;

@@ -831,7 +833,7 @@ static int __run_perf_stat(int argc, const char **argv, int run_idx)
 				continue;
 try_again:
 			if (create_perf_stat_counter(counter, &stat_config, &target,
-						     counter->cpu_iter - 1) < 0) {
+						     cpu_idx) < 0) {
 
 				/*
 				 * Weak group failed. We cannot just undo this here

@@ -877,22 +879,24 @@ static int __run_perf_stat(int argc, const char **argv, int run_idx)
 			evlist__for_each_entry(evsel_list, counter) {
 				if (!counter->reset_group && !counter->errored)
 					continue;
-				if (evsel__cpu_iter_skip_no_inc(counter, cpu))
+				cpu_idx = evsel__find_cpu(counter, cpu);
+				if (cpu_idx < 0)
 					continue;
-				perf_evsel__close_cpu(&counter->core, counter->cpu_iter);
+				perf_evsel__close_cpu(&counter->core, cpu_idx);
 			}
 			/* Now reopen weak */
 			evlist__for_each_entry(evsel_list, counter) {
 				if (!counter->reset_group && !counter->errored)
 					continue;
-				if (evsel__cpu_iter_skip(counter, cpu))
+				cpu_idx = evsel__find_cpu(counter, cpu);
+				if (cpu_idx < 0)
 					continue;
 				if (!counter->reset_group)
 					continue;
 try_again_reset:
 				pr_debug2("reopening weak %s\n", evsel__name(counter));
 				if (create_perf_stat_counter(counter, &stat_config, &target,
-							     counter->cpu_iter - 1) < 0) {
+							     cpu_idx) < 0) {
 
 					switch (stat_handle_error(counter)) {
 					case COUNTER_FATAL:

@@ -341,36 +341,9 @@ static int evlist__nr_threads(struct evlist *evlist, struct evsel *evsel)
 		return perf_thread_map__nr(evlist->core.threads);
 }
 
-void evlist__cpu_iter_start(struct evlist *evlist)
+int evsel__find_cpu(struct evsel *ev, int cpu)
 {
-	struct evsel *pos;
-
-	/*
-	 * Reset the per evsel cpu_iter. This is needed because
-	 * each evsel's cpumap may have a different index space,
-	 * and some operations need the index to modify
-	 * the FD xyarray (e.g. open, close)
-	 */
-	evlist__for_each_entry(evlist, pos)
-		pos->cpu_iter = 0;
-}
-
-bool evsel__cpu_iter_skip_no_inc(struct evsel *ev, int cpu)
-{
-	if (ev->cpu_iter >= ev->core.cpus->nr)
-		return true;
-	if (cpu >= 0 && ev->core.cpus->map[ev->cpu_iter] != cpu)
-		return true;
-	return false;
-}
-
-bool evsel__cpu_iter_skip(struct evsel *ev, int cpu)
-{
-	if (!evsel__cpu_iter_skip_no_inc(ev, cpu)) {
-		ev->cpu_iter++;
-		return false;
-	}
-	return true;
+	return perf_cpu_map__idx(ev->core.cpus, cpu);
 }
 
 static int evsel__strcmp(struct evsel *pos, char *evsel_name)

@@ -400,7 +373,7 @@ static void __evlist__disable(struct evlist *evlist, char *evsel_name)
 {
 	struct evsel *pos;
 	struct affinity affinity;
-	int cpu, i, imm = 0;
+	int cpu, i, imm = 0, cpu_idx;
 	bool has_imm = false;
 
 	if (affinity__setup(&affinity) < 0)

@@ -414,7 +387,8 @@ static void __evlist__disable(struct evlist *evlist, char *evsel_name)
 			evlist__for_each_entry(evlist, pos) {
 				if (evsel__strcmp(pos, evsel_name))
 					continue;
-				if (evsel__cpu_iter_skip(pos, cpu))
+				cpu_idx = evsel__find_cpu(pos, cpu);
+				if (cpu_idx < 0)
 					continue;
 				if (pos->disabled || !evsel__is_group_leader(pos) || !pos->core.fd)
 					continue;

@@ -422,7 +396,7 @@ static void __evlist__disable(struct evlist *evlist, char *evsel_name)
 					has_imm = true;
 				if (pos->immediate != imm)
 					continue;
-				evsel__disable_cpu(pos, pos->cpu_iter - 1);
+				evsel__disable_cpu(pos, cpu_idx);
 			}
 		}
 		if (!has_imm)

@@ -462,7 +436,7 @@ static void __evlist__enable(struct evlist *evlist, char *evsel_name)
 {
 	struct evsel *pos;
 	struct affinity affinity;
-	int cpu, i;
+	int cpu, i, cpu_idx;
 
 	if (affinity__setup(&affinity) < 0)
 		return;

@@ -473,11 +447,12 @@ static void __evlist__enable(struct evlist *evlist, char *evsel_name)
 		evlist__for_each_entry(evlist, pos) {
 			if (evsel__strcmp(pos, evsel_name))
 				continue;
-			if (evsel__cpu_iter_skip(pos, cpu))
+			cpu_idx = evsel__find_cpu(pos, cpu);
+			if (cpu_idx < 0)
 				continue;
 			if (!evsel__is_group_leader(pos) || !pos->core.fd)
 				continue;
-			evsel__enable_cpu(pos, pos->cpu_iter - 1);
+			evsel__enable_cpu(pos, cpu_idx);
 		}
 	}
 	affinity__cleanup(&affinity);

@@ -1264,7 +1239,7 @@ void evlist__close(struct evlist *evlist)
 {
 	struct evsel *evsel;
 	struct affinity affinity;
-	int cpu, i;
+	int cpu, i, cpu_idx;
 
 	/*
 	 * With perf record core.cpus is usually NULL.

@@ -1282,9 +1257,10 @@ void evlist__close(struct evlist *evlist)
 		affinity__set(&affinity, cpu);
 
 		evlist__for_each_entry_reverse(evlist, evsel) {
-			if (evsel__cpu_iter_skip(evsel, cpu))
-			    continue;
-			perf_evsel__close_cpu(&evsel->core, evsel->cpu_iter - 1);
+			cpu_idx = evsel__find_cpu(evsel, cpu);
+			if (cpu_idx < 0)
+				continue;
+			perf_evsel__close_cpu(&evsel->core, cpu_idx);
 		}
 	}
 	affinity__cleanup(&affinity);

@@ -310,15 +310,12 @@ void evlist__to_front(struct evlist *evlist, struct evsel *move_evsel);
 	__evlist__for_each_entry_safe(&(evlist)->core.entries, tmp, evsel)
 
 #define evlist__for_each_cpu(evlist, index, cpu)	\
-	evlist__cpu_iter_start(evlist);			\
 	perf_cpu_map__for_each_cpu (cpu, index, (evlist)->core.all_cpus)
 
 struct evsel *evlist__get_tracking_event(struct evlist *evlist);
 void evlist__set_tracking_event(struct evlist *evlist, struct evsel *tracking_evsel);
 
-void evlist__cpu_iter_start(struct evlist *evlist);
-bool evsel__cpu_iter_skip(struct evsel *ev, int cpu);
-bool evsel__cpu_iter_skip_no_inc(struct evsel *ev, int cpu);
+int evsel__find_cpu(struct evsel *ev, int cpu);
 
 struct evsel *evlist__find_evsel_by_str(struct evlist *evlist, const char *str);
 

@@ -119,7 +119,6 @@ struct evsel {
 	bool			errored;
 	struct hashmap		*per_pkg_mask;
 	int			err;
-	int			cpu_iter;
 	struct {
 		evsel__sb_cb_t	*cb;
 		void		*data;

@@ -350,3 +350,15 @@ int perf_mmap__push(struct mmap *md, void *to,
 out:
 	return rc;
 }
+
+int mmap_cpu_mask__duplicate(struct mmap_cpu_mask *original,
+				struct mmap_cpu_mask *clone)
+{
+	clone->nbits = original->nbits;
+	clone->bits = bitmap_alloc(original->nbits);
+	if (!clone->bits)
+		return -ENOMEM;
+
+	memcpy(clone->bits, original->bits, MMAP_CPU_MASK_BYTES(original));
+	return 0;
+}

@@ -63,4 +63,7 @@ size_t mmap__mmap_len(struct mmap *map);
 
 void mmap_cpu_mask__scnprintf(struct mmap_cpu_mask *mask, const char *tag);
 
+int mmap_cpu_mask__duplicate(struct mmap_cpu_mask *original,
+				struct mmap_cpu_mask *clone);
+
 #endif /*__PERF_MMAP_H */

@@ -6,6 +6,7 @@
 #include <linux/refcount.h>
 #include <linux/types.h>
 #include <linux/ring_buffer.h>
+#include <linux/bitops.h>
 #include <stdbool.h>
 #include <pthread.h> // for cpu_set_t
 #ifdef HAVE_AIO_SUPPORT

@@ -4,6 +4,8 @@
 #include <linux/kernel.h>
 #include <linux/err.h>
 #include <linux/zalloc.h>
+#include <linux/bitmap.h>
+#include <perf/cpumap.h>
 #include "tests.h"
 #include "util/debug.h"
 #include "util/workqueue/threadpool.h"

@@ -31,6 +33,7 @@ struct test_task {
 	struct task_struct task;
 	int n_threads;
 	int *array;
+	struct mmap_cpu_mask *affinity_masks;
 };
 
 /**

@@ -54,9 +57,36 @@ static void dummy_work(int idx)
 static void test_task_fn1(int tidx, struct task_struct *task)
 {
 	struct test_task *mtask = container_of(task, struct test_task, task);
+	struct mmap_cpu_mask real_affinity_mask, *set_affinity_mask;
+	int ret;
+
+	set_affinity_mask = &mtask->affinity_masks[tidx];
+	real_affinity_mask.nbits = set_affinity_mask->nbits;
+	real_affinity_mask.bits = bitmap_alloc(real_affinity_mask.nbits);
+	if (!real_affinity_mask.bits) {
+		pr_err("ENOMEM in malloc real_affinity_mask.bits\n");
+		goto out;
+	}
+
+	ret = pthread_getaffinity_np(pthread_self(), real_affinity_mask.nbits,
+				(cpu_set_t *)real_affinity_mask.bits);
+	if (ret) {
+		pr_err("Error in pthread_getaffinity_np: %s\n", strerror(ret));
+		goto out;
+	}
+
+	if (!bitmap_equal(real_affinity_mask.bits, set_affinity_mask->bits,
+			real_affinity_mask.nbits)) {
+		pr_err("affinity mismatch!\n");
+		mmap_cpu_mask__scnprintf(set_affinity_mask, "set affinity");
+		mmap_cpu_mask__scnprintf(&real_affinity_mask, "real affinity");
+		goto out;
+	}
 
 	dummy_work(tidx);
 	mtask->array[tidx] = tidx+1;
+out:
+	bitmap_free(real_affinity_mask.bits);
 }
 
 static void test_task_fn2(int tidx, struct task_struct *task)

@@ -116,21 +146,58 @@ static int __test__threadpool(void *_args)
 {
 	struct threadpool_test_args_t *args = _args;
 	struct threadpool *pool;
+	int ret, i, nr_cpus, nr_bits, cpu;
 	struct test_task task;
 	int pool_size = args->pool_size ?: sysconf(_SC_NPROCESSORS_ONLN);
-	int i, ret = __threadpool__prepare(&pool, pool_size);
+	struct perf_cpu_map *cpumap = perf_cpu_map__new(NULL);
+	struct mmap_cpu_mask *affinity_masks;
 
-	if (ret)
-		goto out;
+	if (!cpumap) {
+		pr_err("ENOMEM in perf_cpu_map__new\n");
+		return TEST_FAIL;
+	}
+
+	nr_cpus = perf_cpu_map__nr(cpumap);
+	nr_bits = BITS_TO_LONGS(nr_cpus) * sizeof(unsigned long);
+
+	affinity_masks = calloc(pool_size, sizeof(*affinity_masks));
+	if (!affinity_masks) {
+		pr_err("ENOMEM in calloc affinity_masks\n");
+		ret = TEST_FAIL;
+		goto out_put_cpumap;
+	}
+
+	for (i = 0; i < pool_size; i++) {
+		affinity_masks[i].nbits = nr_bits;
+		affinity_masks[i].bits = bitmap_alloc(nr_cpus);
+		if (!affinity_masks[i].bits) {
+			ret = TEST_FAIL;
+			goto out_free_affinity_masks;
+		}
+		bitmap_zero(affinity_masks[i].bits, affinity_masks[i].nbits);
+		cpu = perf_cpu_map__cpu(cpumap, i % nr_cpus);
+		test_and_set_bit(cpu, affinity_masks[i].bits);
+	}
 
 	task.task.fn = test_task_fn1;
 	task.n_threads = pool_size;
+	task.affinity_masks = affinity_masks;
 	task.array = calloc(pool_size, sizeof(*task.array));
 	TEST_ASSERT_VAL("calloc failure", task.array);
 
+	ret = __threadpool__prepare(&pool, pool_size);
+	if (ret)
+		goto out_free_tasks;
+
+	ret = threadpool__set_affinities(pool, task.affinity_masks);
+	if (ret) {
+		ret = TEST_FAIL;
+		goto out_free_tasks;
+	}
+
 	ret = __threadpool__exec_wait(pool, &task.task);
 	if (ret)
-		goto out;
+		goto out_free_tasks;
 
 	for (i = 0; i < pool_size; i++)
 		TEST_ASSERT_VAL("failed array check (1)", task.array[i] == i+1);

@@ -139,17 +206,23 @@ static int __test__threadpool(void *_args)
 
 	ret = __threadpool__exec_wait(pool, &task.task);
 	if (ret)
-		goto out;
+		goto out_free_tasks;
 
 	for (i = 0; i < pool_size; i++)
 		TEST_ASSERT_VAL("failed array check (2)", task.array[i] == 2*i);
 
 	ret = __threadpool__teardown(pool);
 	if (ret)
-		goto out;
+		goto out_free_tasks;
 
-out:
+out_free_tasks:
 	free(task.array);
+out_free_affinity_masks:
+	for (i = 0; i < pool_size; i++)
+		bitmap_free(affinity_masks[i].bits);
+	free(affinity_masks);
+out_put_cpumap:
+	perf_cpu_map__put(cpumap);
 	return ret;
 }
 

@@ -13,7 +13,9 @@
 #include <linux/err.h>
 #include <linux/kernel.h>
 #include <pthread.h>
+#include <linux/bitmap.h>
 #include <internal/lib.h>
+#include "util/mmap.h"
 #include "threadpool.h"
 
 #ifndef HAVE_GETTID

@@ -39,6 +41,7 @@ struct threadpool_entry {
 		int cmd[2];			/* messages to thread (commands) */
 	} pipes;
 	bool				running; /* has this thread been started? */
+	struct mmap_cpu_mask		affinity_mask;
 };
 
 enum threadpool_msg {

@@ -255,6 +258,16 @@ static int threadpool_entry__recv_cmd(struct threadpool_entry *thread,
 	return 0;
 }
 
+/**
+ * threadpool_entry__apply_affinity - apply @thread->affinity
+ */
+static int threadpool_entry__apply_affinity(struct threadpool_entry *thread)
+{
+	return -pthread_setaffinity_np(thread->ptid,
+			MMAP_CPU_MASK_BYTES(&thread->affinity_mask),
+			(cpu_set_t *)(thread->affinity_mask.bits));
+}
+
 /**
  * threadpool_entry__function - function running on thread
  *

@@ -339,6 +352,7 @@ struct threadpool *threadpool__new(int n_threads)
 		pool->threads[t].ptid = 0;
 		pool->threads[t].pool = pool;
 		pool->threads[t].running = false;
+		// affinity is set to zero due to calloc
 		threadpool_entry__init_pipes(&pool->threads[t]);
 	}
 

@@ -414,6 +428,7 @@ void threadpool__delete(struct threadpool *pool)
 	for (t = 0; t < pool->nr_threads; t++) {
 		thread = &pool->threads[t];
 		threadpool_entry__close_pipes(thread);
+		bitmap_free(thread->affinity_mask.bits);
 	}
 
 	zfree(&pool->threads);

@@ -455,6 +470,16 @@ int threadpool__start_thread(struct threadpool *pool, int tidx)
 	pthread_attr_init(&attrs);
 	pthread_attr_setdetachstate(&attrs, PTHREAD_CREATE_DETACHED);
 
+	if (thread->affinity_mask.bits) {
+		ret = pthread_attr_setaffinity_np(&attrs,
+				MMAP_CPU_MASK_BYTES(&thread->affinity_mask),
+				(cpu_set_t *)(thread->affinity_mask.bits));
+		if (ret) {
+			err = -ret;
+			goto out;
+		}
+	}
+
 	ret = pthread_create(&thread->ptid, &attrs, threadpool_entry__function, thread);
 	if (ret) {
 		err = -ret;

@@ -617,3 +642,48 @@ bool threadpool__is_busy(struct threadpool *pool)
 {
 	return pool->current_task;
 }
+
+/**
+ * threadpool__set_affinity - set @affinity of the @tid thread in @pool
+ *
+ * If threadpool is not running affinity will be set on start.
+ * If threadpool is running, affinity is immediately set.
+ *
+ * This function can be called from any thread.
+ */
+int threadpool__set_affinity(struct threadpool *pool, int tid,
+				struct mmap_cpu_mask *affinity)
+{
+	struct threadpool_entry *thread = &pool->threads[tid];
+	int ret = mmap_cpu_mask__duplicate(affinity, &thread->affinity_mask);
+
+	if (ret)
+		return ret;
+
+	if (thread->running)
+		return threadpool_entry__apply_affinity(thread);
+	else
+		return 0;
+}
+
+/**
+ * threadpool__set_affinities - set @affinities of all threads in @pool
+ *
+ * If threadpool is not running, affinities will be set on start.
+ * If threadpool is running, affinities are immediately set.
+ *
+ * This function can be called from any thread.
+ */
+int threadpool__set_affinities(struct threadpool *pool,
+				struct mmap_cpu_mask *affinities)
+{
+	int i, ret;
+
+	for (i = 0; i < pool->nr_threads; i++) {
+		ret = threadpool__set_affinity(pool, i, &affinities[i]);
+		if (ret)
+			return ret;
+	}
+
+	return 0;
+}

@@ -2,6 +2,8 @@
 #ifndef __WORKQUEUE_THREADPOOL_H
 #define __WORKQUEUE_THREADPOOL_H
 
+#include "util/mmap.h"
+
 struct threadpool;
 struct task_struct;
 

@@ -25,6 +27,11 @@ extern int threadpool__size(struct threadpool *pool);
 extern bool threadpool__is_running(struct threadpool *pool);
 extern bool threadpool__is_busy(struct threadpool *pool);
 
+extern int threadpool__set_affinities(struct threadpool *pool,
+				struct mmap_cpu_mask *affinities);
+extern int threadpool__set_affinity(struct threadpool *pool, int tid,
+				struct mmap_cpu_mask *affinity);
+
 /* Error management */
 #define THREADPOOL_STRERR_BUFSIZE (128+STRERR_BUFSIZE)
 #define THREADPOOL_ERROR__OFFSET 512

@@ -50,6 +50,7 @@ static const char * const workqueue_errno_str[] = {
 	"Error executing function in threadpool",
 	"Error stopping the threadpool",
 	"Error starting thread in the threadpool",
+	"Error setting affinity in threadpool",
 	"Error sending message to worker",
 	"Error receiving message from worker",
 	"Received unexpected message from worker",

@@ -758,6 +759,26 @@ int flush_workqueue(struct workqueue_struct *wq)
 	return err;
 }
 
+/**
+ * workqueue_set_affinities - set affinities to all threads in @wq->pool
+ */
+int workqueue_set_affinities(struct workqueue_struct *wq,
+				struct mmap_cpu_mask *affinities)
+{
+	wq->pool_errno = threadpool__set_affinities(wq->pool, affinities);
+	return wq->pool_errno ? -WORKQUEUE_ERROR__POOLAFFINITY : 0;
+}
+
+/**
+ * workqueue_set_affinities - set affinity to thread @tidx in @wq->pool
+ */
+int workqueue_set_affinity(struct workqueue_struct *wq, int tidx,
+				struct mmap_cpu_mask *affinity)
+{
+	wq->pool_errno = threadpool__set_affinity(wq->pool, tidx, affinity);
+	return wq->pool_errno ? -WORKQUEUE_ERROR__POOLAFFINITY : 0;
+}
+
 /**
  * init_work - initialize the @work struct
  */

@@ -28,6 +28,11 @@ extern int queue_work_on_worker(int tidx, struct workqueue_struct *wq, struct wo
 
 extern int flush_workqueue(struct workqueue_struct *wq);
 
+extern int workqueue_set_affinities(struct workqueue_struct *wq,
+				struct mmap_cpu_mask *affinities);
+extern int workqueue_set_affinity(struct workqueue_struct *wq, int tidx,
+				struct mmap_cpu_mask *affinity);
+
 extern void init_work(struct work_struct *work);
 
 /* parallel_for utility */

@@ -92,6 +97,7 @@ enum {
 	WORKQUEUE_ERROR__POOLEXE,
 	WORKQUEUE_ERROR__POOLSTOP,
 	WORKQUEUE_ERROR__POOLSTARTTHREAD,
+	WORKQUEUE_ERROR__POOLAFFINITY,
 	WORKQUEUE_ERROR__WRITEPIPE,
 	WORKQUEUE_ERROR__READPIPE,
 	WORKQUEUE_ERROR__INVALIDMSG,

@@ -10,9 +10,12 @@
 #include <linux/string.h>
 #include <linux/zalloc.h>
 #include <linux/kernel.h>
+#include <linux/bitmap.h>
 #include "debug.h"
 #include <internal/lib.h>
 #include "workqueue.h"
+#include <perf/cpumap.h>
+#include "util/affinity.h"
 
 struct workqueue_struct *global_wq;
 

@@ -43,6 +46,10 @@ struct workqueue_struct {
 	struct worker		**workers;	/* array of all workers */
 	struct worker		*next_worker;	/* next worker to choose (round robin) */
 	int			first_stopped_worker; /* next worker to start if needed */
+	struct {
+		int		*map;		/* maps cpu to thread idx */
+		int		size;		/* size of the map array */
+	} cpu_to_tidx_map;
 };
 
 static const char * const workqueue_errno_str[] = {

@@ -429,6 +436,7 @@ static void worker_thread(int tidx, struct task_struct *task)
 struct workqueue_struct *create_workqueue(int nr_threads)
 {
 	int ret, err = 0;
+	int nr_cpus = sysconf(_SC_NPROCESSORS_ONLN);
 	struct workqueue_struct *wq = zalloc(sizeof(struct workqueue_struct));
 
 	if (!wq) {

@@ -449,10 +457,18 @@ struct workqueue_struct *create_workqueue(int nr_threads)
 		goto out_delete_pool;
 	}
 
+	wq->cpu_to_tidx_map.size = nr_cpus;
+	wq->cpu_to_tidx_map.map = calloc(nr_cpus, sizeof(*wq->cpu_to_tidx_map.map));
+	if (!wq->workers) {
+		err = -ENOMEM;
+		goto out_free_workers;
+	}
+	memset(wq->cpu_to_tidx_map.map, -1, nr_cpus * sizeof(*wq->cpu_to_tidx_map.map));
+
 	ret = pthread_mutex_init(&wq->lock, NULL);
 	if (ret) {
 		err = -ret;
-		goto out_free_workers;
+		goto out_free_cpu_to_idx_map;
 	}
 
 	ret = pthread_cond_init(&wq->idle_cond, NULL);

@@ -494,6 +510,8 @@ struct workqueue_struct *create_workqueue(int nr_threads)
 	pthread_mutex_destroy(&wq->lock);
 out_free_workers:
 	free(wq->workers);
+out_free_cpu_to_idx_map:
+	free(wq->cpu_to_tidx_map.map);
 out_delete_pool:
 	threadpool__delete(wq->pool);
 out_free_wq:

@@ -552,6 +570,7 @@ int destroy_workqueue(struct workqueue_struct *wq)
 	wq->msg_pipe[1] = -1;
 
 	zfree(&wq->workers);
+	zfree(&wq->cpu_to_tidx_map.map);
 	free(wq);
 	return err;
 }

@@ -779,6 +798,118 @@ int workqueue_set_affinity(struct workqueue_struct *wq, int tidx,
 	return wq->pool_errno ? -WORKQUEUE_ERROR__POOLAFFINITY : 0;
 }
 
+/**
+ * workqueue_set_affinity_cpu - set affinity to @cpu to thread @tidx in @wq->pool
+ *
+ * If cpu is -1, then affinity is set to all online processors.
+ */
+int workqueue_set_affinity_cpu(struct workqueue_struct *wq, int tidx, int cpu)
+{
+	struct mmap_cpu_mask affinity;
+	int i, err;
+
+	if (cpu >= 0)
+		affinity.nbits = cpu+1;
+	else
+		affinity.nbits = wq->cpu_to_tidx_map.size;
+
+	affinity.bits = bitmap_alloc(affinity.nbits);
+	if (!affinity.bits) {
+		pr_debug2("Failed allocation of bitmapset\n");
+		return -ENOMEM;
+	}
+
+	if (cpu >= 0)
+		test_and_set_bit(cpu, affinity.bits);
+	else
+		bitmap_fill(affinity.bits, affinity.nbits);
+
+	err = workqueue_set_affinity(wq, tidx, &affinity);
+	if (err)
+		goto out;
+
+	// find and unset this thread from the map
+	for (i = 0; i < wq->cpu_to_tidx_map.size; i++) {
+		if (wq->cpu_to_tidx_map.map[i] == tidx)
+			wq->cpu_to_tidx_map.map[i] = -1;
+	}
+
+	if (cpu >= 0)
+		wq->cpu_to_tidx_map.map[cpu] = tidx;
+
+out:
+	bitmap_free(affinity.bits);
+	return err;
+}
+
+/**
+ * workqueue_set_affinities_cpu - set single-cpu affinities to all threads in @wq->pool
+ */
+int workqueue_set_affinities_cpu(struct workqueue_struct *wq,
+				struct perf_cpu_map *cpus)
+{
+	int cpu, idx, err;
+
+	if (perf_cpu_map__nr(cpus) > threadpool__size(wq->pool))
+		return -EINVAL;
+
+
+	perf_cpu_map__for_each_cpu(cpu, idx, cpus) {
+		err = workqueue_set_affinity_cpu(wq, idx, cpu);
+		if (err)
+			return err;
+	}
+
+	return 0;
+}
+
+struct cpu_bound_work {
+	struct work_struct work;
+	int cpu;
+	struct work_struct *original_work;
+};
+
+static void set_affinity_and_execute(struct work_struct *work)
+{
+	struct cpu_bound_work *cpu_bound_work = container_of(work, struct cpu_bound_work, work);
+	struct affinity affinity;
+
+	if (affinity__setup(&affinity) < 0)
+		goto out;
+
+	affinity__set(&affinity, cpu_bound_work->cpu);
+	cpu_bound_work->original_work->func(cpu_bound_work->original_work);
+	affinity__cleanup(&affinity);
+
+out:
+	free(cpu_bound_work);
+}
+
+/**
+ * queue_work_on - execute @work on @cpu
+ *
+ * The work is assigned to the worker pinned to @cpu, if any.
+ * Otherwise, affinity is set before running the work and unset after.
+ */
+int queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work)
+{
+	struct cpu_bound_work *cpu_bound_work;
+	int tidx = wq->cpu_to_tidx_map.map[cpu];
+
+	if (tidx >= 0)
+		return queue_work_on_worker(tidx, wq, work);
+
+	cpu_bound_work = malloc(sizeof(*cpu_bound_work));
+	if (!cpu_bound_work)
+		return -ENOMEM;
+
+	init_work(&cpu_bound_work->work);
+	cpu_bound_work->work.func = set_affinity_and_execute;
+	cpu_bound_work->cpu = cpu;
+	cpu_bound_work->original_work = work;
+	return queue_work(wq, &cpu_bound_work->work);
+}
+
 /**
  * init_work - initialize the @work struct
  */

@@ -25,6 +25,7 @@ extern int workqueue_nr_threads(struct workqueue_struct *wq);
 
 extern int queue_work(struct workqueue_struct *wq, struct work_struct *work);
 extern int queue_work_on_worker(int tidx, struct workqueue_struct *wq, struct work_struct *work);
+extern int queue_work_on(int cpu, struct workqueue_struct *wq, struct work_struct *work);
 
 extern int flush_workqueue(struct workqueue_struct *wq);
 

@@ -32,6 +33,9 @@ extern int workqueue_set_affinities(struct workqueue_struct *wq,
 				struct mmap_cpu_mask *affinities);
 extern int workqueue_set_affinity(struct workqueue_struct *wq, int tidx,
 				struct mmap_cpu_mask *affinity);
+extern int workqueue_set_affinity_cpu(struct workqueue_struct *wq, int tidx, int cpu);
+extern int workqueue_set_affinities_cpu(struct workqueue_struct *wq,
+					struct perf_cpu_map *cpus);
 
 extern void init_work(struct work_struct *work);
 

@@ -82,6 +86,14 @@ static inline int schedule_work_on_worker(int tidx, struct work_struct *work)
 	return queue_work_on_worker(tidx, global_wq, work);
 }
 
+/**
+ * schedule_work_on - queue @work to be executed on @cpu by global_wq
+ */
+static inline int schedule_work_on(int cpu, struct work_struct *work)
+{
+	return queue_work_on(cpu, global_wq, work);
+}
+
 /**
  * flush_scheduled_work - ensure that any scheduled work in global_wq has run to completion
  */

@@ -39,3 +39,5 @@ util/rwsem.c
 util/hashmap.c
 util/pmu-hybrid.c
 util/fncache.c
+util/workqueue/threadpool.c
+util/workqueue/workqueue.c

@@ -27,6 +27,8 @@
 #include "util/perf_api_probe.h"
 #include "util/evsel_fprintf.h"
 #include "util/evlist-hybrid.h"
+#include "util/util.h"
+#include "util/workqueue/workqueue.h"
 #include <signal.h>
 #include <unistd.h>
 #include <sched.h>

@@ -41,6 +43,7 @@
 #include <sys/prctl.h>
 
 #include <linux/bitops.h>
+#include <linux/kernel.h>
 #include <linux/hash.h>
 #include <linux/log2.h>
 #include <linux/err.h>

@@ -369,6 +372,120 @@ static int evlist__is_enabled(struct evlist *evlist)
 	return false;
 }
 
+struct evlist_work {
+	struct work_struct work;
+	struct evlist *evlist;
+	int cpu;
+	evsel__cpu_func func;
+	void *args;
+	int ret;
+};
+
+static void evlist__for_each_evsel_cpu_thread_func(struct work_struct *_work)
+{
+	struct evlist_work *work = container_of(_work, struct evlist_work, work);
+	int cpu_idx, ret, err = 0;
+	struct evsel *pos;
+
+	work->ret = 0;
+	evlist__for_each_entry(work->evlist, pos) {
+		cpu_idx = evsel__find_cpu(pos, work->cpu);
+		if (cpu_idx < 0)
+			continue;
+		ret = work->func(work->evlist, pos, cpu_idx, work->args);
+		if (ret) {
+			work->ret = ret;
+			if (err < 0)		// error
+				return;
+		}
+	}
+}
+
+static int evlist__for_each_evsel_cpu_multithreaded(struct evlist *evlist,
+					evsel__cpu_func func, void *args)
+{
+	int i, cpu, ret;
+	struct evlist_work *works;
+	char errbuf[WORKQUEUE_STRERR_BUFSIZE];
+
+	works = calloc(perf_cpu_map__nr(evlist->core.all_cpus), sizeof(*works));
+	perf_cpu_map__for_each_cpu(cpu, i, evlist->core.all_cpus) {
+		init_work(&works[i].work);
+		works[i].evlist = evlist;
+		works[i].work.func = evlist__for_each_evsel_cpu_thread_func;
+		works[i].cpu = cpu;
+		works[i].func = func;
+		works[i].args = args;
+		works[i].ret = 0;
+
+		ret = schedule_work_on(cpu, &works[i].work);
+		if (ret) {
+			workqueue_strerror(global_wq, ret, errbuf, sizeof(errbuf));
+			pr_debug("schedule_work: %s\n", errbuf);
+			break;
+		}
+	}
+
+	ret = flush_scheduled_work();
+	if (ret) {
+		workqueue_strerror(global_wq, ret, errbuf, sizeof(errbuf));
+		pr_debug("flush_scheduled_work: %s\n", errbuf);
+		goto out;
+	}
+
+	perf_cpu_map__for_each_cpu(cpu, i, evlist->core.all_cpus) {
+		if (works[i].ret) {
+			ret = works[i].ret;
+			if (works[i].ret < 0)		// error
+				goto out;
+		}
+	}
+out:
+	free(works);
+
+	return ret;
+}
+
+static int evlist__for_each_evsel_cpu_singlethreaded(struct evlist *evlist,
+					evsel__cpu_func func, void *args)
+{
+	int ret, err = 0, i, cpu, cpu_idx;
+	struct affinity affinity;
+	struct evsel *pos;
+
+	if (affinity__setup(&affinity) < 0)
+		return -1;
+
+	evlist__for_each_cpu(evlist, i, cpu) {
+		affinity__set(&affinity, cpu);
+
+		evlist__for_each_entry(evlist, pos) {
+			cpu_idx = evsel__find_cpu(pos, cpu);
+			if (cpu_idx < 0)
+				continue;
+			ret = func(evlist, pos, cpu_idx, args);
+			if (ret) {
+				err = ret;
+				if (err < 0)		// error
+					goto out;
+			}
+		}
+	}
+
+out:
+	affinity__cleanup(&affinity);
+	return err;
+}
+
+int evlist__for_each_evsel_cpu(struct evlist *evlist, evsel__cpu_func func, void *args)
+{
+	if (perf_singlethreaded)
+		return evlist__for_each_evsel_cpu_singlethreaded(evlist, func, args);
+	else
+		return evlist__for_each_evsel_cpu_multithreaded(evlist, func, args);
+
+}
+
 static void __evlist__disable(struct evlist *evlist, char *evsel_name)
 {
 	struct evsel *pos;