Tune queue limits, some other adjustments

4 files changed, 174 insertions, 29 deletions

diff --git a/src/defaultsettings.cpp b/src/defaultsettings.cpp
index 219cda9e7..0e82c4e2f 100644
--- a/src/defaultsettings.cpp
+++ b/src/defaultsettings.cpp
@@ -185,8 +185,8 @@ void set_default_settings(Settings *settings)
 	settings->setDefault("remote_media", "");
 	settings->setDefault("debug_log_level", "0");
 	settings->setDefault("emergequeue_limit_total", "256");
-	settings->setDefault("emergequeue_limit_diskonly", "5");
-	settings->setDefault("emergequeue_limit_generate", "1");
+	settings->setDefault("emergequeue_limit_diskonly", "");
+	settings->setDefault("emergequeue_limit_generate", "");
 	settings->setDefault("num_emerge_threads", "");
 	
 	// physics stuff
diff --git a/src/emerge.cpp b/src/emerge.cpp
index ac654f368..dd97734c5 100644
--- a/src/emerge.cpp
+++ b/src/emerge.cpp
@@ -49,18 +49,29 @@ EmergeManager::EmergeManager(IGameDef *gamedef, BiomeDefManager *bdef) {
 	this->params   = NULL;
 	
 	mapgen_debug_info = g_settings->getBool("enable_mapgen_debug_info");
-	
-	qlimit_total    = g_settings->getU16("emergequeue_limit_total");
-	qlimit_diskonly = g_settings->getU16("emergequeue_limit_diskonly");
-	qlimit_generate = g_settings->getU16("emergequeue_limit_generate");
 
 	queuemutex.Init();
-	int nthreads = g_settings->get("num_emerge_threads").empty() ?
-					porting::getNumberOfProcessors() :
-					g_settings->getU16("num_emerge_threads");
+	
+	int nthreads;
+	if (g_settings->get("num_emerge_threads").empty()) {
+		int nprocs = porting::getNumberOfProcessors();
+		// leave a proc for the main thread and one for some other misc threads
+		if (nprocs > 2)
+			nthreads = nprocs - 2;
+	} else {
+		nthreads = g_settings->getU16("num_emerge_threads");
+	}
 	if (nthreads < 1)
 		nthreads = 1;
 	
+	qlimit_total    = g_settings->getU16("emergequeue_limit_total");
+	qlimit_diskonly = g_settings->get("emergequeue_limit_diskonly").empty() ?
+		nthreads * 5 + 1 :
+		g_settings->getU16("emergequeue_limit_diskonly");
+	qlimit_generate = g_settings->get("emergequeue_limit_generate").empty() ?
+		nthreads + 1 :
+		g_settings->getU16("emergequeue_limit_generate");
+	
 	for (int i = 0; i != nthreads; i++)
 		emergethread.push_back(new EmergeThread((Server *)gamedef, i));
 		
@@ -138,6 +149,7 @@ bool EmergeManager::enqueueBlockEmerge(u16 peer_id, v3s16 p, bool allow_generate
 		
 		peer_queue_count[peer_id] = count + 1;
 		
+		// insert into the EmergeThread queue with the least items
 		int lowestitems = emergethread[0]->blockqueue.size();
 		for (int i = 1; i != emergethread.size(); i++) {
 			int nitems = emergethread[i]->blockqueue.size();
@@ -183,7 +195,7 @@ bool EmergeThread::popBlockEmerge(v3s16 *pos, u8 *flags) {
 
 
 int EmergeManager::getGroundLevelAtPoint(v2s16 p) {
-	if (!mapgen[0]) {
+	if (mapgen.size() == 0 || !mapgen[0]) {
 		errorstream << "EmergeManager: getGroundLevelAtPoint() called"
 		" before mapgen initialized" << std::endl;
 		return 0;
@@ -365,7 +377,7 @@ bool EmergeThread::getBlockOrStartGen(v3s16 p, MapBlock **b,
 
 void *EmergeThread::Thread() {
 	ThreadStarted();
-	log_register_thread("EmergeThread");
+	log_register_thread("EmergeThread" + id);
 	DSTACK(__FUNCTION_NAME);
 	BEGIN_DEBUG_EXCEPTION_HANDLER
 
diff --git a/src/porting.cpp b/src/porting.cpp
index 58d71e4aa..84df15b30 100644
--- a/src/porting.cpp
+++ b/src/porting.cpp
@@ -131,29 +131,127 @@ void signal_handler_init(void)
 
 #endif
 
+
 /*
 	Multithreading support
 */
 int getNumberOfProcessors() {
-	#if defined(_SC_NPROCESSORS_ONLN)
-		return sysconf(_SC_NPROCESSORS_ONLN);
-	#elif defined(__FreeBSD__) || defined(__APPLE__)
-		unsigned int len, count;
-		len = sizeof(count);
-		return sysctlbyname("hw.ncpu", &count, &len, NULL, 0);
-	#elif defined(_GNU_SOURCE)
-		return get_nprocs();
-	#elif defined(_WIN32)
-		SYSTEM_INFO sysinfo;
-		GetSystemInfo(&sysinfo);
-		return sysinfo.dwNumberOfProcessors;
-	#elif defined(PTW32_VERSION) || defined(__hpux)
-		return pthread_num_processors_np();
-	#else
-		return 1;
-	#endif
+#if defined(_SC_NPROCESSORS_ONLN)
+
+	return sysconf(_SC_NPROCESSORS_ONLN);
+
+#elif defined(__FreeBSD__) || defined(__APPLE__)
+
+	unsigned int len, count;
+	len = sizeof(count);
+	return sysctlbyname("hw.ncpu", &count, &len, NULL, 0);
+
+#elif defined(_GNU_SOURCE)
+
+	return get_nprocs();
+
+#elif defined(_WIN32)
+
+	SYSTEM_INFO sysinfo;
+	GetSystemInfo(&sysinfo);
+	return sysinfo.dwNumberOfProcessors;
+
+#elif defined(PTW32_VERSION) || defined(__hpux)
+
+	return pthread_num_processors_np();
+
+#else
+
+	return 1;
+
+#endif
 }
 
+
+bool threadBindToProcessor(threadid_t tid, int pnumber) {
+#if defined(_WIN32)
+
+	HANDLE hThread = OpenThread(THREAD_ALL_ACCESS, 0, tid);
+	if (!hThread)
+		return false;
+
+	bool success = SetThreadAffinityMask(hThread, 1 << pnumber) != 0;
+
+	CloseHandle(hThread);
+	return success;
+
+#elif (defined(__FreeBSD__) && (__FreeBSD_version >= 702106)) \
+	|| defined(__linux) || defined(linux)
+
+	cpu_set_t cpuset;
+
+	CPU_ZERO(&cpuset);
+	CPU_SET(pnumber, &cpuset);
+	return pthread_setaffinity_np(tid, sizeof(cpuset), &cpuset) == 0;
+
+#elif defined(__sun) || defined(sun)
+
+	return processor_bind(P_LWPID, MAKE_LWPID_PTHREAD(tid), 
+						pnumber, NULL) == 0;
+
+#elif defined(_AIX)
+	
+	return bindprocessor(BINDTHREAD, (tid_t)tid, pnumber) == 0;
+
+#elif defined(__hpux) || defined(hpux)
+
+	pthread_spu_t answer;
+
+	return pthread_processor_bind_np(PTHREAD_BIND_ADVISORY_NP,
+									&answer, pnumber, tid) == 0;
+	
+#elif defined(__APPLE__)
+
+	struct thread_affinity_policy tapol;
+	
+	thread_port_t threadport = pthread_mach_thread_np(tid);
+	tapol.affinity_tag = pnumber + 1;
+	return thread_policy_set(threadport, THREAD_AFFINITY_POLICY,
+			(thread_policy_t)&tapol, THREAD_AFFINITY_POLICY_COUNT) == KERN_SUCCESS;
+
+#else
+
+	return false;
+
+#endif
+}
+
+
+bool threadSetPriority(threadid_t tid, int prio) {
+#if defined(_WIN32)
+
+	HANDLE hThread = OpenThread(THREAD_ALL_ACCESS, 0, tid);
+	if (!hThread)
+		return false;
+
+	bool success = SetThreadPriority(hThread, prio) != 0;
+
+	CloseHandle(hThread);
+	return success;
+	
+#else
+
+	struct sched_param sparam;
+	int policy;
+	
+	if (pthread_getschedparam(tid, &policy, &sparam) != 0)
+		return false;
+		
+	int min = sched_get_priority_min(policy);
+	int max = sched_get_priority_max(policy);
+
+	sparam.sched_priority = min + prio * (max - min) / THREAD_PRIORITY_HIGHEST;
+	return pthread_setschedparam(tid, policy, &sparam) == 0;
+	
+#endif
+}
+
+
 /*
 	Path mangler
 */
diff --git a/src/porting.h b/src/porting.h
index 53aad6171..74ee97f88 100644
--- a/src/porting.h
+++ b/src/porting.h
@@ -46,8 +46,33 @@ with this program; if not, write to the Free Software Foundation, Inc.,
 #else
 	#include <unistd.h>
 	#include <stdint.h> //for uintptr_t
-	
+
+	#if defined(linux) || defined(__linux)
+		#define _GNU_SOURCE
+	#endif
+
+	#include <sched.h>
+
+	#ifdef __FreeBSD__
+		#include <pthread_np.h>
+		typedef cpuset_t cpu_set_t;
+	#elif defined(__sun) || defined(sun)
+		#include <sys/types.h>
+		#include <sys/processor.h>
+	#elif defined(_AIX)
+		#include <sys/processor.h>
+	#elif __APPLE__
+		#include <mach/mach_init.h>
+		#include <mach/thread_policy.h>
+	#endif
+
 	#define sleep_ms(x) usleep(x*1000)
+	
+	#define THREAD_PRIORITY_LOWEST       0
+	#define THREAD_PRIORITY_BELOW_NORMAL 1
+	#define THREAD_PRIORITY_NORMAL       2
+	#define THREAD_PRIORITY_ABOVE_NORMAL 3
+	#define THREAD_PRIORITY_HIGHEST	     4
 #endif
 
 #ifdef _MSC_VER
@@ -109,6 +134,16 @@ void initializePaths();
 int getNumberOfProcessors();
 
 /*
+	Set a thread's affinity to a particular processor.
+*/
+bool threadBindToProcessor(threadid_t tid, int pnumber);
+
+/*
+	Set a thread's priority.
+*/
+bool threadSetPriority(threadid_t tid, int prio);
+
+/*
 	Resolution is 10-20ms.
 	Remember to check for overflows.
 	Overflow can occur at any value higher than 10000000.