Set the scheduling policy individually per thread

include/crucible/task.h

@@ -25,7 +25,7 @@ namespace crucible {
 		Task() = default;
 
 		/// Create Task object containing closure and description.
-		Task(string title, function<void()> exec_fn);
+		Task(string title, int policy, function<void()> exec_fn);
 
 		/// Schedule Task for at most one future execution.
 		/// May run Task in current thread or in other thread.

lib/task.cc

@@ -47,6 +47,7 @@ namespace crucible {
 	};
 
 	class TaskState : public enable_shared_from_this<TaskState> {
+		const int                               m_sched_policy;
 		mutex					m_mutex;
 		const function<void()> 			m_exec_fn;
 		const string				m_title;
@@ -93,7 +94,7 @@ namespace crucible {
 
 	public:
 		~TaskState();
-		TaskState(string title, function<void()> exec_fn);
+		TaskState(string title, int policy, function<void()> exec_fn);
 
 		/// Run the task at most one more time.  If task has
 		/// already started running, a new instance is scheduled.
@@ -206,7 +207,7 @@ namespace crucible {
 			} else {
 				// If there are multiple tasks, create a new task to wrap our post-exec queue,
 				// then push it to the front of the global queue using normal locking methods.
-				TaskStatePtr rescue_task(make_shared<TaskState>("rescue_task", [](){}));
+				TaskStatePtr rescue_task(make_shared<TaskState>("rescue_task", SCHED_IDLE, [](){})); //TODO(kakra) Check prio
 				swap(rescue_task->m_post_exec_queue, queue);
 				TaskQueue tq_one { rescue_task };
 				TaskMasterState::push_front(tq_one);
@@ -220,7 +221,8 @@ namespace crucible {
 		--s_instance_count;
 	}
 
-	TaskState::TaskState(string title, function<void()> exec_fn) :
+	TaskState::TaskState(string title, int policy, function<void()> exec_fn) :
+		m_sched_policy(policy),
 		m_exec_fn(exec_fn),
 		m_title(title),
 		m_id(++s_next_id)
@@ -302,10 +304,17 @@ namespace crucible {
 		DIE_IF_MINUS_ERRNO(pthread_getname_np(pthread_self(), buf, sizeof(buf)));
 		DIE_IF_MINUS_ERRNO(pthread_setname_np(pthread_self(), m_title.c_str()));
 
+		int policy = SCHED_OTHER;
+		sched_param param = { .sched_priority = 0 };
+		DIE_IF_MINUS_ERRNO(pthread_getschedparam(pthread_self(), &policy, &param));
+
 		catch_all([&]() {
+			sched_param param = { .sched_priority = 0 };
+			pthread_setschedparam(pthread_self(), m_sched_policy, &param);
 			m_exec_fn();
 		});
 
+		pthread_setschedparam(pthread_self(), policy, &param);
 		pthread_setname_np(pthread_self(), buf);
 
 		lock.lock();
@@ -585,6 +594,8 @@ namespace crucible {
 	void
 	TaskMasterState::loadavg_thread_fn()
 	{
+		sched_param param = { .sched_priority = 0 };
+		pthread_setschedparam(pthread_self(), SCHED_ISO, &param);
 		pthread_setname_np(pthread_self(), "load_tracker");
 		while (!m_cancelled) {
 			adjust_thread_count();
@@ -627,8 +638,8 @@ namespace crucible {
 	{
 	}
 
-	Task::Task(string title, function<void()> exec_fn) :
-		m_task_state(make_shared<TaskState>(title, exec_fn))
+	Task::Task(string title, int policy, function<void()> exec_fn) :
+		m_task_state(make_shared<TaskState>(title, policy, exec_fn))
 	{
 	}
 
@@ -867,7 +878,7 @@ namespace crucible {
 	}
 
 	ExclusionState::ExclusionState(const string &title) :
-		m_task(title, [](){})
+		m_task(title, SCHED_IDLE, [](){}) //TODO(kakra) Check prio
 	{
 	}
 

scripts/beesd@.service.in

@@ -7,7 +7,6 @@ After=sysinit.target
 Type=simple
 ExecStart=@PREFIX@/sbin/beesd --no-timestamps %i
 CPUAccounting=true
-CPUSchedulingPolicy=batch
 CPUWeight=12
 IOSchedulingClass=idle
 IOSchedulingPriority=7

src/bees-context.cc

@@ -327,7 +327,7 @@ BeesContext::scan_one_extent(const BeesFileRange &bfr, const Extent &e)
 		// Prealloc is all zero and we replace it with a hole.
 		// No special handling is required here.  Nuke it and move on.
 		Task(
-			"dedup_prealloc",
+			"dedup_prealloc", SCHED_IDLE,
 			[m_ctx, bfr, e]() {
 				BEESLOGINFO("prealloc extent " << e);
 				// Must not extend past EOF
@@ -980,10 +980,10 @@ BeesContext::start()
 
 	m_progress_thread = make_shared<BeesThread>("progress_report");
 	m_status_thread = make_shared<BeesThread>("status_report");
-	m_progress_thread->exec([=]() {
+	m_progress_thread->exec(SCHED_OTHER, [=]() {
 		show_progress();
 	});
-	m_status_thread->exec([=]() {
+	m_status_thread->exec(SCHED_OTHER, [=]() {
 		dump_status();
 	});
 

src/bees-hash.cc

@@ -760,13 +760,13 @@ BeesHashTable::BeesHashTable(shared_ptr<BeesContext> ctx, string filename, off_t
 
 	m_extent_metadata.resize(m_extents);
 
-	m_writeback_thread.exec([&]() {
+	m_writeback_thread.exec(SCHED_IDLE, [&]() {
 		writeback_loop();
-        });
+	});
 
-	m_prefetch_thread.exec([&]() {
+	m_prefetch_thread.exec(SCHED_IDLE, [&]() {
 		prefetch_loop();
-        });
+	});
 
 	// Blacklist might fail if the hash table is not stored on a btrfs
 	catch_all([&]() {

src/bees-roots.cc

@@ -257,7 +257,7 @@ BeesRoots::crawl_batch(shared_ptr<BeesCrawl> this_crawl)
 		auto this_hold = this_crawl->hold_state(this_range);
 		auto shared_this_copy = shared_from_this();
 		BEESNOTE("Starting task " << this_range);
-		Task(task_title, [ctx_copy, this_hold, this_range, shared_this_copy]() {
+		Task(task_title, SCHED_BATCH, [ctx_copy, this_hold, this_range, shared_this_copy]() {
 			BEESNOTE("scan_forward " << this_range);
 			ctx_copy->scan_forward(this_range);
 			shared_this_copy->crawl_state_set_dirty();
@@ -382,7 +382,7 @@ BeesRoots::crawl_thread()
 
 	// Create the Task that does the crawling
 	auto shared_this = shared_from_this();
-	m_crawl_task = Task("crawl_master", [shared_this]() {
+	m_crawl_task = Task("crawl_master", SCHED_IDLE, [shared_this]() {
 		auto tqs = TaskMaster::get_queue_count();
 		BEESNOTE("queueing extents to scan, " << tqs << " of " << BEES_MAX_QUEUE_SIZE);
 		bool run_again = true;
@@ -572,7 +572,7 @@ BeesRoots::BeesRoots(shared_ptr<BeesContext> ctx) :
 void
 BeesRoots::start()
 {
-	m_crawl_thread.exec([&]() {
+	m_crawl_thread.exec(SCHED_IDLE, [&]() {
 		// Measure current transid before creating any crawlers
 		catch_all([&]() {
 			m_transid_re.update(transid_max_nocache());
@@ -583,7 +583,7 @@ BeesRoots::start()
 			state_load();
 		});
 
-		m_writeback_thread.exec([&]() {
+		m_writeback_thread.exec(SCHED_IDLE, [&]() {
 			writeback_thread();
 		});
 		crawl_thread();

src/bees-thread.cc

@@ -10,28 +10,30 @@ BeesThread::BeesThread(string name) :
 }
 
 void
-BeesThread::exec(function<void()> func)
+BeesThread::exec(int policy, function<void()> func)
 {
 	m_timer.reset();
 	BEESLOGDEBUG("BeesThread exec " << m_name);
 	m_thread_ptr = make_shared<thread>([=]() {
 		BeesNote::set_name(m_name);
-		BEESLOGDEBUG("Starting thread " << m_name);
+		BEESLOGDEBUG("Starting thread " << m_name << " policy " << policy);
 		BEESNOTE("thread function");
 		Timer thread_time;
 		catch_all([&]() {
+			sched_param param = { .sched_priority = 0 };
+			pthread_setschedparam(pthread_self(), policy, &param);
 			func();
 		});
 		BEESLOGDEBUG("Exiting thread " << m_name << ", " << thread_time << " sec");
 	});
 }
 
-BeesThread::BeesThread(string name, function<void()> func) :
+BeesThread::BeesThread(string name, int policy, function<void()> func) :
 	m_name(name)
 {
 	THROW_CHECK1(invalid_argument, name, !name.empty());
 	BEESLOGDEBUG("BeesThread construct " << m_name);
-	exec(func);
+	exec(policy, func);
 }
 
 void

src/bees-trace.cc

@@ -101,6 +101,7 @@ BeesNote::BeesNote(function<void(ostream &os)> f) :
 	m_func(f)
 {
 	m_name = get_name();
+	m_policy = get_policy();
 	m_prev = tl_next;
 	tl_next = this;
 	unique_lock<mutex> lock(s_mutex);
@@ -149,6 +150,16 @@ BeesNote::get_name()
 	return buf;
 }
 
+int
+BeesNote::get_policy()
+{
+	int policy = SCHED_OTHER;
+	sched_param param = { .sched_priority = 0 };
+	pthread_getschedparam(pthread_self(), &policy, &param);
+
+	return policy;
+}
+
 BeesNote::ThreadStatusMap
 BeesNote::get_status()
 {
@@ -162,6 +173,14 @@ BeesNote::get_status()
 		if (t.second->m_timer.age() > BEES_TOO_LONG) {
 			oss << "[" << t.second->m_timer << "s] ";
 		}
+		switch (t.second->m_policy) {
+			case SCHED_BATCH:
+				oss << "[BT] ";
+				break;
+			case SCHED_IDLE:
+				oss << "[ID] ";
+				break;
+		}
 		t.second->m_func(oss);
 		rv[t.first] = oss.str();
 	}

src/bees.h

@@ -206,6 +206,7 @@ class BeesTracer {
 class BeesNote {
 	function<void(ostream &)>	m_func;
 	BeesNote			*m_prev;
+	int					m_policy;
 	Timer				m_timer;
 	string				m_name;
 
@@ -225,6 +226,7 @@ class BeesNote {
 
 	static void set_name(const string &name);
 	static string get_name();
+	static int get_policy();
 };
 
 // C++ threads dumbed down even further
@@ -236,8 +238,8 @@ class BeesThread {
 public:
 	~BeesThread();
 	BeesThread(string name);
-	BeesThread(string name, function<void()> args);
-	void exec(function<void()> args);
+	BeesThread(string name, int policy, function<void()> args);
+	void exec(int policy, function<void()> args);
 	void join();
 	void set_name(const string &name);
 };

test/task.cc

@@ -33,7 +33,7 @@ test_tasks(size_t count)
 		ostringstream oss;
 		oss << "task #" << c;
 		Task t(
-			oss.str(),
+			oss.str(), SCHED_OTHER,
 			[c, &task_done, &mtx, &cv]() {
 				unique_lock<mutex> lock(mtx);
 				// cerr << "Task #" << c << endl;
@@ -101,7 +101,7 @@ test_barrier(size_t count)
 		ostringstream oss;
 		oss << "task #" << c;
 		Task t(
-			oss.str(),
+			oss.str(), SCHED_OTHER,
 			[c, &task_done, &mtx, bl]() mutable {
 				// cerr << "Task #" << c << endl;
 				unique_lock<mutex> lock(mtx);
@@ -120,7 +120,7 @@ test_barrier(size_t count)
 	bool done_flag = false;
 
 	Task completed(
-		"Waiting for Barrier",
+		"Waiting for Barrier", SCHED_OTHER,
 		[&mtx, &cv, &done_flag]() {
 			unique_lock<mutex> lock(mtx);
 			// cerr << "Running cv notify" << endl;
@@ -170,7 +170,7 @@ test_exclusion(size_t count)
 		ostringstream oss;
 		oss << "task #" << c;
 		Task t(
-			oss.str(),
+			oss.str(), SCHED_OTHER,
 			[c, &only_one, excl, &lock_success_count, &lock_failure_count, &pings, &tasks_running, &cv, &mtx]() mutable {
 				// cerr << "Task #" << c << endl;
 				(void)c;