sheep: add a new option for setting threads for dynamic workqueues

This commit adds a new option -x to sheep for configuring the maximum
number of threads for dynamic workqueues.

Example:

  # Set the max number of threads for dynamic workqueues to 500
  $ sheep -x 500

If this is not passed, sheep uses a default formula for configuring
the maximum number (that is, max(#nodes,#cores,16)*2).

This is for tuning performance, so please be careful to use. Sheep
works very slowly if you give a too small number, or may be shot by
OOM-killer under heavy load if you give a huge number.

Signed-off-by: Takashi Menjo <menjo.takashi@lab.ntt.co.jp>
(cherry picked from commit dd924bd)

Conflicts:
	lib/work.c

include/work.h

@@ -65,6 +65,7 @@ struct work_queue *create_fixed_work_queue(const char *name, int nr_threads);
 void queue_work(struct work_queue *q, struct work *work);
 bool work_queue_empty(struct work_queue *q);
 int wq_trace_init(void);
+void set_max_dynamic_threads(size_t nr_max);
 
 #ifdef HAVE_TRACE
 void suspend_worker_threads(void);

lib/work.c

@@ -33,6 +33,9 @@
 #include "work.h"
 #include "event.h"
 
+/* If this is greater than 0, the number of threads grows based on it. */
+static size_t max_dynamic_threads = 0;
+
 /*
  * The protection period from shrinking work queue.  This is necessary
  * to avoid many calls of pthread_create.  Without it, threads are
@@ -203,9 +206,13 @@ static inline uint64_t wq_get_roof(struct wq_info *wi)
 	case WQ_ORDERED:
 		break;
 	case WQ_DYNAMIC:
-		/* max(#nodes,#cores,16)*2 threads */
-		nr = (uint64_t)max(nr_nodes, nr_cores);
-		nr = max(nr, UINT64_C(16)) * 2;
+		if (max_dynamic_threads > 0) {
+			nr = (uint64_t)max_dynamic_threads;
+		} else {
+			/* max(#nodes,#cores,16)*2 threads */
+			nr = (uint64_t)max(nr_nodes, nr_cores);
+			nr = max(nr, UINT64_C(16)) * 2;
+		}
 		break;
 	case WQ_FIXED:
 		nr = wi->nr_threads;
@@ -449,6 +456,11 @@ bool work_queue_empty(struct work_queue *q)
 	return uatomic_read(&wi->nr_queued_work) == 0;
 }
 
+void set_max_dynamic_threads(size_t nr_max)
+{
+	max_dynamic_threads = nr_max;
+}
+
 struct thread_args {
 	const char *name;
 	void *(*start_routine)(void *);

sheep/sheep.c

@@ -154,6 +154,8 @@ static struct sd_option sheep_options[] = {
 	{'w', "cache", true, "enable object cache", cache_help},
 	{'q', "wq-threads", true, "specify a number of threads for workqueue"},
 	{'W', "wildcard-recovery", false, "wildcard recovery for first time"},
+	{'x', "max-dynamic-threads", true,
+	 "specify the maximum number of threads for dynamic workqueue"},
 	{'y', "myaddr", true, "specify the address advertised to other sheep",
 	 myaddr_help},
 	{'z', "zone", true,
@@ -730,6 +732,7 @@ int main(int argc, char **argv)
 	     *argp = NULL;
 	bool explicit_addr = false;
 	int64_t zone = -1;
+	uint32_t max_dynamic_threads = 0;
 	struct cluster_driver *cdrv;
 	struct option *long_options;
 #ifdef HAVE_HTTP
@@ -884,6 +887,16 @@ int main(int argc, char **argv)
 			if (option_parse(optarg, ",", wq_parsers) < 0)
 				exit(1);
 			break;
+		case 'x':
+			max_dynamic_threads = str_to_u32(optarg);
+			if (errno != 0 || max_dynamic_threads < 1) {
+				sd_err("Invalid number of threads '%s': "
+				       "must be an integer between 1 and %"PRIu32,
+				       optarg, UINT32_MAX);
+				exit(1);
+			}
+			set_max_dynamic_threads((size_t)max_dynamic_threads);
+			break;
 		default:
 			usage(1);
 			break;