From 9d747addfefb39bb2b7ea6f29fac647bc8e9d234 Mon Sep 17 00:00:00 2001
From: Hitoshi Mitake <mitake.hitoshi@lab.ntt.co.jp>
Date: Fri, 15 Jul 2016 14:06:48 +0900
Subject: [PATCH] sheep: use fixed workqueues for stable performance

Example of usage:
$ sheep -w gway=100,net=300
---
 sheep/sheep.c | 64 +++++++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 59 insertions(+), 5 deletions(-)

diff --git a/sheep/sheep.c b/sheep/sheep.c
index ad6f39cc5..d1400c30c 100644
--- a/sheep/sheep.c
+++ b/sheep/sheep.c
@@ -153,6 +153,7 @@ static struct sd_option sheep_options[] = {
 	{'u', "upgrade", false, "upgrade to the latest data layout"},
 	{'v', "version", false, "show the version"},
 	{'V', "vnodes", true, "set number of vnodes", vnodes_help},
+	{'w', "wq-threads", true, "specify a number of threads for workqueue"},
 	{'W', "wildcard-recovery", false, "wildcard recovery for first time"},
 	{'y', "myaddr", true, "specify the address advertised to other sheep",
 	 myaddr_help},
@@ -345,6 +346,50 @@ static struct option_parser log_parsers[] = {
 	{ NULL, NULL },
 };
 
+#define DEFAULT_FIXED_WQ_THREADS 1024
+
+static int wq_net_threads = DEFAULT_FIXED_WQ_THREADS;
+static int wq_net_parser(const char *s)
+{
+	wq_net_threads = atoi(s);
+	return 0;
+}
+
+static int wq_gway_threads = DEFAULT_FIXED_WQ_THREADS;
+static int wq_gway_parser(const char *s)
+{
+	wq_gway_threads = atoi(s);
+	return 0;
+}
+
+static int wq_io_threads = DEFAULT_FIXED_WQ_THREADS;
+static int wq_io_parser(const char *s)
+{
+	wq_io_threads = atoi(s);
+	return 0;
+}
+
+static int wq_recovery_threads = DEFAULT_FIXED_WQ_THREADS;
+static int wq_recovery_parser(const char *s)
+{
+	wq_recovery_threads = atoi(s);
+	return 0;
+}
+
+static int wq_async_threads = DEFAULT_FIXED_WQ_THREADS;
+static int wq_async_parser(const char *s)
+{
+	wq_async_threads = atoi(s);
+	return 0;
+}
+
+static struct option_parser wq_parsers[] = {
+	{ "net=", wq_net_parser },
+	{ "gway=", wq_gway_parser },
+	{ "io=", wq_io_parser },
+	{ "recovery=", wq_recovery_parser },
+	{ "async=", wq_async_parser },
+};
 
 static const char *io_addr, *io_pt;
 static int ionic_host_parser(const char *s)
@@ -453,14 +498,19 @@ static int create_work_queues(void)
 	if (init_work_queue(get_nr_nodes))
 		return -1;
 
-	sys->net_wqueue = create_work_queue("net", WQ_UNLIMITED);
-	sys->gateway_wqueue = create_work_queue("gway", WQ_UNLIMITED);
-	sys->io_wqueue = create_work_queue("io", WQ_UNLIMITED);
-	sys->recovery_wqueue = create_work_queue("rw", WQ_UNLIMITED);
+	sd_info("# of threads in net workqueue: %d", wq_net_threads);
+	sys->net_wqueue = create_fixed_work_queue("net", wq_net_threads);
+	sd_info("# of threads in gway workqueue: %d", wq_net_threads);
+	sys->gateway_wqueue = create_fixed_work_queue("gway", wq_gway_threads);
+	sd_info("# of threads in io workqueue: %d", wq_net_threads);
+	sys->io_wqueue = create_fixed_work_queue("io", wq_io_threads);
+	sd_info("# of threads in rw workqueue: %d", wq_net_threads);
+	sys->recovery_wqueue = create_fixed_work_queue("rw", wq_io_threads);
 	sys->deletion_wqueue = create_ordered_work_queue("deletion");
 	sys->block_wqueue = create_ordered_work_queue("block");
 	sys->md_wqueue = create_ordered_work_queue("md");
-	sys->areq_wqueue = create_work_queue("async_req", WQ_UNLIMITED);
+	sd_info("# of threads in async workqueue: %d", wq_net_threads);
+	sys->areq_wqueue = create_fixed_work_queue("async_req", wq_async_threads);
 	if (!sys->gateway_wqueue || !sys->io_wqueue || !sys->recovery_wqueue ||
 	    !sys->deletion_wqueue || !sys->block_wqueue || !sys->md_wqueue ||
 	    !sys->areq_wqueue)
@@ -805,6 +855,10 @@ int main(int argc, char **argv)
 		case 'W':
 			wildcard_recovery = true;
 			break;
+		case 'w':
+			if (option_parse(optarg, ",", wq_parsers) < 0)
+				exit(1);
+			break;
 		default:
 			usage(1);
 			break;